#!/usr/bin/env vpython3
# Copyright 2012 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Unit tests for subprocess2.py."""

import os
import sys
import unittest
from unittest import mock

DEPOT_TOOLS = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, DEPOT_TOOLS)

import subprocess  # noqa: E402
import subprocess2  # noqa: E402

TEST_FILENAME = "subprocess2_test_script.py"

TEST_COMMAND = [
    sys.executable,
    os.path.join(DEPOT_TOOLS, "testing_support", TEST_FILENAME),
]


class DefaultsTest(unittest.TestCase):
    @mock.patch("subprocess2.communicate")
    def test_check_call_defaults(self, mockCommunicate):
        mockCommunicate.return_value = (("stdout", "stderr"), 0)
        self.assertEqual(
            ("stdout", "stderr"), subprocess2.check_call_out(["foo"], a=True)
        )
        mockCommunicate.assert_called_with(["foo"], a=True)

    @mock.patch("subprocess2.communicate")
    def test_capture_defaults(self, mockCommunicate):
        mockCommunicate.return_value = (("stdout", "stderr"), 0)
        self.assertEqual("stdout", subprocess2.capture(["foo"], a=True))
        mockCommunicate.assert_called_with(
            ["foo"], a=True, stdin=subprocess2.DEVNULL, stdout=subprocess2.PIPE
        )

    @mock.patch("subprocess2.Popen")
    def test_communicate_defaults(self, mockPopen):
        mockPopen().communicate.return_value = ("bar", "baz")
        mockPopen().returncode = -8
        self.assertEqual(
            (("bar", "baz"), -8), subprocess2.communicate(["foo"], a=True)
        )
        mockPopen.assert_called_with(["foo"], a=True)

    @mock.patch("os.environ", {})
    @mock.patch("subprocess.Popen.__init__")
    def test_Popen_defaults(self, mockPopen):
        with mock.patch("sys.platform", "win32"):
            subprocess2.Popen(["foo"], a=True)
            mockPopen.assert_called_with(["foo"], a=True, shell=True)

        with mock.patch("sys.platform", "non-win32"):
            subprocess2.Popen(["foo"], a=True)
            mockPopen.assert_called_with(["foo"], a=True, shell=False)

    def test_get_english_env(self):
        with mock.patch("sys.platform", "win32"):
            self.assertIsNone(subprocess2.get_english_env({}))

        with mock.patch("sys.platform", "non-win32"):
            self.assertIsNone(subprocess2.get_english_env({}))
            self.assertIsNone(
                subprocess2.get_english_env(
                    {"LANG": "en_XX", "LANGUAGE": "en_YY"}
                )
            )
            self.assertEqual(
                {"LANG": "en_US.UTF-8", "LANGUAGE": "en_US.UTF-8"},
                subprocess2.get_english_env({"LANG": "bar", "LANGUAGE": "baz"}),
            )

    @mock.patch("subprocess2.communicate")
    def test_check_output_defaults(self, mockCommunicate):
        mockCommunicate.return_value = (("stdout", "stderr"), 0)
        self.assertEqual("stdout", subprocess2.check_output(["foo"], a=True))
        mockCommunicate.assert_called_with(
            ["foo"], a=True, stdin=subprocess2.DEVNULL, stdout=subprocess2.PIPE
        )

    @mock.patch("subprocess.Popen.__init__")
    def test_env_type(self, mockPopen):
        subprocess2.Popen(["foo"], env={b"key": b"value"})
        mockPopen.assert_called_with(
            ["foo"], env={"key": "value"}, shell=mock.ANY
        )


def _run_test(with_subprocess=True):
    """Runs a tests in 12 combinations:
    - With universal_newlines=True and False.
    - With LF, CR, and CRLF output.
    - With subprocess and subprocess2.
    """
    subps = (subprocess2, subprocess) if with_subprocess else (subprocess2,)
    no_op = lambda s: s  # noqa: E731
    to_bytes = lambda s: s.encode()  # noqa: E731
    to_cr_bytes = lambda s: s.replace("\n", "\r").encode()  # noqa: E731
    to_crlf_bytes = lambda s: s.replace("\n", "\r\n").encode()  # noqa: E731

    def wrapper(test):
        def inner(self):
            for subp in subps:
                # universal_newlines = False
                test(self, to_bytes, TEST_COMMAND, False, subp)
                test(self, to_cr_bytes, TEST_COMMAND + ["--cr"], False, subp)
                test(
                    self, to_crlf_bytes, TEST_COMMAND + ["--crlf"], False, subp
                )
                # universal_newlines = True
                test(self, no_op, TEST_COMMAND, True, subp)
                test(self, no_op, TEST_COMMAND + ["--cr"], True, subp)
                test(self, no_op, TEST_COMMAND + ["--crlf"], True, subp)

        return inner

    return wrapper


class SmokeTests(unittest.TestCase):
    # Regression tests to ensure that subprocess and subprocess2 have the same
    # behavior.
    def _check_res(self, res, stdout, stderr, returncode):
        (out, err), code = res
        self.assertEqual(stdout, out)
        self.assertEqual(stderr, err)
        self.assertEqual(returncode, code)

    def _check_exception(self, subp, e, stdout, stderr, returncode):
        """On exception, look if the exception members are set correctly."""
        self.assertEqual(returncode, e.returncode)
        self.assertEqual(stdout, e.stdout)
        self.assertEqual(stderr, e.stderr)

    def test_check_output_no_stdout(self):
        for subp in (subprocess, subprocess2):
            with self.assertRaises(ValueError):
                subp.check_output(TEST_COMMAND, stdout=subp.PIPE)

    def test_print_exception(self):
        with self.assertRaises(subprocess2.CalledProcessError) as e:
            subprocess2.check_output(TEST_COMMAND + ["--fail", "--stdout"])
        exception_str = str(e.exception)
        # Windows escapes backslashes so check only filename
        self.assertIn(TEST_FILENAME + " --fail --stdout", exception_str)
        self.assertIn(str(e.exception.returncode), exception_str)
        self.assertIn(
            e.exception.stdout.decode("utf-8", "ignore"), exception_str
        )

    @_run_test()
    def test_check_output_throw_stdout(self, c, cmd, un, subp):
        with self.assertRaises(subp.CalledProcessError) as e:
            subp.check_output(
                cmd + ["--fail", "--stdout"], universal_newlines=un
            )
        self._check_exception(subp, e.exception, c("A\nBB\nCCC\n"), None, 64)

    @_run_test()
    def test_check_output_throw_no_stderr(self, c, cmd, un, subp):
        with self.assertRaises(subp.CalledProcessError) as e:
            subp.check_output(
                cmd + ["--fail", "--stderr"], universal_newlines=un
            )
        self._check_exception(subp, e.exception, c(""), None, 64)

    @_run_test()
    def test_check_output_throw_stderr(self, c, cmd, un, subp):
        with self.assertRaises(subp.CalledProcessError) as e:
            subp.check_output(
                cmd + ["--fail", "--stderr"],
                stderr=subp.PIPE,
                universal_newlines=un,
            )
        self._check_exception(subp, e.exception, c(""), c("a\nbb\nccc\n"), 64)

    @_run_test()
    def test_check_output_throw_stderr_stdout(self, c, cmd, un, subp):
        with self.assertRaises(subp.CalledProcessError) as e:
            subp.check_output(
                cmd + ["--fail", "--stderr"],
                stderr=subp.STDOUT,
                universal_newlines=un,
            )
        self._check_exception(subp, e.exception, c("a\nbb\nccc\n"), None, 64)

    def test_check_call_throw(self):
        for subp in (subprocess, subprocess2):
            with self.assertRaises(subp.CalledProcessError) as e:
                subp.check_call(TEST_COMMAND + ["--fail", "--stderr"])
            self._check_exception(subp, e.exception, None, None, 64)

    @_run_test()
    def test_redirect_stderr_to_stdout_pipe(self, c, cmd, un, subp):
        # stderr output into stdout.
        proc = subp.Popen(
            cmd + ["--stderr"],
            stdout=subp.PIPE,
            stderr=subp.STDOUT,
            universal_newlines=un,
        )
        res = proc.communicate(), proc.returncode
        self._check_res(res, c("a\nbb\nccc\n"), None, 0)

    @_run_test()
    def test_redirect_stderr_to_stdout(self, c, cmd, un, subp):
        # stderr output into stdout but stdout is not piped.
        proc = subp.Popen(
            cmd + ["--stderr"], stderr=subprocess2.STDOUT, universal_newlines=un
        )
        res = proc.communicate(), proc.returncode
        self._check_res(res, None, None, 0)

    @_run_test()
    def test_stderr(self, c, cmd, un, subp):
        cmd = ["expr", "1", "/", "0"]
        if sys.platform == "win32":
            cmd = ["cmd.exe", "/c", "exit", "1"]
        p1 = subprocess.Popen(cmd, stderr=subprocess.PIPE, shell=False)
        p2 = subprocess2.Popen(cmd, stderr=subprocess.PIPE, shell=False)
        r1 = p1.communicate()
        r2 = p2.communicate()
        self.assertEqual(r1, r2)

    @_run_test(with_subprocess=False)
    def test_stdin(self, c, cmd, un, subp):
        stdin = c("0123456789")
        res = subprocess2.communicate(
            cmd + ["--read"], stdin=stdin, universal_newlines=un
        )
        self._check_res(res, None, None, 10)

    @_run_test(with_subprocess=False)
    def test_stdin_empty(self, c, cmd, un, subp):
        stdin = c("")
        res = subprocess2.communicate(
            cmd + ["--read"], stdin=stdin, universal_newlines=un
        )
        self._check_res(res, None, None, 0)

    def test_stdin_void(self):
        res = subprocess2.communicate(
            TEST_COMMAND + ["--read"], stdin=subprocess2.DEVNULL
        )
        self._check_res(res, None, None, 0)

    @_run_test(with_subprocess=False)
    def test_stdin_void_stdout(self, c, cmd, un, subp):
        # Make sure a mix ofsubprocess2.DEVNULL andsubprocess2.PIPE works.
        res = subprocess2.communicate(
            cmd + ["--stdout", "--read"],
            stdin=subprocess2.DEVNULL,
            stdout=subprocess2.PIPE,
            universal_newlines=un,
            shell=False,
        )
        self._check_res(res, c("A\nBB\nCCC\n"), None, 0)

    @_run_test(with_subprocess=False)
    def test_stdout_void(self, c, cmd, un, subp):
        res = subprocess2.communicate(
            cmd + ["--stdout", "--stderr"],
            stdout=subprocess2.DEVNULL,
            stderr=subprocess2.PIPE,
            universal_newlines=un,
        )
        self._check_res(res, None, c("a\nbb\nccc\n"), 0)

    @_run_test(with_subprocess=False)
    def test_stderr_void(self, c, cmd, un, subp):
        res = subprocess2.communicate(
            cmd + ["--stdout", "--stderr"],
            stdout=subprocess2.PIPE,
            stderr=subprocess2.DEVNULL,
            universal_newlines=un,
        )
        self._check_res(res, c("A\nBB\nCCC\n"), None, 0)

    @_run_test(with_subprocess=False)
    def test_stdout_void_stderr_redirect(self, c, cmd, un, subp):
        res = subprocess2.communicate(
            cmd + ["--stdout", "--stderr"],
            stdout=subprocess2.DEVNULL,
            stderr=subprocess2.STDOUT,
            universal_newlines=un,
        )
        self._check_res(res, None, None, 0)


if __name__ == "__main__":
    unittest.main()