import unittest
from unittest import mock
from test import support
import subprocess
import sys
import signal
import io
import itertools
import os
import errno
import tempfile
import time
import traceback
import selectors
import sysconfig
import select
import shutil
import threading
import gc
import textwrap
from test.support import FakePath

try:
    import _testcapi
except ImportError:
    _testcapi = None


if support.PGO:
    raise unittest.SkipTest("test is not helpful for PGO")

mswindows = (sys.platform == "win32")

#
# Depends on the following external programs: Python
#

if mswindows:
    SETBINARY = ('import msvcrt; msvcrt.setmode(sys.stdout.fileno(), '
                                                'os.O_BINARY);')
else:
    SETBINARY = ''

NONEXISTING_CMD = ('nonexisting_i_hope',)
# Ignore errors that indicate the command was not found
NONEXISTING_ERRORS = (FileNotFoundError, NotADirectoryError, PermissionError)

ZERO_RETURN_CMD = (sys.executable, '-c', 'pass')


def setUpModule():
    shell_true = shutil.which('true')
    if shell_true is None:
        return
    if (os.access(shell_true, os.X_OK) and
        subprocess.run([shell_true]).returncode == 0):
        global ZERO_RETURN_CMD
        ZERO_RETURN_CMD = (shell_true,)  # Faster than Python startup.


class BaseTestCase(unittest.TestCase):
    def setUp(self):
        # Try to minimize the number of children we have so this test
        # doesn't crash on some buildbots (Alphas in particular).
        support.reap_children()

    def tearDown(self):
        if not mswindows:
            # subprocess._active is not used on Windows and is set to None.
            for inst in subprocess._active:
                inst.wait()
            subprocess._cleanup()
            self.assertFalse(
                subprocess._active, "subprocess._active not empty"
            )
        self.doCleanups()
        support.reap_children()

    def assertStderrEqual(self, stderr, expected, msg=None):
        # In a debug build, stuff like "[6580 refs]" is printed to stderr at
        # shutdown time.  That frustrates tests trying to check stderr produced
        # from a spawned Python process.
        actual = support.strip_python_stderr(stderr)
        # strip_python_stderr also strips whitespace, so we do too.
        expected = expected.strip()
        self.assertEqual(actual, expected, msg)


class PopenTestException(Exception):
    pass


class PopenExecuteChildRaises(subprocess.Popen):
    """Popen subclass for testing cleanup of subprocess.PIPE filehandles when
    _execute_child fails.
    """
    def _execute_child(self, *args, **kwargs):
        raise PopenTestException("Forced Exception for Test")


class ProcessTestCase(BaseTestCase):

    def test_io_buffered_by_default(self):
        p = subprocess.Popen(ZERO_RETURN_CMD,
                             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE)
        try:
            self.assertIsInstance(p.stdin, io.BufferedIOBase)
            self.assertIsInstance(p.stdout, io.BufferedIOBase)
            self.assertIsInstance(p.stderr, io.BufferedIOBase)
        finally:
            p.stdin.close()
            p.stdout.close()
            p.stderr.close()
            p.wait()

    def test_io_unbuffered_works(self):
        p = subprocess.Popen(ZERO_RETURN_CMD,
                             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE, bufsize=0)
        try:
            self.assertIsInstance(p.stdin, io.RawIOBase)
            self.assertIsInstance(p.stdout, io.RawIOBase)
            self.assertIsInstance(p.stderr, io.RawIOBase)
        finally:
            p.stdin.close()
            p.stdout.close()
            p.stderr.close()
            p.wait()

    def test_call_seq(self):
        # call() function with sequence argument
        rc = subprocess.call([sys.executable, "-c",
                              "import sys; sys.exit(47)"])
        self.assertEqual(rc, 47)

    def test_call_timeout(self):
        # call() function with timeout argument; we want to test that the child
        # process gets killed when the timeout expires.  If the child isn't
        # killed, this call will deadlock since subprocess.call waits for the
        # child.
        self.assertRaises(subprocess.TimeoutExpired, subprocess.call,
                          [sys.executable, "-c", "while True: pass"],
                          timeout=0.1)

    def test_check_call_zero(self):
        # check_call() function with zero return code
        rc = subprocess.check_call(ZERO_RETURN_CMD)
        self.assertEqual(rc, 0)

    def test_check_call_nonzero(self):
        # check_call() function with non-zero return code
        with self.assertRaises(subprocess.CalledProcessError) as c:
            subprocess.check_call([sys.executable, "-c",
                                   "import sys; sys.exit(47)"])
        self.assertEqual(c.exception.returncode, 47)

    def test_check_output(self):
        # check_output() function with zero return code
        output = subprocess.check_output(
                [sys.executable, "-c", "print('BDFL')"])
        self.assertIn(b'BDFL', output)

    def test_check_output_nonzero(self):
        # check_call() function with non-zero return code
        with self.assertRaises(subprocess.CalledProcessError) as c:
            subprocess.check_output(
                    [sys.executable, "-c", "import sys; sys.exit(5)"])
        self.assertEqual(c.exception.returncode, 5)

    def test_check_output_stderr(self):
        # check_output() function stderr redirected to stdout
        output = subprocess.check_output(
                [sys.executable, "-c", "import sys; sys.stderr.write('BDFL')"],
                stderr=subprocess.STDOUT)
        self.assertIn(b'BDFL', output)

    def test_check_output_stdin_arg(self):
        # check_output() can be called with stdin set to a file
        tf = tempfile.TemporaryFile()
        self.addCleanup(tf.close)
        tf.write(b'pear')
        tf.seek(0)
        output = subprocess.check_output(
                [sys.executable, "-c",
                 "import sys; sys.stdout.write(sys.stdin.read().upper())"],
                stdin=tf)
        self.assertIn(b'PEAR', output)

    def test_check_output_input_arg(self):
        # check_output() can be called with input set to a string
        output = subprocess.check_output(
                [sys.executable, "-c",
                 "import sys; sys.stdout.write(sys.stdin.read().upper())"],
                input=b'pear')
        self.assertIn(b'PEAR', output)

    def test_check_output_stdout_arg(self):
        # check_output() refuses to accept 'stdout' argument
        with self.assertRaises(ValueError) as c:
            output = subprocess.check_output(
                    [sys.executable, "-c", "print('will not be run')"],
                    stdout=sys.stdout)
            self.fail("Expected ValueError when stdout arg supplied.")
        self.assertIn('stdout', c.exception.args[0])

    def test_check_output_stdin_with_input_arg(self):
        # check_output() refuses to accept 'stdin' with 'input'
        tf = tempfile.TemporaryFile()
        self.addCleanup(tf.close)
        tf.write(b'pear')
        tf.seek(0)
        with self.assertRaises(ValueError) as c:
            output = subprocess.check_output(
                    [sys.executable, "-c", "print('will not be run')"],
                    stdin=tf, input=b'hare')
            self.fail("Expected ValueError when stdin and input args supplied.")
        self.assertIn('stdin', c.exception.args[0])
        self.assertIn('input', c.exception.args[0])

    def test_check_output_timeout(self):
        # check_output() function with timeout arg
        with self.assertRaises(subprocess.TimeoutExpired) as c:
            output = subprocess.check_output(
                    [sys.executable, "-c",
                     "import sys, time\n"
                     "sys.stdout.write('BDFL')\n"
                     "sys.stdout.flush()\n"
                     "time.sleep(3600)"],
                    # Some heavily loaded buildbots (sparc Debian 3.x) require
                    # this much time to start and print.
                    timeout=3)
            self.fail("Expected TimeoutExpired.")
        self.assertEqual(c.exception.output, b'BDFL')

    def test_call_kwargs(self):
        # call() function with keyword args
        newenv = os.environ.copy()
        newenv["FRUIT"] = "banana"
        rc = subprocess.call([sys.executable, "-c",
                              'import sys, os;'
                              'sys.exit(os.getenv("FRUIT")=="banana")'],
                             env=newenv)
        self.assertEqual(rc, 1)

    def test_invalid_args(self):
        # Popen() called with invalid arguments should raise TypeError
        # but Popen.__del__ should not complain (issue #12085)
        with support.captured_stderr() as s:
            self.assertRaises(TypeError, subprocess.Popen, invalid_arg_name=1)
            argcount = subprocess.Popen.__init__.__code__.co_argcount
            too_many_args = [0] * (argcount + 1)
            self.assertRaises(TypeError, subprocess.Popen, *too_many_args)
        self.assertEqual(s.getvalue(), '')

    def test_stdin_none(self):
        # .stdin is None when not redirected
        p = subprocess.Popen([sys.executable, "-c", 'print("banana")'],
                         stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        self.addCleanup(p.stdout.close)
        self.addCleanup(p.stderr.close)
        p.wait()
        self.assertEqual(p.stdin, None)

    def test_stdout_none(self):
        # .stdout is None when not redirected, and the child's stdout will
        # be inherited from the parent.  In order to test this we run a
        # subprocess in a subprocess:
        # this_test
        #   \-- subprocess created by this test (parent)
        #          \-- subprocess created by the parent subprocess (child)
        # The parent doesn't specify stdout, so the child will use the
        # parent's stdout.  This test checks that the message printed by the
        # child goes to the parent stdout.  The parent also checks that the
        # child's stdout is None.  See #11963.
        code = ('import sys; from subprocess import Popen, PIPE;'
                'p = Popen([sys.executable, "-c", "print(\'test_stdout_none\')"],'
                '          stdin=PIPE, stderr=PIPE);'
                'p.wait(); assert p.stdout is None;')
        p = subprocess.Popen([sys.executable, "-c", code],
                             stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        self.addCleanup(p.stdout.close)
        self.addCleanup(p.stderr.close)
        out, err = p.communicate()
        self.assertEqual(p.returncode, 0, err)
        self.assertEqual(out.rstrip(), b'test_stdout_none')

    def test_stderr_none(self):
        # .stderr is None when not redirected
        p = subprocess.Popen([sys.executable, "-c", 'print("banana")'],
                         stdin=subprocess.PIPE, stdout=subprocess.PIPE)
        self.addCleanup(p.stdout.close)
        self.addCleanup(p.stdin.close)
        p.wait()
        self.assertEqual(p.stderr, None)

    def _assert_python(self, pre_args, **kwargs):
        # We include sys.exit() to prevent the test runner from hanging
        # whenever python is found.
        args = pre_args + ["import sys; sys.exit(47)"]
        p = subprocess.Popen(args, **kwargs)
        p.wait()
        self.assertEqual(47, p.returncode)

    def test_executable(self):
        # Check that the executable argument works.
        #
        # On Unix (non-Mac and non-Windows), Python looks at args[0] to
        # determine where its standard library is, so we need the directory
        # of args[0] to be valid for the Popen() call to Python to succeed.
        # See also issue #16170 and issue #7774.
        doesnotexist = os.path.join(os.path.dirname(sys.executable),
                                    "doesnotexist")
        self._assert_python([doesnotexist, "-c"], executable=sys.executable)

    def test_bytes_executable(self):
        doesnotexist = os.path.join(os.path.dirname(sys.executable),
                                    "doesnotexist")
        self._assert_python([doesnotexist, "-c"],
                            executable=os.fsencode(sys.executable))

    def test_pathlike_executable(self):
        doesnotexist = os.path.join(os.path.dirname(sys.executable),
                                    "doesnotexist")
        self._assert_python([doesnotexist, "-c"],
                            executable=FakePath(sys.executable))

    def test_executable_takes_precedence(self):
        # Check that the executable argument takes precedence over args[0].
        #
        # Verify first that the call succeeds without the executable arg.
        pre_args = [sys.executable, "-c"]
        self._assert_python(pre_args)
        self.assertRaises(NONEXISTING_ERRORS,
                          self._assert_python, pre_args,
                          executable=NONEXISTING_CMD[0])

    @unittest.skipIf(mswindows, "executable argument replaces shell")
    def test_executable_replaces_shell(self):
        # Check that the executable argument replaces the default shell
        # when shell=True.
        self._assert_python([], executable=sys.executable, shell=True)

    @unittest.skipIf(mswindows, "executable argument replaces shell")
    def test_bytes_executable_replaces_shell(self):