Back to index

python3.2  3.2.2
Public Member Functions
test.test_subprocess.ProcessTestCasePOSIXPurePython Class Reference
Inheritance diagram for test.test_subprocess.ProcessTestCasePOSIXPurePython:
Inheritance graph
[legend]
Collaboration diagram for test.test_subprocess.ProcessTestCasePOSIXPurePython:
Collaboration graph
[legend]

List of all members.

Public Member Functions

def setUpClass
def tearDownClass
def test_call_seq
def test_check_call_zero
def test_check_call_nonzero
def test_check_output
def test_check_output_nonzero
def test_check_output_stderr
def test_check_output_stdout_arg
def test_call_kwargs
def test_invalid_args
def test_stdin_none
def test_stdout_none
def test_stderr_none
def test_executable_with_cwd
def test_executable_without_cwd
def test_stdin_pipe
def test_stdin_filedes
def test_stdin_fileobj
def test_stdout_pipe
def test_stdout_filedes
def test_stdout_fileobj
def test_stderr_pipe
def test_stderr_filedes
def test_stderr_fileobj
def test_stdout_stderr_pipe
def test_stdout_stderr_file
def test_stdout_filedes_of_stdout
def test_cwd
def test_env
def test_empty_env
def test_communicate_stdin
def test_communicate_stdout
def test_communicate_stderr
def test_communicate
def test_communicate_pipe_fd_leak
def test_communicate_returns
def test_communicate_pipe_buf
def test_writes_before_communicate
def test_universal_newlines
def test_universal_newlines_communicate
def test_universal_newlines_communicate_stdin
def test_no_leaking
def test_list2cmdline
def test_poll
def test_wait
def test_invalid_bufsize
def test_bufsize_is_none
def test_leaking_fds_on_error
def test_issue8780
def test_handles_closed_on_exception
def test_communicate_epipe
def test_communicate_epipe_only_stdin
def test_communicate_eintr
def setUp
def tearDown
def assertStderrEqual
def test_exceptions
def test_restore_signals
def test_start_new_session
def test_run_abort
def test_preexec
def test_preexec_exception
def test_preexec_gc_module_failure
def test_args_string
def test_invalid_args
def test_shell_sequence
def test_shell_string
def test_call_string
def test_specific_shell
def test_send_signal
def test_kill
def test_terminate
def check_close_std_fds
def test_close_fd_0
def test_close_fd_1
def test_close_fd_2
def test_close_fds_0_1
def test_close_fds_0_2
def test_close_fds_1_2
def test_close_fds_0_1_2
def test_remapping_std_fds
def check_swap_fds
def test_swap_fds
def test_surrogates_error_message
def test_undecodable_env
def test_bytes_program
def test_pipe_cloexec
def test_pipe_cloexec_real_tools
def test_close_fds
def test_pass_fds
def test_stdout_stdin_are_single_inout_fd
def test_stdout_stderr_are_single_inout_fd
def test_stderr_stdin_are_single_inout_fd
def test_wait_when_sigchild_ignored
def test_select_unbuffered
def setUp
def tearDown
def assertStderrEqual

Detailed Description

Definition at line 1543 of file test_subprocess.py.


Member Function Documentation

def test.test_subprocess.BaseTestCase.assertStderrEqual (   self,
  stderr,
  expected,
  msg = None 
) [inherited]

Definition at line 56 of file test_subprocess.py.

00056 
00057     def assertStderrEqual(self, stderr, expected, msg=None):
00058         # In a debug build, stuff like "[6580 refs]" is printed to stderr at
00059         # shutdown time.  That frustrates tests trying to check stderr produced
00060         # from a spawned Python process.
00061         actual = support.strip_python_stderr(stderr)
00062         self.assertEqual(actual, expected, msg)
00063 

Here is the call graph for this function:

Here is the caller graph for this function:

def test.test_subprocess.BaseTestCase.assertStderrEqual (   self,
  stderr,
  expected,
  msg = None 
) [inherited]

Definition at line 56 of file test_subprocess.py.

00056 
00057     def assertStderrEqual(self, stderr, expected, msg=None):
00058         # In a debug build, stuff like "[6580 refs]" is printed to stderr at
00059         # shutdown time.  That frustrates tests trying to check stderr produced
00060         # from a spawned Python process.
00061         actual = support.strip_python_stderr(stderr)
00062         self.assertEqual(actual, expected, msg)
00063 

Here is the call graph for this function:

Here is the caller graph for this function:

Definition at line 1009 of file test_subprocess.py.

01009 
01010     def check_close_std_fds(self, fds):
01011         # Issue #9905: test that subprocess pipes still work properly with
01012         # some standard fds closed
01013         stdin = 0
01014         newfds = []
01015         for a in fds:
01016             b = os.dup(a)
01017             newfds.append(b)
01018             if a == 0:
01019                 stdin = b
01020         try:
01021             for fd in fds:
01022                 os.close(fd)
01023             out, err = subprocess.Popen([sys.executable, "-c",
01024                               'import sys;'
01025                               'sys.stdout.write("apple");'
01026                               'sys.stdout.flush();'
01027                               'sys.stderr.write("orange")'],
01028                        stdin=stdin,
01029                        stdout=subprocess.PIPE,
01030                        stderr=subprocess.PIPE).communicate()
01031             err = support.strip_python_stderr(err)
01032             self.assertEqual((out, err), (b'apple', b'orange'))
01033         finally:
01034             for b, a in zip(newfds, fds):
01035                 os.dup2(b, a)
01036             for b in newfds:
01037                 os.close(b)

Here is the call graph for this function:

Here is the caller graph for this function:

def test.test_subprocess.POSIXProcessTestCase.check_swap_fds (   self,
  stdin_no,
  stdout_no,
  stderr_no 
) [inherited]

Definition at line 1109 of file test_subprocess.py.

01109 
01110     def check_swap_fds(self, stdin_no, stdout_no, stderr_no):
01111         # open up some temporary files
01112         temps = [mkstemp() for i in range(3)]
01113         temp_fds = [fd for fd, fname in temps]
01114         try:
01115             # unlink the files -- we won't need to reopen them
01116             for fd, fname in temps:
01117                 os.unlink(fname)
01118 
01119             # save a copy of the standard file descriptors
01120             saved_fds = [os.dup(fd) for fd in range(3)]
01121             try:
01122                 # duplicate the temp files over the standard fd's 0, 1, 2
01123                 for fd, temp_fd in enumerate(temp_fds):
01124                     os.dup2(temp_fd, fd)
01125 
01126                 # write some data to what will become stdin, and rewind
01127                 os.write(stdin_no, b"STDIN")
01128                 os.lseek(stdin_no, 0, 0)
01129 
01130                 # now use those files in the given order, so that subprocess
01131                 # has to rearrange them in the child
01132                 p = subprocess.Popen([sys.executable, "-c",
01133                     'import sys; got = sys.stdin.read();'
01134                     'sys.stdout.write("got %s"%got); sys.stderr.write("err")'],
01135                     stdin=stdin_no,
01136                     stdout=stdout_no,
01137                     stderr=stderr_no)
01138                 p.wait()
01139 
01140                 for fd in temp_fds:
01141                     os.lseek(fd, 0, 0)
01142 
01143                 out = os.read(stdout_no, 1024)
01144                 err = support.strip_python_stderr(os.read(stderr_no, 1024))
01145             finally:
01146                 for std, saved in enumerate(saved_fds):
01147                     os.dup2(saved, std)
01148                     os.close(saved)
01149 
01150             self.assertEqual(out, b"got STDIN")
01151             self.assertEqual(err, b"err")
01152 
01153         finally:
01154             for fd in temp_fds:
01155                 os.close(fd)

Here is the call graph for this function:

Here is the caller graph for this function:

def test.test_subprocess.BaseTestCase.setUp (   self) [inherited]

Reimplemented in test.test_subprocess.CommandsWithSpaces, and test.test_subprocess.ProcessTestCaseNoPoll.

Definition at line 45 of file test_subprocess.py.

00045 
00046     def setUp(self):
00047         # Try to minimize the number of children we have so this test
00048         # doesn't crash on some buildbots (Alphas in particular).
00049         support.reap_children()

Here is the caller graph for this function:

def test.test_subprocess.BaseTestCase.setUp (   self) [inherited]

Reimplemented in test.test_subprocess.CommandsWithSpaces, and test.test_subprocess.ProcessTestCaseNoPoll.

Definition at line 45 of file test_subprocess.py.

00045 
00046     def setUp(self):
00047         # Try to minimize the number of children we have so this test
00048         # doesn't crash on some buildbots (Alphas in particular).
00049         support.reap_children()

Here is the caller graph for this function:

Definition at line 1545 of file test_subprocess.py.

01545 
01546     def setUpClass(cls):
01547         global subprocess
01548         assert subprocess._posixsubprocess
01549         # Reimport subprocess while forcing _posixsubprocess to not exist.
01550         with support.check_warnings(('.*_posixsubprocess .* not being used.*',
01551                                      RuntimeWarning)):
01552             subprocess = support.import_fresh_module(
01553                     'subprocess', blocked=['_posixsubprocess'])
01554         assert not subprocess._posixsubprocess

def test.test_subprocess.BaseTestCase.tearDown (   self) [inherited]

Reimplemented in test.test_subprocess.CommandsWithSpaces, and test.test_subprocess.ProcessTestCaseNoPoll.

Definition at line 50 of file test_subprocess.py.

00050 
00051     def tearDown(self):
00052         for inst in subprocess._active:
00053             inst.wait()
00054         subprocess._cleanup()
00055         self.assertFalse(subprocess._active, "subprocess._active not empty")

Here is the call graph for this function:

Here is the caller graph for this function:

def test.test_subprocess.BaseTestCase.tearDown (   self) [inherited]

Reimplemented in test.test_subprocess.CommandsWithSpaces, and test.test_subprocess.ProcessTestCaseNoPoll.

Definition at line 50 of file test_subprocess.py.

00050 
00051     def tearDown(self):
00052         for inst in subprocess._active:
00053             inst.wait()
00054         subprocess._cleanup()
00055         self.assertFalse(subprocess._active, "subprocess._active not empty")

Here is the call graph for this function:

Here is the caller graph for this function:

Definition at line 1556 of file test_subprocess.py.

01556 
01557     def tearDownClass(cls):
01558         global subprocess
01559         # Reimport subprocess as it should be, restoring order to the universe.
01560         subprocess = support.import_fresh_module('subprocess')
01561         assert subprocess._posixsubprocess
01562 

Definition at line 894 of file test_subprocess.py.

00894 
00895     def test_args_string(self):
00896         # args is a string
00897         fd, fname = mkstemp()
00898         # reopen in text mode
00899         with open(fd, "w", errors="surrogateescape") as fobj:
00900             fobj.write("#!/bin/sh\n")
00901             fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" %
00902                        sys.executable)
00903         os.chmod(fname, 0o700)
00904         p = subprocess.Popen(fname)
00905         p.wait()
00906         os.remove(fname)
00907         self.assertEqual(p.returncode, 47)

Here is the call graph for this function:

Definition at line 635 of file test_subprocess.py.

00635 
00636     def test_bufsize_is_none(self):
00637         # bufsize=None should be the same as bufsize=0.
00638         p = subprocess.Popen([sys.executable, "-c", "pass"], None)
00639         self.assertEqual(p.wait(), 0)
00640         # Again with keyword arg
00641         p = subprocess.Popen([sys.executable, "-c", "pass"], bufsize=None)
00642         self.assertEqual(p.wait(), 0)

Here is the call graph for this function:

Definition at line 1214 of file test_subprocess.py.

01214 
01215     def test_bytes_program(self):
01216         abs_program = os.fsencode(sys.executable)
01217         path, program = os.path.split(sys.executable)
01218         program = os.fsencode(program)
01219 
01220         # absolute bytes path
01221         exitcode = subprocess.call([abs_program, "-c", "pass"])
01222         self.assertEqual(exitcode, 0)
01223 
01224         # bytes program, unicode PATH
01225         env = os.environ.copy()
01226         env["PATH"] = path
01227         exitcode = subprocess.call([program, "-c", "pass"], env=env)
01228         self.assertEqual(exitcode, 0)
01229 
01230         # bytes program, bytes PATH
01231         envb = os.environb.copy()
01232         envb[b"PATH"] = os.fsencode(path)
01233         exitcode = subprocess.call([program, "-c", "pass"], env=envb)
01234         self.assertEqual(exitcode, 0)

Here is the call graph for this function:

Definition at line 114 of file test_subprocess.py.

00114 
00115     def test_call_kwargs(self):
00116         # call() function with keyword args
00117         newenv = os.environ.copy()
00118         newenv["FRUIT"] = "banana"
00119         rc = subprocess.call([sys.executable, "-c",
00120                               'import sys, os;'
00121                               'sys.exit(os.getenv("FRUIT")=="banana")'],
00122                              env=newenv)
00123         self.assertEqual(rc, 1)

Here is the call graph for this function:

Definition at line 66 of file test_subprocess.py.

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

Here is the call graph for this function:

Definition at line 939 of file test_subprocess.py.

00939 
00940     def test_call_string(self):
00941         # call() function with string argument on UNIX
00942         fd, fname = mkstemp()
00943         # reopen in text mode
00944         with open(fd, "w", errors="surrogateescape") as fobj:
00945             fobj.write("#!/bin/sh\n")
00946             fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" %
00947                        sys.executable)
00948         os.chmod(fname, 0o700)
00949         rc = subprocess.call(fname)
00950         os.remove(fname)
00951         self.assertEqual(rc, 47)

Here is the call graph for this function:

Definition at line 78 of file test_subprocess.py.

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

Here is the call graph for this function:

Definition at line 72 of file test_subprocess.py.

00072 
00073     def test_check_call_zero(self):
00074         # check_call() function with zero return code
00075         rc = subprocess.check_call([sys.executable, "-c",
00076                                     "import sys; sys.exit(0)"])
00077         self.assertEqual(rc, 0)

Here is the call graph for this function:

Definition at line 85 of file test_subprocess.py.

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

Here is the call graph for this function:

Definition at line 91 of file test_subprocess.py.

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

Here is the call graph for this function:

Definition at line 98 of file test_subprocess.py.

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

Here is the call graph for this function:

Definition at line 105 of file test_subprocess.py.

00105 
00106     def test_check_output_stdout_arg(self):
00107         # check_output() function stderr redirected to stdout
00108         with self.assertRaises(ValueError) as c:
00109             output = subprocess.check_output(
00110                     [sys.executable, "-c", "print('will not be run')"],
00111                     stdout=sys.stdout)
00112             self.fail("Expected ValueError when stdout arg supplied.")
00113         self.assertIn('stdout', c.exception.args[0])

Here is the call graph for this function:

Definition at line 1038 of file test_subprocess.py.

01038 
01039     def test_close_fd_0(self):
01040         self.check_close_std_fds([0])

Here is the call graph for this function:

Definition at line 1041 of file test_subprocess.py.

01041 
01042     def test_close_fd_1(self):
01043         self.check_close_std_fds([1])

Here is the call graph for this function:

Definition at line 1044 of file test_subprocess.py.

01044 
01045     def test_close_fd_2(self):
01046         self.check_close_std_fds([2])

Here is the call graph for this function:

Definition at line 1289 of file test_subprocess.py.

01289 
01290     def test_close_fds(self):
01291         fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
01292 
01293         fds = os.pipe()
01294         self.addCleanup(os.close, fds[0])
01295         self.addCleanup(os.close, fds[1])
01296 
01297         open_fds = set(fds)
01298 
01299         p = subprocess.Popen([sys.executable, fd_status],
01300                              stdout=subprocess.PIPE, close_fds=False)
01301         output, ignored = p.communicate()
01302         remaining_fds = set(map(int, output.split(b',')))
01303 
01304         self.assertEqual(remaining_fds & open_fds, open_fds,
01305                          "Some fds were closed")
01306 
01307         p = subprocess.Popen([sys.executable, fd_status],
01308                              stdout=subprocess.PIPE, close_fds=True)
01309         output, ignored = p.communicate()
01310         remaining_fds = set(map(int, output.split(b',')))
01311 
01312         self.assertFalse(remaining_fds & open_fds,
01313                          "Some fds were left open")
01314         self.assertIn(1, remaining_fds, "Subprocess failed")

Here is the call graph for this function:

Definition at line 1047 of file test_subprocess.py.

01047 
01048     def test_close_fds_0_1(self):
01049         self.check_close_std_fds([0, 1])

Here is the call graph for this function:

Definition at line 1056 of file test_subprocess.py.

01056 
01057     def test_close_fds_0_1_2(self):
01058         # Issue #10806: test that subprocess pipes still work properly with
01059         # all standard fds closed.
01060         self.check_close_std_fds([0, 1, 2])

Here is the call graph for this function:

Definition at line 1050 of file test_subprocess.py.

01050 
01051     def test_close_fds_0_2(self):
01052         self.check_close_std_fds([0, 2])

Here is the call graph for this function:

Definition at line 1053 of file test_subprocess.py.

01053 
01054     def test_close_fds_1_2(self):
01055         self.check_close_std_fds([1, 2])

Here is the call graph for this function:

Definition at line 386 of file test_subprocess.py.

00386 
00387     def test_communicate(self):
00388         p = subprocess.Popen([sys.executable, "-c",
00389                               'import sys,os;'
00390                               'sys.stderr.write("pineapple");'
00391                               'sys.stdout.write(sys.stdin.read())'],
00392                              stdin=subprocess.PIPE,
00393                              stdout=subprocess.PIPE,
00394                              stderr=subprocess.PIPE)
00395         self.addCleanup(p.stdout.close)
00396         self.addCleanup(p.stderr.close)
00397         self.addCleanup(p.stdin.close)
00398         (stdout, stderr) = p.communicate(b"banana")
00399         self.assertEqual(stdout, b"banana")
00400         self.assertStderrEqual(stderr, b"pineapple")

Here is the call graph for this function:

Definition at line 711 of file test_subprocess.py.

00711 
00712     def test_communicate_eintr(self):
00713         # Issue #12493: communicate() should handle EINTR
00714         def handler(signum, frame):
00715             pass
00716         old_handler = signal.signal(signal.SIGALRM, handler)
00717         self.addCleanup(signal.signal, signal.SIGALRM, old_handler)
00718 
00719         # the process is running for 2 seconds
00720         args = [sys.executable, "-c", 'import time; time.sleep(2)']
00721         for stream in ('stdout', 'stderr'):
00722             kw = {stream: subprocess.PIPE}
00723             with subprocess.Popen(args, **kw) as process:
00724                 signal.alarm(1)
00725                 # communicate() will be interrupted by SIGALRM
00726                 process.communicate()
00727 
00728 
# context manager

Definition at line 690 of file test_subprocess.py.

00690 
00691     def test_communicate_epipe(self):
00692         # Issue 10963: communicate() should hide EPIPE
00693         p = subprocess.Popen([sys.executable, "-c", 'pass'],
00694                              stdin=subprocess.PIPE,
00695                              stdout=subprocess.PIPE,
00696                              stderr=subprocess.PIPE)
00697         self.addCleanup(p.stdout.close)
00698         self.addCleanup(p.stderr.close)
00699         self.addCleanup(p.stdin.close)
00700         p.communicate(b"x" * 2**20)

Here is the call graph for this function:

Definition at line 701 of file test_subprocess.py.

00701 
00702     def test_communicate_epipe_only_stdin(self):
00703         # Issue 10963: communicate() should hide EPIPE
00704         p = subprocess.Popen([sys.executable, "-c", 'pass'],
00705                              stdin=subprocess.PIPE)
00706         self.addCleanup(p.stdin.close)
00707         time.sleep(2)
00708         p.communicate(b"x" * 2**20)

Here is the call graph for this function:

Definition at line 432 of file test_subprocess.py.

00432 
00433     def test_communicate_pipe_buf(self):
00434         # communicate() with writes larger than pipe_buf
00435         # This test will probably deadlock rather than fail, if
00436         # communicate() does not work properly.
00437         x, y = os.pipe()
00438         if mswindows:
00439             pipe_buf = 512
00440         else:
00441             pipe_buf = os.fpathconf(x, "PC_PIPE_BUF")
00442         os.close(x)
00443         os.close(y)
00444         p = subprocess.Popen([sys.executable, "-c",
00445                               'import sys,os;'
00446                               'sys.stdout.write(sys.stdin.read(47));'
00447                               'sys.stderr.write("xyz"*%d);'
00448                               'sys.stdout.write(sys.stdin.read())' % pipe_buf],
00449                              stdin=subprocess.PIPE,
00450                              stdout=subprocess.PIPE,
00451                              stderr=subprocess.PIPE)
00452         self.addCleanup(p.stdout.close)
00453         self.addCleanup(p.stderr.close)
00454         self.addCleanup(p.stdin.close)
00455         string_to_write = b"abc"*pipe_buf
00456         (stdout, stderr) = p.communicate(string_to_write)
00457         self.assertEqual(stdout, string_to_write)

Here is the call graph for this function:

Definition at line 402 of file test_subprocess.py.

00402 
00403     def test_communicate_pipe_fd_leak(self):
00404         for stdin_pipe in (False, True):
00405             for stdout_pipe in (False, True):
00406                 for stderr_pipe in (False, True):
00407                     options = {}
00408                     if stdin_pipe:
00409                         options['stdin'] = subprocess.PIPE
00410                     if stdout_pipe:
00411                         options['stdout'] = subprocess.PIPE
00412                     if stderr_pipe:
00413                         options['stderr'] = subprocess.PIPE
00414                     if not options:
00415                         continue
00416                     p = subprocess.Popen((sys.executable, "-c", "pass"), **options)
00417                     p.communicate()
00418                     if p.stdin is not None:
00419                         self.assertTrue(p.stdin.closed)
00420                     if p.stdout is not None:
00421                         self.assertTrue(p.stdout.closed)
00422                     if p.stderr is not None:
00423                         self.assertTrue(p.stderr.closed)

Here is the call graph for this function:

Definition at line 424 of file test_subprocess.py.

00424 
00425     def test_communicate_returns(self):
00426         # communicate() should return None if no redirection is active
00427         p = subprocess.Popen([sys.executable, "-c",
00428                               "import sys; sys.exit(47)"])
00429         (stdout, stderr) = p.communicate()
00430         self.assertEqual(stdout, None)
00431         self.assertEqual(stderr, None)

Here is the call graph for this function:

Definition at line 378 of file test_subprocess.py.

00378 
00379     def test_communicate_stderr(self):
00380         p = subprocess.Popen([sys.executable, "-c",
00381                               'import sys; sys.stderr.write("pineapple")'],
00382                              stderr=subprocess.PIPE)
00383         (stdout, stderr) = p.communicate()
00384         self.assertEqual(stdout, None)
00385         self.assertStderrEqual(stderr, b"pineapple")

Here is the call graph for this function:

Reimplemented in test.test_subprocess.ContextManagerTests.

Definition at line 362 of file test_subprocess.py.

00362 
00363     def test_communicate_stdin(self):
00364         p = subprocess.Popen([sys.executable, "-c",
00365                               'import sys;'
00366                               'sys.exit(sys.stdin.read() == "pear")'],
00367                              stdin=subprocess.PIPE)
00368         p.communicate(b"pear")
00369         self.assertEqual(p.returncode, 1)

Here is the call graph for this function:

Definition at line 370 of file test_subprocess.py.

00370 
00371     def test_communicate_stdout(self):
00372         p = subprocess.Popen([sys.executable, "-c",
00373                               'import sys; sys.stdout.write("pineapple")'],
00374                              stdout=subprocess.PIPE)
00375         (stdout, stderr) = p.communicate()
00376         self.assertEqual(stdout, b"pineapple")
00377         self.assertEqual(stderr, None)

Here is the call graph for this function:

Definition at line 313 of file test_subprocess.py.

00313 
00314     def test_cwd(self):
00315         tmpdir = tempfile.gettempdir()
00316         # We cannot use os.path.realpath to canonicalize the path,
00317         # since it doesn't expand Tru64 {memb} strings. See bug 1063571.
00318         cwd = os.getcwd()
00319         os.chdir(tmpdir)
00320         tmpdir = os.getcwd()
00321         os.chdir(cwd)
00322         p = subprocess.Popen([sys.executable, "-c",
00323                               'import sys,os;'
00324                               'sys.stdout.write(os.getcwd())'],
00325                              stdout=subprocess.PIPE,
00326                              cwd=tmpdir)
00327         self.addCleanup(p.stdout.close)
00328         normcase = os.path.normcase
00329         self.assertEqual(normcase(p.stdout.read().decode("utf-8")),
00330                          normcase(tmpdir))

Here is the call graph for this function:

Definition at line 349 of file test_subprocess.py.

00349 
00350     def test_empty_env(self):
00351         with subprocess.Popen([sys.executable, "-c",
00352                                'import os; '
00353                                'print(list(os.environ.keys()))'],
00354                               stdout=subprocess.PIPE,
00355                               env={}) as p:
00356             stdout, stderr = p.communicate()
00357             self.assertIn(stdout.strip(),
00358                 (b"[]",
00359                  # Mac OS X adds __CF_USER_TEXT_ENCODING variable to an empty
00360                  # environment
00361                  b"['__CF_USER_TEXT_ENCODING']"))

Here is the call graph for this function:

Definition at line 331 of file test_subprocess.py.

00331 
00332     def test_env(self):
00333         newenv = os.environ.copy()
00334         newenv["FRUIT"] = "orange"
00335         with subprocess.Popen([sys.executable, "-c",
00336                                'import sys,os;'
00337                                'sys.stdout.write(os.getenv("FRUIT"))'],
00338                               stdout=subprocess.PIPE,
00339                               env=newenv) as p:
00340             stdout, stderr = p.communicate()
00341             self.assertEqual(stdout, b"orange")

Here is the call graph for this function:

Definition at line 771 of file test_subprocess.py.

00771 
00772     def test_exceptions(self):
00773         nonexistent_dir = "/_this/pa.th/does/not/exist"
00774         try:
00775             os.chdir(nonexistent_dir)
00776         except OSError as e:
00777             # This avoids hard coding the errno value or the OS perror()
00778             # string and instead capture the exception that we want to see
00779             # below for comparison.
00780             desired_exception = e
00781             desired_exception.strerror += ': ' + repr(sys.executable)
00782         else:
00783             self.fail("chdir to nonexistant directory %s succeeded." %
00784                       nonexistent_dir)
00785 
00786         # Error in the child re-raised in the parent.
00787         try:
00788             p = subprocess.Popen([sys.executable, "-c", ""],
00789                                  cwd=nonexistent_dir)
00790         except OSError as e:
00791             # Test that the child process chdir failure actually makes
00792             # it up to the parent process as the correct exception.
00793             self.assertEqual(desired_exception.errno, e.errno)
00794             self.assertEqual(desired_exception.strerror, e.strerror)
00795         else:
00796             self.fail("Expected OSError: %s" % desired_exception)

Here is the call graph for this function:

Definition at line 164 of file test_subprocess.py.

00164 
00165     def test_executable_with_cwd(self):
00166         python_dir = os.path.dirname(os.path.realpath(sys.executable))
00167         p = subprocess.Popen(["somethingyoudonthave", "-c",
00168                               "import sys; sys.exit(47)"],
00169                              executable=sys.executable, cwd=python_dir)
00170         p.wait()
00171         self.assertEqual(p.returncode, 47)

Here is the call graph for this function:

Definition at line 174 of file test_subprocess.py.

00174 
00175     def test_executable_without_cwd(self):
00176         # For a normal installation, it should work without 'cwd'
00177         # argument.  For test runs in the build directory, see #7774.
00178         p = subprocess.Popen(["somethingyoudonthave", "-c",
00179                               "import sys; sys.exit(47)"],
00180                              executable=sys.executable)
00181         p.wait()
00182         self.assertEqual(p.returncode, 47)

Here is the call graph for this function:

Definition at line 670 of file test_subprocess.py.

00670 
00671     def test_handles_closed_on_exception(self):
00672         # If CreateProcess exits with an error, ensure the
00673         # duplicate output handles are released
00674         ifhandle, ifname = mkstemp()
00675         ofhandle, ofname = mkstemp()
00676         efhandle, efname = mkstemp()
00677         try:
00678             subprocess.Popen (["*"], stdin=ifhandle, stdout=ofhandle,
00679               stderr=efhandle)
00680         except OSError:
00681             os.close(ifhandle)
00682             os.remove(ifname)
00683             os.close(ofhandle)
00684             os.remove(ofname)
00685             os.close(efhandle)
00686             os.remove(efname)
00687         self.assertFalse(os.path.exists(ifname))
00688         self.assertFalse(os.path.exists(ofname))
00689         self.assertFalse(os.path.exists(efname))

Here is the call graph for this function:

Reimplemented in test.test_subprocess.ContextManagerTests.

Definition at line 124 of file test_subprocess.py.

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

Here is the call graph for this function:

Definition at line 908 of file test_subprocess.py.

00908 
00909     def test_invalid_args(self):
00910         # invalid arguments should raise ValueError
00911         self.assertRaises(ValueError, subprocess.call,
00912                           [sys.executable, "-c",
00913                            "import sys; sys.exit(47)"],
00914                           startupinfo=47)
00915         self.assertRaises(ValueError, subprocess.call,
00916                           [sys.executable, "-c",
00917                            "import sys; sys.exit(47)"],
00918                           creationflags=47)

Here is the call graph for this function:

Definition at line 629 of file test_subprocess.py.

00629 
00630     def test_invalid_bufsize(self):
00631         # an invalid type of the bufsize argument should raise
00632         # TypeError.
00633         with self.assertRaises(TypeError):
00634             subprocess.Popen([sys.executable, "-c", "pass"], "orange")

Here is the call graph for this function:

Definition at line 659 of file test_subprocess.py.

00659 
00660     def test_issue8780(self):
00661         # Ensure that stdout is inherited from the parent
00662         # if stdout=PIPE is not used
00663         code = ';'.join((
00664             'import subprocess, sys',
00665             'retcode = subprocess.call('
00666                 "[sys.executable, '-c', 'print(\"Hello World!\")'])",
00667             'assert retcode == 0'))
00668         output = subprocess.check_output([sys.executable, '-c', code])
00669         self.assertTrue(output.startswith(b'Hello World!'), ascii(output))

Here is the call graph for this function:

Definition at line 997 of file test_subprocess.py.

00997 
00998     def test_kill(self):
00999         p = self._kill_process('kill')
01000         _, stderr = p.communicate()
01001         self.assertStderrEqual(stderr, b'')
01002         self.assertEqual(p.wait(), -signal.SIGKILL)

Here is the call graph for this function:

Definition at line 643 of file test_subprocess.py.

00643 
00644     def test_leaking_fds_on_error(self):
00645         # see bug #5179: Popen leaks file descriptors to PIPEs if
00646         # the child fails to execute; this will eventually exhaust
00647         # the maximum number of open fds. 1024 seems a very common
00648         # value for that limit, but Windows has 2048, so we loop
00649         # 1024 times (each call leaked two fds).
00650         for i in range(1024):
00651             # Windows raises IOError.  Others raise OSError.
00652             with self.assertRaises(EnvironmentError) as c:
00653                 subprocess.Popen(['nonexisting_i_hope'],
00654                                  stdout=subprocess.PIPE,
00655                                  stderr=subprocess.PIPE)
00656             # ignore errors that indicate the command was not found
00657             if c.exception.errno not in (errno.ENOENT, errno.EACCES):
00658                 raise c.exception

Here is the call graph for this function:

Definition at line 586 of file test_subprocess.py.

00586 
00587     def test_list2cmdline(self):
00588         self.assertEqual(subprocess.list2cmdline(['a b c', 'd', 'e']),
00589                          '"a b c" d e')
00590         self.assertEqual(subprocess.list2cmdline(['ab"c', '\\', 'd']),
00591                          'ab\\"c \\ d')
00592         self.assertEqual(subprocess.list2cmdline(['ab"c', ' \\', 'd']),
00593                          'ab\\"c " \\\\" d')
00594         self.assertEqual(subprocess.list2cmdline(['a\\\\\\b', 'de fg', 'h']),
00595                          'a\\\\\\b "de fg" h')
00596         self.assertEqual(subprocess.list2cmdline(['a\\"b', 'c', 'd']),
00597                          'a\\\\\\"b c d')
00598         self.assertEqual(subprocess.list2cmdline(['a\\\\b c', 'd', 'e']),
00599                          '"a\\\\b c" d e')
00600         self.assertEqual(subprocess.list2cmdline(['a\\\\b\\ c', 'd', 'e']),
00601                          '"a\\\\b\\ c" d e')
00602         self.assertEqual(subprocess.list2cmdline(['ab', '']),
00603                          'ab ""')
00604 

Here is the call graph for this function:

Definition at line 547 of file test_subprocess.py.

00547 
00548     def test_no_leaking(self):
00549         # Make sure we leak no resources
00550         if not mswindows:
00551             max_handles = 1026 # too much for most UNIX systems
00552         else:
00553             max_handles = 2050 # too much for (at least some) Windows setups
00554         handles = []
00555         tmpdir = tempfile.mkdtemp()
00556         try:
00557             for i in range(max_handles):
00558                 try:
00559                     tmpfile = os.path.join(tmpdir, support.TESTFN)
00560                     handles.append(os.open(tmpfile, os.O_WRONLY|os.O_CREAT))
00561                 except OSError as e:
00562                     if e.errno != errno.EMFILE:
00563                         raise
00564                     break
00565             else:
00566                 self.skipTest("failed to reach the file descriptor limit "
00567                     "(tried %d)" % max_handles)
00568             # Close a couple of them (should be enough for a subprocess)
00569             for i in range(10):
00570                 os.close(handles.pop())
00571             # Loop creating some subprocesses. If one of them leaks some fds,
00572             # the next loop iteration will fail by reaching the max fd limit.
00573             for i in range(15):
00574                 p = subprocess.Popen([sys.executable, "-c",
00575                                       "import sys;"
00576                                       "sys.stdout.write(sys.stdin.read())"],
00577                                      stdin=subprocess.PIPE,
00578                                      stdout=subprocess.PIPE,
00579                                      stderr=subprocess.PIPE)
00580                 data = p.communicate(b"lime")[0]
00581                 self.assertEqual(data, b"lime")
00582         finally:
00583             for h in handles:
00584                 os.close(h)
00585             shutil.rmtree(tmpdir)

Here is the call graph for this function:

Definition at line 1320 of file test_subprocess.py.

01320 
01321     def test_pass_fds(self):
01322         fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
01323 
01324         open_fds = set()
01325 
01326         for x in range(5):
01327             fds = os.pipe()
01328             self.addCleanup(os.close, fds[0])
01329             self.addCleanup(os.close, fds[1])
01330             open_fds.update(fds)
01331 
01332         for fd in open_fds:
01333             p = subprocess.Popen([sys.executable, fd_status],
01334                                  stdout=subprocess.PIPE, close_fds=True,
01335                                  pass_fds=(fd, ))
01336             output, ignored = p.communicate()
01337 
01338             remaining_fds = set(map(int, output.split(b',')))
01339             to_be_closed = open_fds - {fd}
01340 
01341             self.assertIn(fd, remaining_fds, "fd to be passed not passed")
01342             self.assertFalse(remaining_fds & to_be_closed,
01343                              "fd to be closed passed")
01344 
01345             # pass_fds overrides close_fds with a warning.
01346             with self.assertWarns(RuntimeWarning) as context:
01347                 self.assertFalse(subprocess.call(
01348                         [sys.executable, "-c", "import sys; sys.exit(0)"],
01349                         close_fds=False, pass_fds=(fd, )))
01350             self.assertIn('overriding close_fds', str(context.warning))

Here is the call graph for this function:

Definition at line 1235 of file test_subprocess.py.

01235 
01236     def test_pipe_cloexec(self):
01237         sleeper = support.findfile("input_reader.py", subdir="subprocessdata")
01238         fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
01239 
01240         p1 = subprocess.Popen([sys.executable, sleeper],
01241                               stdin=subprocess.PIPE, stdout=subprocess.PIPE,
01242                               stderr=subprocess.PIPE, close_fds=False)
01243 
01244         self.addCleanup(p1.communicate, b'')
01245 
01246         p2 = subprocess.Popen([sys.executable, fd_status],
01247                               stdout=subprocess.PIPE, close_fds=False)
01248 
01249         output, error = p2.communicate()
01250         result_fds = set(map(int, output.split(b',')))
01251         unwanted_fds = set([p1.stdin.fileno(), p1.stdout.fileno(),
01252                             p1.stderr.fileno()])
01253 
01254         self.assertFalse(result_fds & unwanted_fds,
01255                          "Expected no fds from %r to be open in child, "
01256                          "found %r" %
01257                               (unwanted_fds, result_fds & unwanted_fds))

Here is the call graph for this function:

Definition at line 1258 of file test_subprocess.py.

01258 
01259     def test_pipe_cloexec_real_tools(self):
01260         qcat = support.findfile("qcat.py", subdir="subprocessdata")
01261         qgrep = support.findfile("qgrep.py", subdir="subprocessdata")
01262 
01263         subdata = b'zxcvbn'
01264         data = subdata * 4 + b'\n'
01265 
01266         p1 = subprocess.Popen([sys.executable, qcat],
01267                               stdin=subprocess.PIPE, stdout=subprocess.PIPE,
01268                               close_fds=False)
01269 
01270         p2 = subprocess.Popen([sys.executable, qgrep, subdata],
01271                               stdin=p1.stdout, stdout=subprocess.PIPE,
01272                               close_fds=False)
01273 
01274         self.addCleanup(p1.wait)
01275         self.addCleanup(p2.wait)
01276         self.addCleanup(p1.terminate)
01277         self.addCleanup(p2.terminate)
01278 
01279         p1.stdin.write(data)
01280         p1.stdin.close()
01281 
01282         readfiles, ignored1, ignored2 = select.select([p2.stdout], [], [], 10)
01283 
01284         self.assertTrue(readfiles, "The child hung")
01285         self.assertEqual(p2.stdout.read(), data)
01286 
01287         p1.stdout.close()
01288         p2.stdout.close()

Here is the call graph for this function:

Definition at line 605 of file test_subprocess.py.

00605 
00606     def test_poll(self):
00607         p = subprocess.Popen([sys.executable,
00608                           "-c", "import time; time.sleep(1)"])
00609         count = 0
00610         while p.poll() is None:
00611             time.sleep(0.1)
00612             count += 1
00613         # We expect that the poll loop probably went around about 10 times,
00614         # but, based on system scheduling we can't control, it's possible
00615         # poll() never returned None.  It "should be" very rare that it
00616         # didn't go around at least twice.
00617         self.assertGreaterEqual(count, 2)
00618         # Subsequent invocations should just return the returncode
00619         self.assertEqual(p.poll(), 0)
00620 

Here is the call graph for this function:

Definition at line 829 of file test_subprocess.py.

00829 
00830     def test_preexec(self):
00831         # DISCLAIMER: Setting environment variables is *not* a good use
00832         # of a preexec_fn.  This is merely a test.
00833         p = subprocess.Popen([sys.executable, "-c",
00834                               'import sys,os;'
00835                               'sys.stdout.write(os.getenv("FRUIT"))'],
00836                              stdout=subprocess.PIPE,
00837                              preexec_fn=lambda: os.putenv("FRUIT", "apple"))
00838         self.addCleanup(p.stdout.close)
00839         self.assertEqual(p.stdout.read(), b"apple")

Here is the call graph for this function:

Definition at line 840 of file test_subprocess.py.

00840 
00841     def test_preexec_exception(self):
00842         def raise_it():
00843             raise ValueError("What if two swallows carried a coconut?")
00844         try:
00845             p = subprocess.Popen([sys.executable, "-c", ""],
00846                                  preexec_fn=raise_it)
00847         except RuntimeError as e:
00848             self.assertTrue(
00849                     subprocess._posixsubprocess,
00850                     "Expected a ValueError from the preexec_fn")
00851         except ValueError as e:
00852             self.assertIn("coconut", e.args[0])
00853         else:
00854             self.fail("Exception raised by preexec_fn did not make it "
00855                       "to the parent process.")

Here is the call graph for this function:

Definition at line 857 of file test_subprocess.py.

00857 
00858     def test_preexec_gc_module_failure(self):
00859         # This tests the code that disables garbage collection if the child
00860         # process will execute any Python.
00861         def raise_runtime_error():
00862             raise RuntimeError("this shouldn't escape")
00863         enabled = gc.isenabled()
00864         orig_gc_disable = gc.disable
00865         orig_gc_isenabled = gc.isenabled
00866         try:
00867             gc.disable()
00868             self.assertFalse(gc.isenabled())
00869             subprocess.call([sys.executable, '-c', ''],
00870                             preexec_fn=lambda: None)
00871             self.assertFalse(gc.isenabled(),
00872                              "Popen enabled gc when it shouldn't.")
00873 
00874             gc.enable()
00875             self.assertTrue(gc.isenabled())
00876             subprocess.call([sys.executable, '-c', ''],
00877                             preexec_fn=lambda: None)
00878             self.assertTrue(gc.isenabled(), "Popen left gc disabled.")
00879 
00880             gc.disable = raise_runtime_error
00881             self.assertRaises(RuntimeError, subprocess.Popen,
00882                               [sys.executable, '-c', ''],
00883                               preexec_fn=lambda: None)
00884 
00885             del gc.isenabled  # force an AttributeError
00886             self.assertRaises(AttributeError, subprocess.Popen,
00887                               [sys.executable, '-c', ''],
00888                               preexec_fn=lambda: None)
00889         finally:
00890             gc.disable = orig_gc_disable
00891             gc.isenabled = orig_gc_isenabled
00892             if not enabled:
00893                 gc.disable()

Here is the call graph for this function:

Definition at line 1061 of file test_subprocess.py.

01061 
01062     def test_remapping_std_fds(self):
01063         # open up some temporary files
01064         temps = [mkstemp() for i in range(3)]
01065         try:
01066             temp_fds = [fd for fd, fname in temps]
01067 
01068             # unlink the files -- we won't need to reopen them
01069             for fd, fname in temps:
01070                 os.unlink(fname)
01071 
01072             # write some data to what will become stdin, and rewind
01073             os.write(temp_fds[1], b"STDIN")
01074             os.lseek(temp_fds[1], 0, 0)
01075 
01076             # move the standard file descriptors out of the way
01077             saved_fds = [os.dup(fd) for fd in range(3)]
01078             try:
01079                 # duplicate the file objects over the standard fd's
01080                 for fd, temp_fd in enumerate(temp_fds):
01081                     os.dup2(temp_fd, fd)
01082 
01083                 # now use those files in the "wrong" order, so that subprocess
01084                 # has to rearrange them in the child
01085                 p = subprocess.Popen([sys.executable, "-c",
01086                     'import sys; got = sys.stdin.read();'
01087                     'sys.stdout.write("got %s"%got); sys.stderr.write("err")'],
01088                     stdin=temp_fds[1],
01089                     stdout=temp_fds[2],
01090                     stderr=temp_fds[0])
01091                 p.wait()
01092             finally:
01093                 # restore the original fd's underneath sys.stdin, etc.
01094                 for std, saved in enumerate(saved_fds):
01095                     os.dup2(saved, std)
01096                     os.close(saved)
01097 
01098             for fd in temp_fds:
01099                 os.lseek(fd, 0, 0)
01100 
01101             out = os.read(temp_fds[2], 1024)
01102             err = support.strip_python_stderr(os.read(temp_fds[0], 1024))
01103             self.assertEqual(out, b"got STDIN")
01104             self.assertEqual(err, b"err")
01105 
01106         finally:
01107             for fd in temp_fds:
01108                 os.close(fd)

Here is the call graph for this function:

Definition at line 797 of file test_subprocess.py.

00797 
00798     def test_restore_signals(self):
00799         # Code coverage for both values of restore_signals to make sure it
00800         # at least does not blow up.
00801         # A test for behavior would be complex.  Contributions welcome.
00802         subprocess.call([sys.executable, "-c", ""], restore_signals=True)
00803         subprocess.call([sys.executable, "-c", ""], restore_signals=False)

Here is the call graph for this function:

Definition at line 821 of file test_subprocess.py.

00821 
00822     def test_run_abort(self):
00823         # returncode handles signal termination
00824         with _SuppressCoreFiles():
00825             p = subprocess.Popen([sys.executable, "-c",
00826                                   'import os; os.abort()'])
00827             p.wait()
00828         self.assertEqual(-p.returncode, signal.SIGABRT)

Here is the call graph for this function:

Definition at line 1380 of file test_subprocess.py.

01380 
01381     def test_select_unbuffered(self):
01382         # Issue #11459: bufsize=0 should really set the pipes as
01383         # unbuffered (and therefore let select() work properly).
01384         select = support.import_module("select")
01385         p = subprocess.Popen([sys.executable, "-c",
01386                               'import sys;'
01387                               'sys.stdout.write("apple")'],
01388                              stdout=subprocess.PIPE,
01389                              bufsize=0)
01390         f = p.stdout
01391         self.addCleanup(f.close)
01392         try:
01393             self.assertEqual(f.read(4), b"appl")
01394             self.assertIn(f, select.select([f], [], [], 0.0)[0])
01395         finally:
01396             p.wait()
01397 
01398 
@unittest.skipUnless(mswindows, "Windows specific tests")

Here is the call graph for this function:

Definition at line 991 of file test_subprocess.py.

00991 
00992     def test_send_signal(self):
00993         p = self._kill_process('send_signal', signal.SIGINT)
00994         _, stderr = p.communicate()
00995         self.assertIn(b'KeyboardInterrupt', stderr)
00996         self.assertNotEqual(p.wait(), 0)

Here is the call graph for this function:

Definition at line 919 of file test_subprocess.py.

00919 
00920     def test_shell_sequence(self):
00921         # Run command through the shell (sequence)
00922         newenv = os.environ.copy()
00923         newenv["FRUIT"] = "apple"
00924         p = subprocess.Popen(["echo $FRUIT"], shell=1,
00925                              stdout=subprocess.PIPE,
00926                              env=newenv)
00927         self.addCleanup(p.stdout.close)
00928         self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple")

Here is the call graph for this function:

Definition at line 929 of file test_subprocess.py.

00929 
00930     def test_shell_string(self):
00931         # Run command through the shell (string)
00932         newenv = os.environ.copy()
00933         newenv["FRUIT"] = "apple"
00934         p = subprocess.Popen("echo $FRUIT", shell=1,
00935                              stdout=subprocess.PIPE,
00936                              env=newenv)
00937         self.addCleanup(p.stdout.close)
00938         self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple")

Here is the call graph for this function:

Definition at line 952 of file test_subprocess.py.

00952 
00953     def test_specific_shell(self):
00954         # Issue #9265: Incorrect name passed as arg[0].
00955         shells = []
00956         for prefix in ['/bin', '/usr/bin/', '/usr/local/bin']:
00957             for name in ['bash', 'ksh']:
00958                 sh = os.path.join(prefix, name)
00959                 if os.path.isfile(sh):
00960                     shells.append(sh)
00961         if not shells: # Will probably work for any shell but csh.
00962             self.skipTest("bash or ksh required for this test")
00963         sh = '/bin/sh'
00964         if os.path.isfile(sh) and not os.path.islink(sh):
00965             # Test will fail if /bin/sh is a symlink to csh.
00966             shells.append(sh)
00967         for sh in shells:
00968             p = subprocess.Popen("echo $0", executable=sh, shell=True,
00969                                  stdout=subprocess.PIPE)
00970             self.addCleanup(p.stdout.close)
00971             self.assertEqual(p.stdout.read().strip(), bytes(sh, 'ascii'))

Here is the call graph for this function:

Definition at line 804 of file test_subprocess.py.

00804 
00805     def test_start_new_session(self):
00806         # For code coverage of calling setsid().  We don't care if we get an
00807         # EPERM error from it depending on the test execution environment, that
00808         # still indicates that it was called.
00809         try:
00810             output = subprocess.check_output(
00811                     [sys.executable, "-c",
00812                      "import os; print(os.getpgid(os.getpid()))"],
00813                     start_new_session=True)
00814         except OSError as e:
00815             if e.errno != errno.EPERM:
00816                 raise
00817         else:
00818             parent_pgid = os.getpgid(os.getpid())
00819             child_pgid = int(output)
00820             self.assertNotEqual(parent_pgid, child_pgid)

Here is the call graph for this function:

Definition at line 257 of file test_subprocess.py.

00257 
00258     def test_stderr_filedes(self):
00259         # stderr is set to open file descriptor
00260         tf = tempfile.TemporaryFile()
00261         self.addCleanup(tf.close)
00262         d = tf.fileno()
00263         p = subprocess.Popen([sys.executable, "-c",
00264                           'import sys; sys.stderr.write("strawberry")'],
00265                          stderr=d)
00266         p.wait()
00267         os.lseek(d, 0, 0)
00268         self.assertStderrEqual(os.read(d, 1024), b"strawberry")

Here is the call graph for this function:

Definition at line 269 of file test_subprocess.py.

00269 
00270     def test_stderr_fileobj(self):
00271         # stderr is set to open file object
00272         tf = tempfile.TemporaryFile()
00273         self.addCleanup(tf.close)
00274         p = subprocess.Popen([sys.executable, "-c",
00275                           'import sys; sys.stderr.write("strawberry")'],
00276                          stderr=tf)
00277         p.wait()
00278         tf.seek(0)
00279         self.assertStderrEqual(tf.read(), b"strawberry")

Here is the call graph for this function:

Definition at line 155 of file test_subprocess.py.

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

Here is the call graph for this function:

Definition at line 249 of file test_subprocess.py.

00249 
00250     def test_stderr_pipe(self):
00251         # stderr redirection
00252         p = subprocess.Popen([sys.executable, "-c",
00253                           'import sys; sys.stderr.write("strawberry")'],
00254                          stderr=subprocess.PIPE)
00255         self.addCleanup(p.stderr.close)
00256         self.assertStderrEqual(p.stderr.read(), b"strawberry")

Here is the call graph for this function:

Definition at line 1363 of file test_subprocess.py.

01363 
01364     def test_stderr_stdin_are_single_inout_fd(self):
01365         with io.open(os.devnull, "r+") as inout:
01366             p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
01367                                  stderr=inout, stdin=inout)
01368             p.wait()

Definition at line 193 of file test_subprocess.py.

00193 
00194     def test_stdin_filedes(self):
00195         # stdin is set to open file descriptor
00196         tf = tempfile.TemporaryFile()
00197         self.addCleanup(tf.close)
00198         d = tf.fileno()
00199         os.write(d, b"pear")
00200         os.lseek(d, 0, 0)
00201         p = subprocess.Popen([sys.executable, "-c",
00202                          'import sys; sys.exit(sys.stdin.read() == "pear")'],
00203                          stdin=d)
00204         p.wait()
00205         self.assertEqual(p.returncode, 1)

Here is the call graph for this function:

Definition at line 206 of file test_subprocess.py.

00206 
00207     def test_stdin_fileobj(self):
00208         # stdin is set to open file object
00209         tf = tempfile.TemporaryFile()
00210         self.addCleanup(tf.close)
00211         tf.write(b"pear")
00212         tf.seek(0)
00213         p = subprocess.Popen([sys.executable, "-c",
00214                          'import sys; sys.exit(sys.stdin.read() == "pear")'],
00215                          stdin=tf)
00216         p.wait()
00217         self.assertEqual(p.returncode, 1)

Here is the call graph for this function:

Definition at line 134 of file test_subprocess.py.

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

Here is the call graph for this function:

Definition at line 183 of file test_subprocess.py.

00183 
00184     def test_stdin_pipe(self):
00185         # stdin redirection
00186         p = subprocess.Popen([sys.executable, "-c",
00187                          'import sys; sys.exit(sys.stdin.read() == "pear")'],
00188                         stdin=subprocess.PIPE)
00189         p.stdin.write(b"pear")
00190         p.stdin.close()
00191         p.wait()
00192         self.assertEqual(p.returncode, 1)

Here is the call graph for this function:

Definition at line 226 of file test_subprocess.py.

00226 
00227     def test_stdout_filedes(self):
00228         # stdout is set to open file descriptor
00229         tf = tempfile.TemporaryFile()
00230         self.addCleanup(tf.close)
00231         d = tf.fileno()
00232         p = subprocess.Popen([sys.executable, "-c",
00233                           'import sys; sys.stdout.write("orange")'],
00234                          stdout=d)
00235         p.wait()
00236         os.lseek(d, 0, 0)
00237         self.assertEqual(os.read(d, 1024), b"orange")

Here is the call graph for this function:

Definition at line 307 of file test_subprocess.py.

00307 
00308     def test_stdout_filedes_of_stdout(self):
00309         # stdout is set to 1 (#1531862).
00310         cmd = r"import sys, os; sys.exit(os.write(sys.stdout.fileno(), b'.\n'))"
00311         rc = subprocess.call([sys.executable, "-c", cmd], stdout=1)
00312         self.assertEqual(rc, 2)

Here is the call graph for this function:

Definition at line 238 of file test_subprocess.py.

00238 
00239     def test_stdout_fileobj(self):
00240         # stdout is set to open file object
00241         tf = tempfile.TemporaryFile()
00242         self.addCleanup(tf.close)
00243         p = subprocess.Popen([sys.executable, "-c",
00244                           'import sys; sys.stdout.write("orange")'],
00245                          stdout=tf)
00246         p.wait()
00247         tf.seek(0)
00248         self.assertEqual(tf.read(), b"orange")

Here is the call graph for this function:

Definition at line 143 of file test_subprocess.py.

00143 
00144     def test_stdout_none(self):
00145         # .stdout is None when not redirected
00146         p = subprocess.Popen([sys.executable, "-c",
00147                              'print("    this bit of output is from a '
00148                              'test of stdout in a different '
00149                              'process ...")'],
00150                              stdin=subprocess.PIPE, stderr=subprocess.PIPE)
00151         self.addCleanup(p.stdin.close)
00152         self.addCleanup(p.stderr.close)
00153         p.wait()
00154         self.assertEqual(p.stdout, None)

Here is the call graph for this function:

Definition at line 218 of file test_subprocess.py.

00218 
00219     def test_stdout_pipe(self):
00220         # stdout redirection
00221         p = subprocess.Popen([sys.executable, "-c",
00222                           'import sys; sys.stdout.write("orange")'],
00223                          stdout=subprocess.PIPE)
00224         self.addCleanup(p.stdout.close)
00225         self.assertEqual(p.stdout.read(), b"orange")

Here is the call graph for this function:

Definition at line 1357 of file test_subprocess.py.

01357 
01358     def test_stdout_stderr_are_single_inout_fd(self):
01359         with io.open(os.devnull, "r+") as inout:
01360             p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
01361                                  stdout=inout, stderr=inout)
01362             p.wait()

Definition at line 292 of file test_subprocess.py.

00292 
00293     def test_stdout_stderr_file(self):
00294         # capture stdout and stderr to the same open file
00295         tf = tempfile.TemporaryFile()
00296         self.addCleanup(tf.close)
00297         p = subprocess.Popen([sys.executable, "-c",
00298                               'import sys;'
00299                               'sys.stdout.write("apple");'
00300                               'sys.stdout.flush();'
00301                               'sys.stderr.write("orange")'],
00302                              stdout=tf,
00303                              stderr=tf)
00304         p.wait()
00305         tf.seek(0)
00306         self.assertStderrEqual(tf.read(), b"appleorange")

Here is the call graph for this function:

Definition at line 280 of file test_subprocess.py.

00280 
00281     def test_stdout_stderr_pipe(self):
00282         # capture stdout and stderr to the same pipe
00283         p = subprocess.Popen([sys.executable, "-c",
00284                               'import sys;'
00285                               'sys.stdout.write("apple");'
00286                               'sys.stdout.flush();'
00287                               'sys.stderr.write("orange")'],
00288                              stdout=subprocess.PIPE,
00289                              stderr=subprocess.STDOUT)
00290         self.addCleanup(p.stdout.close)
00291         self.assertStderrEqual(p.stdout.read(), b"appleorange")

Here is the call graph for this function:

Definition at line 1351 of file test_subprocess.py.

01351 
01352     def test_stdout_stdin_are_single_inout_fd(self):
01353         with io.open(os.devnull, "r+") as inout:
01354             p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
01355                                  stdout=inout, stdin=inout)
01356             p.wait()

Definition at line 1167 of file test_subprocess.py.

01167 
01168     def test_surrogates_error_message(self):
01169         def prepare():
01170             raise ValueError("surrogate:\uDCff")
01171 
01172         try:
01173             subprocess.call(
01174                 [sys.executable, "-c", "pass"],
01175                 preexec_fn=prepare)
01176         except ValueError as err:
01177             # Pure Python implementations keeps the message
01178             self.assertIsNone(subprocess._posixsubprocess)
01179             self.assertEqual(str(err), "surrogate:\uDCff")
01180         except RuntimeError as err:
01181             # _posixsubprocess uses a default message
01182             self.assertIsNotNone(subprocess._posixsubprocess)
01183             self.assertEqual(str(err), "Exception occurred in preexec_fn.")
01184         else:
01185             self.fail("Expected ValueError or RuntimeError")

Here is the call graph for this function:

Definition at line 1159 of file test_subprocess.py.

01159 
01160     def test_swap_fds(self):
01161         self.check_swap_fds(0, 1, 2)
01162         self.check_swap_fds(0, 2, 1)
01163         self.check_swap_fds(1, 0, 2)
01164         self.check_swap_fds(1, 2, 0)
01165         self.check_swap_fds(2, 0, 1)
01166         self.check_swap_fds(2, 1, 0)

Here is the call graph for this function:

Definition at line 1003 of file test_subprocess.py.

01003 
01004     def test_terminate(self):
01005         p = self._kill_process('terminate')
01006         _, stderr = p.communicate()
01007         self.assertStderrEqual(stderr, b'')
01008         self.assertEqual(p.wait(), -signal.SIGTERM)

Here is the call graph for this function:

Definition at line 1186 of file test_subprocess.py.

01186 
01187     def test_undecodable_env(self):
01188         for key, value in (('test', 'abc\uDCFF'), ('test\uDCFF', '42')):
01189             # test str with surrogates
01190             script = "import os; print(ascii(os.getenv(%s)))" % repr(key)
01191             env = os.environ.copy()
01192             env[key] = value
01193             # Use C locale to get ascii for the locale encoding to force
01194             # surrogate-escaping of \xFF in the child process; otherwise it can
01195             # be decoded as-is if the default locale is latin-1.
01196             env['LC_ALL'] = 'C'
01197             stdout = subprocess.check_output(
01198                 [sys.executable, "-c", script],
01199                 env=env)
01200             stdout = stdout.rstrip(b'\n\r')
01201             self.assertEqual(stdout.decode('ascii'), ascii(value))
01202 
01203             # test bytes
01204             key = key.encode("ascii", "surrogateescape")
01205             value = value.encode("ascii", "surrogateescape")
01206             script = "import os; print(ascii(os.getenvb(%s)))" % repr(key)
01207             env = os.environ.copy()
01208             env[key] = value
01209             stdout = subprocess.check_output(
01210                 [sys.executable, "-c", script],
01211                 env=env)
01212             stdout = stdout.rstrip(b'\n\r')
01213             self.assertEqual(stdout.decode('ascii'), ascii(value))

Here is the call graph for this function:

Definition at line 474 of file test_subprocess.py.

00474 
00475     def test_universal_newlines(self):
00476         p = subprocess.Popen([sys.executable, "-c",
00477                               'import sys,os;' + SETBINARY +
00478                               'sys.stdout.write(sys.stdin.readline());'
00479                               'sys.stdout.flush();'
00480                               'sys.stdout.write("line2\\n");'
00481                               'sys.stdout.flush();'
00482                               'sys.stdout.write(sys.stdin.read());'
00483                               'sys.stdout.flush();'
00484                               'sys.stdout.write("line4\\n");'
00485                               'sys.stdout.flush();'
00486                               'sys.stdout.write("line5\\r\\n");'
00487                               'sys.stdout.flush();'
00488                               'sys.stdout.write("line6\\r");'
00489                               'sys.stdout.flush();'
00490                               'sys.stdout.write("\\nline7");'
00491                               'sys.stdout.flush();'
00492                               'sys.stdout.write("\\nline8");'],
00493                              stdin=subprocess.PIPE,
00494                              stdout=subprocess.PIPE,
00495                              universal_newlines=1)
00496         p.stdin.write("line1\n")
00497         self.assertEqual(p.stdout.readline(), "line1\n")
00498         p.stdin.write("line3\n")
00499         p.stdin.close()
00500         self.addCleanup(p.stdout.close)
00501         self.assertEqual(p.stdout.readline(),
00502                          "line2\n")
00503         self.assertEqual(p.stdout.read(6),
00504                          "line3\n")
00505         self.assertEqual(p.stdout.read(),
00506                          "line4\nline5\nline6\nline7\nline8")

Here is the call graph for this function:

Definition at line 507 of file test_subprocess.py.

00507 
00508     def test_universal_newlines_communicate(self):
00509         # universal newlines through communicate()
00510         p = subprocess.Popen([sys.executable, "-c",
00511                               'import sys,os;' + SETBINARY +
00512                               'sys.stdout.write("line2\\n");'
00513                               'sys.stdout.flush();'
00514                               'sys.stdout.write("line4\\n");'
00515                               'sys.stdout.flush();'
00516                               'sys.stdout.write("line5\\r\\n");'
00517                               'sys.stdout.flush();'
00518                               'sys.stdout.write("line6\\r");'
00519                               'sys.stdout.flush();'
00520                               'sys.stdout.write("\\nline7");'
00521                               'sys.stdout.flush();'
00522                               'sys.stdout.write("\\nline8");'],
00523                              stderr=subprocess.PIPE,
00524                              stdout=subprocess.PIPE,
00525                              universal_newlines=1)
00526         self.addCleanup(p.stdout.close)
00527         self.addCleanup(p.stderr.close)
00528         # BUG: can't give a non-empty stdin because it breaks both the
00529         # select- and poll-based communicate() implementations.
00530         (stdout, stderr) = p.communicate()
00531         self.assertEqual(stdout,
00532                          "line2\nline4\nline5\nline6\nline7\nline8")

Here is the call graph for this function:

Definition at line 533 of file test_subprocess.py.

00533 
00534     def test_universal_newlines_communicate_stdin(self):
00535         # universal newlines through communicate(), with only stdin
00536         p = subprocess.Popen([sys.executable, "-c",
00537                               'import sys,os;' + SETBINARY + '''\nif True:
00538                                   s = sys.stdin.readline()
00539                                   assert s == "line1\\n", repr(s)
00540                                   s = sys.stdin.read()
00541                                   assert s == "line3\\n", repr(s)
00542                               '''],
00543                              stdin=subprocess.PIPE,
00544                              universal_newlines=1)
00545         (stdout, stderr) = p.communicate("line1\nline3\n")
00546         self.assertEqual(p.returncode, 0)

Here is the call graph for this function:

Definition at line 621 of file test_subprocess.py.

00621 
00622     def test_wait(self):
00623         p = subprocess.Popen([sys.executable,
00624                           "-c", "import time; time.sleep(2)"])
00625         self.assertEqual(p.wait(), 0)
00626         # Subsequent invocations should just return the returncode
00627         self.assertEqual(p.wait(), 0)
00628 

Here is the call graph for this function:

Definition at line 1369 of file test_subprocess.py.

01369 
01370     def test_wait_when_sigchild_ignored(self):
01371         # NOTE: sigchild_ignore.py may not be an effective test on all OSes.
01372         sigchild_ignore = support.findfile("sigchild_ignore.py",
01373                                            subdir="subprocessdata")
01374         p = subprocess.Popen([sys.executable, sigchild_ignore],
01375                              stdout=subprocess.PIPE, stderr=subprocess.PIPE)
01376         stdout, stderr = p.communicate()
01377         self.assertEqual(0, p.returncode, "sigchild_ignore.py exited"
01378                          " non-zero with this error:\n%s" %
01379                          stderr.decode('utf8'))

Here is the call graph for this function:

Definition at line 458 of file test_subprocess.py.

00458 
00459     def test_writes_before_communicate(self):
00460         # stdin.write before communicate()
00461         p = subprocess.Popen([sys.executable, "-c",
00462                               'import sys,os;'
00463                               'sys.stdout.write(sys.stdin.read())'],
00464                              stdin=subprocess.PIPE,
00465                              stdout=subprocess.PIPE,
00466                              stderr=subprocess.PIPE)
00467         self.addCleanup(p.stdout.close)
00468         self.addCleanup(p.stderr.close)
00469         self.addCleanup(p.stdin.close)
00470         p.stdin.write(b"banana")
00471         (stdout, stderr) = p.communicate(b"split")
00472         self.assertEqual(stdout, b"bananasplit")
00473         self.assertStderrEqual(stderr, b"")

Here is the call graph for this function:


The documentation for this class was generated from the following file: