Back to index

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

List of all members.

Public Member Functions

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

Private Member Functions

def _kill_process

Detailed Description

Definition at line 769 of file test_subprocess.py.


Member Function Documentation

def test.test_subprocess.POSIXProcessTestCase._kill_process (   self,
  method,
  args 
) [private]

Definition at line 972 of file test_subprocess.py.

00972 
00973     def _kill_process(self, method, *args):
00974         # Do not inherit file handles from the parent.
00975         # It should fix failures on some platforms.
00976         p = subprocess.Popen([sys.executable, "-c", """if 1:
00977                              import sys, time
00978                              sys.stdout.write('x\\n')
00979                              sys.stdout.flush()
00980                              time.sleep(30)
00981                              """],
00982                              close_fds=True,
00983                              stdin=subprocess.PIPE,
00984                              stdout=subprocess.PIPE,
00985                              stderr=subprocess.PIPE)
00986         # Wait for the interpreter to be completely initialized before
00987         # sending any signal.
00988         p.stdout.read(1)
00989         getattr(p, method)(*args)
00990         return p

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 
)

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


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