import unittest

from test import test_support

import subprocess

import sys

import signal

import os

import tempfile

import time

import re

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 = ''

# 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.

def remove_stderr_debug_decorations(stderr):

    return re.sub(r"\[\d+ refs\]\r?\n?$", "", stderr)

class ProcessTestCase(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).

        if hasattr(test_support, "reap_children"):

            test_support.reap_children()

    def tearDown(self):

        # Try to minimize the number of children we have so this test

        # doesn't crash on some buildbots (Alphas in particular).

        if hasattr(test_support, "reap_children"):

            test_support.reap_children()

    def mkstemp(self):

        """wrapper for mkstemp, calling mktemp if mkstemp is not available"""

        if hasattr(tempfile, "mkstemp"):

            return tempfile.mkstemp()

        else:

            fname = tempfile.mktemp()

            return os.open(fname, os.O_RDWR|os.O_CREAT), fname

    #

    # Generic tests

    #

    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_check_call_zero(self):

        # check_call() function with zero return code

        rc = subprocess.check_call([sys.executable, "-c",

                                    "import sys; sys.exit(0)"])

        self.assertEqual(rc, 0)

    def test_check_call_nonzero(self):

        # check_call() function with non-zero return code

        try:

            subprocess.check_call([sys.executable, "-c",

                                   "import sys; sys.exit(47)"])

        except subprocess.CalledProcessError, e:

            self.assertEqual(e.returncode, 47)

        else:

            self.fail("Expected CalledProcessError")

    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_stdin_none(self):

        # .stdin is None when not redirected

        p = subprocess.Popen([sys.executable, "-c", 'print "banana"'],

                         stdout=subprocess.PIPE, stderr=subprocess.PIPE)

        p.wait()

        self.assertEqual(p.stdin, None)

    def test_stdout_none(self):

        # .stdout is None when not redirected

        p = subprocess.Popen([sys.executable, "-c",

                             'print "    this bit of output is from a '

                             'test of stdout in a different '

                             'process ..."'],

                             stdin=subprocess.PIPE, stderr=subprocess.PIPE)

        p.wait()

        self.assertEqual(p.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)

        p.wait()

        self.assertEqual(p.stderr, None)

    def test_executable(self):

        p = subprocess.Popen(["somethingyoudonthave",

                              "-c", "import sys; sys.exit(47)"],

                             executable=sys.executable)

        p.wait()

        self.assertEqual(p.returncode, 47)

    def test_stdin_pipe(self):

        # stdin redirection

        p = subprocess.Popen([sys.executable, "-c",

                         'import sys; sys.exit(sys.stdin.read() == "pear")'],

                        stdin=subprocess.PIPE)

        p.stdin.write("pear")

        p.stdin.close()

        p.wait()

        self.assertEqual(p.returncode, 1)

    def test_stdin_filedes(self):

        # stdin is set to open file descriptor

        tf = tempfile.TemporaryFile()

        d = tf.fileno()

        os.write(d, "pear")

        os.lseek(d, 0, 0)

        p = subprocess.Popen([sys.executable, "-c",

                         'import sys; sys.exit(sys.stdin.read() == "pear")'],

                         stdin=d)

        p.wait()

        self.assertEqual(p.returncode, 1)

    def test_stdin_fileobj(self):

        # stdin is set to open file object

        tf = tempfile.TemporaryFile()

        tf.write("pear")

        tf.seek(0)

        p = subprocess.Popen([sys.executable, "-c",

                         'import sys; sys.exit(sys.stdin.read() == "pear")'],

                         stdin=tf)

        p.wait()

        self.assertEqual(p.returncode, 1)

    def test_stdout_pipe(self):

        # stdout redirection

        p = subprocess.Popen([sys.executable, "-c",

                          'import sys; sys.stdout.write("orange")'],

                         stdout=subprocess.PIPE)

        self.assertEqual(p.stdout.read(), "orange")

    def test_stdout_filedes(self):

        # stdout is set to open file descriptor

        tf = tempfile.TemporaryFile()

        d = tf.fileno()

        p = subprocess.Popen([sys.executable, "-c",

                          'import sys; sys.stdout.write("orange")'],

                         stdout=d)

        p.wait()

        os.lseek(d, 0, 0)

        self.assertEqual(os.read(d, 1024), "orange")

    def test_stdout_fileobj(self):

        # stdout is set to open file object

        tf = tempfile.TemporaryFile()

        p = subprocess.Popen([sys.executable, "-c",

                          'import sys; sys.stdout.write("orange")'],

                         stdout=tf)

        p.wait()

        tf.seek(0)

        self.assertEqual(tf.read(), "orange")

    def test_stderr_pipe(self):

        # stderr redirection

        p = subprocess.Popen([sys.executable, "-c",

                          'import sys; sys.stderr.write("strawberry")'],

                         stderr=subprocess.PIPE)

        self.assertEqual(remove_stderr_debug_decorations(p.stderr.read()),

                         "strawberry")

    def test_stderr_filedes(self):

        # stderr is set to open file descriptor

        tf = tempfile.TemporaryFile()

        d = tf.fileno()

        p = subprocess.Popen([sys.executable, "-c",

                          'import sys; sys.stderr.write("strawberry")'],

                         stderr=d)

        p.wait()

        os.lseek(d, 0, 0)

        self.assertEqual(remove_stderr_debug_decorations(os.read(d, 1024)),

                         "strawberry")

    def test_stderr_fileobj(self):

        # stderr is set to open file object

        tf = tempfile.TemporaryFile()

        p = subprocess.Popen([sys.executable, "-c",

                          'import sys; sys.stderr.write("strawberry")'],

                         stderr=tf)

        p.wait()

        tf.seek(0)

        self.assertEqual(remove_stderr_debug_decorations(tf.read()),

                         "strawberry")

    def test_stdout_stderr_pipe(self):

        # capture stdout and stderr to the same pipe

        p = subprocess.Popen([sys.executable, "-c",

                          'import sys;' \

                          'sys.stdout.write("apple");' \

                          'sys.stdout.flush();' \

                          'sys.stderr.write("orange")'],

                         stdout=subprocess.PIPE,

                         stderr=subprocess.STDOUT)

        output = p.stdout.read()

        stripped = remove_stderr_debug_decorations(output)

        self.assertEqual(stripped, "appleorange")

    def test_stdout_stderr_file(self):

        # capture stdout and stderr to the same open file

        tf = tempfile.TemporaryFile()

        p = subprocess.Popen([sys.executable, "-c",

                          'import sys;' \

                          'sys.stdout.write("apple");' \

                          'sys.stdout.flush();' \

                          'sys.stderr.write("orange")'],

                         stdout=tf,

                         stderr=tf)

        p.wait()

        tf.seek(0)

        output = tf.read()

        stripped = remove_stderr_debug_decorations(output)

        self.assertEqual(stripped, "appleorange")

    def test_stdout_filedes_of_stdout(self):

        # stdout is set to 1 (#1531862).

        cmd = r"import sys, os; sys.exit(os.write(sys.stdout.fileno(), '.\n'))"

        rc = subprocess.call([sys.executable, "-c", cmd], stdout=1)

        self.assertEquals(rc, 2)

    def test_cwd(self):

        tmpdir = tempfile.gettempdir()

        # We cannot use os.path.realpath to canonicalize the path,

        # since it doesn't expand Tru64 {memb} strings. See bug 1063571.

        cwd = os.getcwd()

        os.chdir(tmpdir)

        tmpdir = os.getcwd()

        os.chdir(cwd)

        p = subprocess.Popen([sys.executable, "-c",

                          'import sys,os;' \

                          'sys.stdout.write(os.getcwd())'],

                         stdout=subprocess.PIPE,

                         cwd=tmpdir)

        normcase = os.path.normcase

        self.assertEqual(normcase(p.stdout.read()), normcase(tmpdir))

    def test_env(self):

        newenv = os.environ.copy()

        newenv["FRUIT"] = "orange"

        p = subprocess.Popen([sys.executable, "-c",

                          'import sys,os;' \

                          'sys.stdout.write(os.getenv("FRUIT"))'],

                         stdout=subprocess.PIPE,

                         env=newenv)

        self.assertEqual(p.stdout.read(), "orange")

    def test_communicate_stdin(self):

        p = subprocess.Popen([sys.executable, "-c",

                              'import sys; sys.exit(sys.stdin.read() == "pear")'],

                             stdin=subprocess.PIPE)

        p.communicate("pear")

        self.assertEqual(p.returncode, 1)

    def test_communicate_stdout(self):

        p = subprocess.Popen([sys.executable, "-c",

                              'import sys; sys.stdout.write("pineapple")'],

                             stdout=subprocess.PIPE)

        (stdout, stderr) = p.communicate()

        self.assertEqual(stdout, "pineapple")

        self.assertEqual(stderr, None)

    def test_communicate_stderr(self):

        p = subprocess.Popen([sys.executable, "-c",

                              'import sys; sys.stderr.write("pineapple")'],

                             stderr=subprocess.PIPE)

        (stdout, stderr) = p.communicate()

        self.assertEqual(stdout, None)

        self.assertEqual(remove_stderr_debug_decorations(stderr), "pineapple")

    def test_communicate(self):

        p = subprocess.Popen([sys.executable, "-c",

                          'import sys,os;'

                          'sys.stderr.write("pineapple");'

                          'sys.stdout.write(sys.stdin.read())'],

                         stdin=subprocess.PIPE,

                         stdout=subprocess.PIPE,

                         stderr=subprocess.PIPE)

        (stdout, stderr) = p.communicate("banana")

        self.assertEqual(stdout, "banana")

        self.assertEqual(remove_stderr_debug_decorations(stderr),

                         "pineapple")

    # This test is Linux specific for simplicity to at least have

    # some coverage.  It is not a platform specific bug.

    if os.path.isdir('/proc/%d/fd' % os.getpid()):

        # Test for the fd leak reported in http://bugs.python.org/issue2791.

        def test_communicate_pipe_fd_leak(self):

            fd_directory = '/proc/%d/fd' % os.getpid()

            num_fds_before_popen = len(os.listdir(fd_directory))

            p = subprocess.Popen([sys.executable, '-c', 'print()'],

                                 stdout=subprocess.PIPE)

            p.communicate()

            num_fds_after_communicate = len(os.listdir(fd_directory))

            del p

            num_fds_after_destruction = len(os.listdir(fd_directory))

            self.assertEqual(num_fds_before_popen, num_fds_after_destruction)

            self.assertEqual(num_fds_before_popen, num_fds_after_communicate)

    def test_communicate_returns(self):

        # communicate() should return None if no redirection is active

        p = subprocess.Popen([sys.executable, "-c",

                              "import sys; sys.exit(47)"])

        (stdout, stderr) = p.communicate()

        self.assertEqual(stdout, None)

        self.assertEqual(stderr, None)

    def test_communicate_pipe_buf(self):

        # communicate() with writes larger than pipe_buf

        # This test will probably deadlock rather than fail, if

        # communicate() does not work properly.

        x, y = os.pipe()

        if mswindows:

            pipe_buf = 512

        else:

            pipe_buf = os.fpathconf(x, "PC_PIPE_BUF")

        os.close(x)

        os.close(y)

        p = subprocess.Popen([sys.executable, "-c",

                          'import sys,os;'

                          'sys.stdout.write(sys.stdin.read(47));' \

                          'sys.stderr.write("xyz"*%d);' \

                          'sys.stdout.write(sys.stdin.read())' % pipe_buf],

                         stdin=subprocess.PIPE,

                         stdout=subprocess.PIPE,

                         stderr=subprocess.PIPE)

        string_to_write = "abc"*pipe_buf

        (stdout, stderr) = p.communicate(string_to_write)

        self.assertEqual(stdout, string_to_write)

    def test_writes_before_communicate(self):

        # stdin.write before communicate()

        p = subprocess.Popen([sys.executable, "-c",

                          'import sys,os;' \

                          'sys.stdout.write(sys.stdin.read())'],

                         stdin=subprocess.PIPE,

                         stdout=subprocess.PIPE,

                         stderr=subprocess.PIPE)

        p.stdin.write("banana")

        (stdout, stderr) = p.communicate("split")

        self.assertEqual(stdout, "bananasplit")

        self.assertEqual(remove_stderr_debug_decorations(stderr), "")

    def test_universal_newlines(self):

        p = subprocess.Popen([sys.executable, "-c",

                          'import sys,os;' + SETBINARY +

                          'sys.stdout.write("line1\\n");'

                          'sys.stdout.flush();'

                          'sys.stdout.write("line2\\r");'

                          'sys.stdout.flush();'

                          'sys.stdout.write("line3\\r\\n");'

                          'sys.stdout.flush();'

                          'sys.stdout.write("line4\\r");'

                          'sys.stdout.flush();'

                          'sys.stdout.write("\\nline5");'

                          'sys.stdout.flush();'

                          'sys.stdout.write("\\nline6");'],

                         stdout=subprocess.PIPE,

                         universal_newlines=1)

        stdout = p.stdout.read()

        if hasattr(file, 'newlines'):

            # Interpreter with universal newline support

            self.assertEqual(stdout,

                             "line1\nline2\nline3\nline4\nline5\nline6")

        else:

            # Interpreter without universal newline support

            self.assertEqual(stdout,

                             "line1\nline2\rline3\r\nline4\r\nline5\nline6")

    def test_universal_newlines_communicate(self):

        # universal newlines through communicate()

        p = subprocess.Popen([sys.executable, "-c",

                          'import sys,os;' + SETBINARY +

                          'sys.stdout.write("line1\\n");'

                          'sys.stdout.flush();'

                          'sys.stdout.write("line2\\r");'

                          'sys.stdout.flush();'

                          'sys.stdout.write("line3\\r\\n");'

                          'sys.stdout.flush();'

                          'sys.stdout.write("line4\\r");'

                          'sys.stdout.flush();'

                          'sys.stdout.write("\\nline5");'

                          'sys.stdout.flush();'

                          'sys.stdout.write("\\nline6");'],

                         stdout=subprocess.PIPE, stderr=subprocess.PIPE,

                         universal_newlines=1)

        (stdout, stderr) = p.communicate()

        if hasattr(file, 'newlines'):

            # Interpreter with universal newline support

            self.assertEqual(stdout,

                             "line1\nline2\nline3\nline4\nline5\nline6")

        else:

            # Interpreter without universal newline support

            self.assertEqual(stdout, "line1\nline2\rline3\r\nline4\r\nline5\nline6")

    def test_no_leaking(self):

        # Make sure we leak no resources

        if not hasattr(test_support, "is_resource_enabled") \

               or test_support.is_resource_enabled("subprocess") and not mswindows:

            max_handles = 1026 # too much for most UNIX systems

        else:

            max_handles = 65

        for i in range(max_handles):

            p = subprocess.Popen([sys.executable, "-c",

                    "import sys;sys.stdout.write(sys.stdin.read())"],

                    stdin=subprocess.PIPE,

                    stdout=subprocess.PIPE,

                    stderr=subprocess.PIPE)

            data = p.communicate("lime")[0]

            self.assertEqual(data, "lime")

    def test_list2cmdline(self):

        self.assertEqual(subprocess.list2cmdline(['a b c', 'd', 'e']),

                         '"a b c" d e')

        self.assertEqual(subprocess.list2cmdline(['ab"c', '\\', 'd']),

                         'ab\\"c \\ d')

        self.assertEqual(subprocess.list2cmdline(['ab"c', ' \\', 'd']),

                         'ab\\"c " \\\\" d')

        self.assertEqual(subprocess.list2cmdline(['a\\\\\\b', 'de fg', 'h']),

                         'a\\\\\\b "de fg" h')

        self.assertEqual(subprocess.list2cmdline(['a\\"b', 'c', 'd']),

                         'a\\\\\\"b c d')

        self.assertEqual(subprocess.list2cmdline(['a\\\\b c', 'd', 'e']),

                         '"a\\\\b c" d e')

        self.assertEqual(subprocess.list2cmdline(['a\\\\b\\ c', 'd', 'e']),

                         '"a\\\\b\\ c" d e')

        self.assertEqual(subprocess.list2cmdline(['ab', '']),

                         'ab ""')

        self.assertEqual(subprocess.list2cmdline(['echo', 'foo|bar']),

                         'echo "foo|bar"')

    def test_poll(self):

        p = subprocess.Popen([sys.executable,

                          "-c", "import time; time.sleep(1)"])

        count = 0

        while p.poll() is None:

            time.sleep(0.1)

            count += 1

        # We expect that the poll loop probably went around about 10 times,

        # but, based on system scheduling we can't control, it's possible

        # poll() never returned None.  It "should be" very rare that it

        # didn't go around at least twice.

        self.assert_(count >= 2)

        # Subsequent invocations should just return the returncode

        self.assertEqual(p.poll(), 0)

    def test_wait(self):

        p = subprocess.Popen([sys.executable,

                          "-c", "import time; time.sleep(2)"])

        self.assertEqual(p.wait(), 0)

        # Subsequent invocations should just return the returncode

        self.assertEqual(p.wait(), 0)

    def test_invalid_bufsize(self):

        # an invalid type of the bufsize argument should raise

        # TypeError.

        try:

            subprocess.Popen([sys.executable, "-c", "pass"], "orange")

        except TypeError:

            pass

        else:

            self.fail("Expected TypeError")

    #

    # POSIX tests

    #

    if not mswindows:

        def test_exceptions(self):

            # catched & re-raised exceptions

            try:

                p = subprocess.Popen([sys.executable, "-c", ""],

                                 cwd="/this/path/does/not/exist")

            except OSError, e:

                # The attribute child_traceback should contain "os.chdir"

                # somewhere.

                self.assertNotEqual(e.child_traceback.find("os.chdir"), -1)

            else:

                self.fail("Expected OSError")

        def _suppress_core_files(self):

            """Try to prevent core files from being created.

            Returns previous ulimit if successful, else None.

            """

            try:

                import resource

                old_limit = resource.getrlimit(resource.RLIMIT_CORE)

                resource.setrlimit(resource.RLIMIT_CORE, (0,0))

                return old_limit

            except (ImportError, ValueError, resource.error):

                return None

        def _unsuppress_core_files(self, old_limit):

            """Return core file behavior to default."""

            if old_limit is None:

                return

            try:

                import resource

                resource.setrlimit(resource.RLIMIT_CORE, old_limit)

            except (ImportError, ValueError, resource.error):

                return

        def test_run_abort(self):

            # returncode handles signal termination