1 import unittest 2 from unittest import mock 3 from test import support 4 import subprocess 5 import sys 6 import platform 7 import signal 8 import io 9 import itertools 10 import os 11 import errno 12 import tempfile 13 import time 14 import selectors 15 import sysconfig 16 import select 17 import shutil 18 import threading 19 import gc 20 import textwrap 21 from test.support import FakePath 22 23 try: 24 import ctypes 25 except ImportError: 26 ctypes = None 27 else: 28 import ctypes.util 29 30 try: 31 import _testcapi 32 except ImportError: 33 _testcapi = None 34 35 if support.PGO: 36 raise unittest.SkipTest("test is not helpful for PGO") 37 38 mswindows = (sys.platform == "win32") 39 40 # 41 # Depends on the following external programs: Python 42 # 43 44 if mswindows: 45 SETBINARY = ('import msvcrt; msvcrt.setmode(sys.stdout.fileno(), ' 46 'os.O_BINARY);') 47 else: 48 SETBINARY = '' 49 50 NONEXISTING_CMD = ('nonexisting_i_hope',) 51 # Ignore errors that indicate the command was not found 52 NONEXISTING_ERRORS = (FileNotFoundError, NotADirectoryError, PermissionError) 53 54 55 class BaseTestCase(unittest.TestCase): 56 def setUp(self): 57 # Try to minimize the number of children we have so this test 58 # doesn't crash on some buildbots (Alphas in particular). 59 support.reap_children() 60 61 def tearDown(self): 62 for inst in subprocess._active: 63 inst.wait() 64 subprocess._cleanup() 65 self.assertFalse(subprocess._active, "subprocess._active not empty") 66 self.doCleanups() 67 support.reap_children() 68 69 def assertStderrEqual(self, stderr, expected, msg=None): 70 # In a debug build, stuff like "[6580 refs]" is printed to stderr at 71 # shutdown time. That frustrates tests trying to check stderr produced 72 # from a spawned Python process. 73 actual = support.strip_python_stderr(stderr) 74 # strip_python_stderr also strips whitespace, so we do too. 75 expected = expected.strip() 76 self.assertEqual(actual, expected, msg) 77 78 79 class PopenTestException(Exception): 80 pass 81 82 83 class PopenExecuteChildRaises(subprocess.Popen): 84 """Popen subclass for testing cleanup of subprocess.PIPE filehandles when 85 _execute_child fails. 86 """ 87 def _execute_child(self, *args, **kwargs): 88 raise PopenTestException("Forced Exception for Test") 89 90 91 class ProcessTestCase(BaseTestCase): 92 93 def test_io_buffered_by_default(self): 94 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"], 95 stdin=subprocess.PIPE, stdout=subprocess.PIPE, 96 stderr=subprocess.PIPE) 97 try: 98 self.assertIsInstance(p.stdin, io.BufferedIOBase) 99 self.assertIsInstance(p.stdout, io.BufferedIOBase) 100 self.assertIsInstance(p.stderr, io.BufferedIOBase) 101 finally: 102 p.stdin.close() 103 p.stdout.close() 104 p.stderr.close() 105 p.wait() 106 107 def test_io_unbuffered_works(self): 108 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"], 109 stdin=subprocess.PIPE, stdout=subprocess.PIPE, 110 stderr=subprocess.PIPE, bufsize=0) 111 try: 112 self.assertIsInstance(p.stdin, io.RawIOBase) 113 self.assertIsInstance(p.stdout, io.RawIOBase) 114 self.assertIsInstance(p.stderr, io.RawIOBase) 115 finally: 116 p.stdin.close() 117 p.stdout.close() 118 p.stderr.close() 119 p.wait() 120 121 def test_call_seq(self): 122 # call() function with sequence argument 123 rc = subprocess.call([sys.executable, "-c", 124 "import sys; sys.exit(47)"]) 125 self.assertEqual(rc, 47) 126 127 def test_call_timeout(self): 128 # call() function with timeout argument; we want to test that the child 129 # process gets killed when the timeout expires. If the child isn't 130 # killed, this call will deadlock since subprocess.call waits for the 131 # child. 132 self.assertRaises(subprocess.TimeoutExpired, subprocess.call, 133 [sys.executable, "-c", "while True: pass"], 134 timeout=0.1) 135 136 def test_check_call_zero(self): 137 # check_call() function with zero return code 138 rc = subprocess.check_call([sys.executable, "-c", 139 "import sys; sys.exit(0)"]) 140 self.assertEqual(rc, 0) 141 142 def test_check_call_nonzero(self): 143 # check_call() function with non-zero return code 144 with self.assertRaises(subprocess.CalledProcessError) as c: 145 subprocess.check_call([sys.executable, "-c", 146 "import sys; sys.exit(47)"]) 147 self.assertEqual(c.exception.returncode, 47) 148 149 def test_check_output(self): 150 # check_output() function with zero return code 151 output = subprocess.check_output( 152 [sys.executable, "-c", "print('BDFL')"]) 153 self.assertIn(b'BDFL', output) 154 155 def test_check_output_nonzero(self): 156 # check_call() function with non-zero return code 157 with self.assertRaises(subprocess.CalledProcessError) as c: 158 subprocess.check_output( 159 [sys.executable, "-c", "import sys; sys.exit(5)"]) 160 self.assertEqual(c.exception.returncode, 5) 161 162 def test_check_output_stderr(self): 163 # check_output() function stderr redirected to stdout 164 output = subprocess.check_output( 165 [sys.executable, "-c", "import sys; sys.stderr.write('BDFL')"], 166 stderr=subprocess.STDOUT) 167 self.assertIn(b'BDFL', output) 168 169 def test_check_output_stdin_arg(self): 170 # check_output() can be called with stdin set to a file 171 tf = tempfile.TemporaryFile() 172 self.addCleanup(tf.close) 173 tf.write(b'pear') 174 tf.seek(0) 175 output = subprocess.check_output( 176 [sys.executable, "-c", 177 "import sys; sys.stdout.write(sys.stdin.read().upper())"], 178 stdin=tf) 179 self.assertIn(b'PEAR', output) 180 181 def test_check_output_input_arg(self): 182 # check_output() can be called with input set to a string 183 output = subprocess.check_output( 184 [sys.executable, "-c", 185 "import sys; sys.stdout.write(sys.stdin.read().upper())"], 186 input=b'pear') 187 self.assertIn(b'PEAR', output) 188 189 def test_check_output_stdout_arg(self): 190 # check_output() refuses to accept 'stdout' argument 191 with self.assertRaises(ValueError) as c: 192 output = subprocess.check_output( 193 [sys.executable, "-c", "print('will not be run')"], 194 stdout=sys.stdout) 195 self.fail("Expected ValueError when stdout arg supplied.") 196 self.assertIn('stdout', c.exception.args[0]) 197 198 def test_check_output_stdin_with_input_arg(self): 199 # check_output() refuses to accept 'stdin' with 'input' 200 tf = tempfile.TemporaryFile() 201 self.addCleanup(tf.close) 202 tf.write(b'pear') 203 tf.seek(0) 204 with self.assertRaises(ValueError) as c: 205 output = subprocess.check_output( 206 [sys.executable, "-c", "print('will not be run')"], 207 stdin=tf, input=b'hare') 208 self.fail("Expected ValueError when stdin and input args supplied.") 209 self.assertIn('stdin', c.exception.args[0]) 210 self.assertIn('input', c.exception.args[0]) 211 212 def test_check_output_timeout(self): 213 # check_output() function with timeout arg 214 with self.assertRaises(subprocess.TimeoutExpired) as c: 215 output = subprocess.check_output( 216 [sys.executable, "-c", 217 "import sys, time\n" 218 "sys.stdout.write('BDFL')\n" 219 "sys.stdout.flush()\n" 220 "time.sleep(3600)"], 221 # Some heavily loaded buildbots (sparc Debian 3.x) require 222 # this much time to start and print. 223 timeout=3) 224 self.fail("Expected TimeoutExpired.") 225 self.assertEqual(c.exception.output, b'BDFL') 226 227 def test_call_kwargs(self): 228 # call() function with keyword args 229 newenv = os.environ.copy() 230 newenv["FRUIT"] = "banana" 231 rc = subprocess.call([sys.executable, "-c", 232 'import sys, os;' 233 'sys.exit(os.getenv("FRUIT")=="banana")'], 234 env=newenv) 235 self.assertEqual(rc, 1) 236 237 def test_invalid_args(self): 238 # Popen() called with invalid arguments should raise TypeError 239 # but Popen.__del__ should not complain (issue #12085) 240 with support.captured_stderr() as s: 241 self.assertRaises(TypeError, subprocess.Popen, invalid_arg_name=1) 242 argcount = subprocess.Popen.__init__.__code__.co_argcount 243 too_many_args = [0] * (argcount + 1) 244 self.assertRaises(TypeError, subprocess.Popen, *too_many_args) 245 self.assertEqual(s.getvalue(), '') 246 247 def test_stdin_none(self): 248 # .stdin is None when not redirected 249 p = subprocess.Popen([sys.executable, "-c", 'print("banana")'], 250 stdout=subprocess.PIPE, stderr=subprocess.PIPE) 251 self.addCleanup(p.stdout.close) 252 self.addCleanup(p.stderr.close) 253 p.wait() 254 self.assertEqual(p.stdin, None) 255 256 def test_stdout_none(self): 257 # .stdout is None when not redirected, and the child's stdout will 258 # be inherited from the parent. In order to test this we run a 259 # subprocess in a subprocess: 260 # this_test 261 # \-- subprocess created by this test (parent) 262 # \-- subprocess created by the parent subprocess (child) 263 # The parent doesn't specify stdout, so the child will use the 264 # parent's stdout. This test checks that the message printed by the 265 # child goes to the parent stdout. The parent also checks that the 266 # child's stdout is None. See #11963. 267 code = ('import sys; from subprocess import Popen, PIPE;' 268 'p = Popen([sys.executable, "-c", "print(\'test_stdout_none\')"],' 269 ' stdin=PIPE, stderr=PIPE);' 270 'p.wait(); assert p.stdout is None;') 271 p = subprocess.Popen([sys.executable, "-c", code], 272 stdout=subprocess.PIPE, stderr=subprocess.PIPE) 273 self.addCleanup(p.stdout.close) 274 self.addCleanup(p.stderr.close) 275 out, err = p.communicate() 276 self.assertEqual(p.returncode, 0, err) 277 self.assertEqual(out.rstrip(), b'test_stdout_none') 278 279 def test_stderr_none(self): 280 # .stderr is None when not redirected 281 p = subprocess.Popen([sys.executable, "-c", 'print("banana")'], 282 stdin=subprocess.PIPE, stdout=subprocess.PIPE) 283 self.addCleanup(p.stdout.close) 284 self.addCleanup(p.stdin.close) 285 p.wait() 286 self.assertEqual(p.stderr, None) 287 288 def _assert_python(self, pre_args, **kwargs): 289 # We include sys.exit() to prevent the test runner from hanging 290 # whenever python is found. 291 args = pre_args + ["import sys; sys.exit(47)"] 292 p = subprocess.Popen(args, **kwargs) 293 p.wait() 294 self.assertEqual(47, p.returncode) 295 296 def test_executable(self): 297 # Check that the executable argument works. 298 # 299 # On Unix (non-Mac and non-Windows), Python looks at args[0] to 300 # determine where its standard library is, so we need the directory 301 # of args[0] to be valid for the Popen() call to Python to succeed. 302 # See also issue #16170 and issue #7774. 303 doesnotexist = os.path.join(os.path.dirname(sys.executable), 304 "doesnotexist") 305 self._assert_python([doesnotexist, "-c"], executable=sys.executable) 306 307 def test_executable_takes_precedence(self): 308 # Check that the executable argument takes precedence over args[0]. 309 # 310 # Verify first that the call succeeds without the executable arg. 311 pre_args = [sys.executable, "-c"] 312 self._assert_python(pre_args) 313 self.assertRaises(NONEXISTING_ERRORS, 314 self._assert_python, pre_args, 315 executable=NONEXISTING_CMD[0]) 316 317 @unittest.skipIf(mswindows, "executable argument replaces shell") 318 def test_executable_replaces_shell(self): 319 # Check that the executable argument replaces the default shell 320 # when shell=True. 321 self._assert_python([], executable=sys.executable, shell=True) 322 323 # For use in the test_cwd* tests below. 324 def _normalize_cwd(self, cwd): 325 # Normalize an expected cwd (for Tru64 support). 326 # We can't use os.path.realpath since it doesn't expand Tru64 {memb} 327 # strings. See bug #1063571. 328 with support.change_cwd(cwd): 329 return os.getcwd() 330 331 # For use in the test_cwd* tests below. 332 def _split_python_path(self): 333 # Return normalized (python_dir, python_base). 334 python_path = os.path.realpath(sys.executable) 335 return os.path.split(python_path) 336 337 # For use in the test_cwd* tests below. 338 def _assert_cwd(self, expected_cwd, python_arg, **kwargs): 339 # Invoke Python via Popen, and assert that (1) the call succeeds, 340 # and that (2) the current working directory of the child process 341 # matches *expected_cwd*. 342 p = subprocess.Popen([python_arg, "-c", 343 "import os, sys; " 344 "sys.stdout.write(os.getcwd()); " 345 "sys.exit(47)"], 346 stdout=subprocess.PIPE, 347 **kwargs) 348 self.addCleanup(p.stdout.close) 349 p.wait() 350 self.assertEqual(47, p.returncode) 351 normcase = os.path.normcase 352 self.assertEqual(normcase(expected_cwd), 353 normcase(p.stdout.read().decode("utf-8"))) 354 355 def test_cwd(self): 356 # Check that cwd changes the cwd for the child process. 357 temp_dir = tempfile.gettempdir() 358 temp_dir = self._normalize_cwd(temp_dir) 359 self._assert_cwd(temp_dir, sys.executable, cwd=temp_dir) 360 361 def test_cwd_with_pathlike(self): 362 temp_dir = tempfile.gettempdir() 363 temp_dir = self._normalize_cwd(temp_dir) 364 self._assert_cwd(temp_dir, sys.executable, cwd=FakePath(temp_dir)) 365 366 @unittest.skipIf(mswindows, "pending resolution of issue #15533") 367 def test_cwd_with_relative_arg(self): 368 # Check that Popen looks for args[0] relative to cwd if args[0] 369 # is relative. 370 python_dir, python_base = self._split_python_path() 371 rel_python = os.path.join(os.curdir, python_base) 372 with support.temp_cwd() as wrong_dir: 373 # Before calling with the correct cwd, confirm that the call fails 374 # without cwd and with the wrong cwd. 375 self.assertRaises(FileNotFoundError, subprocess.Popen, 376 [rel_python]) 377 self.assertRaises(FileNotFoundError, subprocess.Popen, 378 [rel_python], cwd=wrong_dir) 379 python_dir = self._normalize_cwd(python_dir) 380 self._assert_cwd(python_dir, rel_python, cwd=python_dir) 381 382 @unittest.skipIf(mswindows, "pending resolution of issue #15533") 383 def test_cwd_with_relative_executable(self): 384 # Check that Popen looks for executable relative to cwd if executable 385 # is relative (and that executable takes precedence over args[0]). 386 python_dir, python_base = self._split_python_path() 387 rel_python = os.path.join(os.curdir, python_base) 388 doesntexist = "somethingyoudonthave" 389 with support.temp_cwd() as wrong_dir: 390 # Before calling with the correct cwd, confirm that the call fails 391 # without cwd and with the wrong cwd. 392 self.assertRaises(FileNotFoundError, subprocess.Popen, 393 [doesntexist], executable=rel_python) 394 self.assertRaises(FileNotFoundError, subprocess.Popen, 395 [doesntexist], executable=rel_python, 396 cwd=wrong_dir) 397 python_dir = self._normalize_cwd(python_dir) 398 self._assert_cwd(python_dir, doesntexist, executable=rel_python, 399 cwd=python_dir) 400 401 def test_cwd_with_absolute_arg(self): 402 # Check that Popen can find the executable when the cwd is wrong 403 # if args[0] is an absolute path. 404 python_dir, python_base = self._split_python_path() 405 abs_python = os.path.join(python_dir, python_base) 406 rel_python = os.path.join(os.curdir, python_base) 407 with support.temp_dir() as wrong_dir: 408 # Before calling with an absolute path, confirm that using a 409 # relative path fails. 410 self.assertRaises(FileNotFoundError, subprocess.Popen, 411 [rel_python], cwd=wrong_dir) 412 wrong_dir = self._normalize_cwd(wrong_dir) 413 self._assert_cwd(wrong_dir, abs_python, cwd=wrong_dir) 414 415 @unittest.skipIf(sys.base_prefix != sys.prefix, 416 'Test is not venv-compatible') 417 def test_executable_with_cwd(self): 418 python_dir, python_base = self._split_python_path() 419 python_dir = self._normalize_cwd(python_dir) 420 self._assert_cwd(python_dir, "somethingyoudonthave", 421 executable=sys.executable, cwd=python_dir) 422 423 @unittest.skipIf(sys.base_prefix != sys.prefix, 424 'Test is not venv-compatible') 425 @unittest.skipIf(sysconfig.is_python_build(), 426 "need an installed Python. See #7774") 427 def test_executable_without_cwd(self): 428 # For a normal installation, it should work without 'cwd' 429 # argument. For test runs in the build directory, see #7774. 430 self._assert_cwd(os.getcwd(), "somethingyoudonthave", 431 executable=sys.executable) 432 433 def test_stdin_pipe(self): 434 # stdin redirection 435 p = subprocess.Popen([sys.executable, "-c", 436 'import sys; sys.exit(sys.stdin.read() == "pear")'], 437 stdin=subprocess.PIPE) 438 p.stdin.write(b"pear") 439 p.stdin.close() 440 p.wait() 441 self.assertEqual(p.returncode, 1) 442 443 def test_stdin_filedes(self): 444 # stdin is set to open file descriptor 445 tf = tempfile.TemporaryFile() 446 self.addCleanup(tf.close) 447 d = tf.fileno() 448 os.write(d, b"pear") 449 os.lseek(d, 0, 0) 450 p = subprocess.Popen([sys.executable, "-c", 451 'import sys; sys.exit(sys.stdin.read() == "pear")'], 452 stdin=d) 453 p.wait() 454 self.assertEqual(p.returncode, 1) 455 456 def test_stdin_fileobj(self): 457 # stdin is set to open file object 458 tf = tempfile.TemporaryFile() 459 self.addCleanup(tf.close) 460 tf.write(b"pear") 461 tf.seek(0) 462 p = subprocess.Popen([sys.executable, "-c", 463 'import sys; sys.exit(sys.stdin.read() == "pear")'], 464 stdin=tf) 465 p.wait() 466 self.assertEqual(p.returncode, 1) 467 468 def test_stdout_pipe(self): 469 # stdout redirection 470 p = subprocess.Popen([sys.executable, "-c", 471 'import sys; sys.stdout.write("orange")'], 472 stdout=subprocess.PIPE) 473 with p: 474 self.assertEqual(p.stdout.read(), b"orange") 475 476 def test_stdout_filedes(self): 477 # stdout is set to open file descriptor 478 tf = tempfile.TemporaryFile() 479 self.addCleanup(tf.close) 480 d = tf.fileno() 481 p = subprocess.Popen([sys.executable, "-c", 482 'import sys; sys.stdout.write("orange")'], 483 stdout=d) 484 p.wait() 485 os.lseek(d, 0, 0) 486 self.assertEqual(os.read(d, 1024), b"orange") 487 488 def test_stdout_fileobj(self): 489 # stdout is set to open file object 490 tf = tempfile.TemporaryFile() 491 self.addCleanup(tf.close) 492 p = subprocess.Popen([sys.executable, "-c", 493 'import sys; sys.stdout.write("orange")'], 494 stdout=tf) 495 p.wait() 496 tf.seek(0) 497 self.assertEqual(tf.read(), b"orange") 498 499 def test_stderr_pipe(self): 500 # stderr redirection 501 p = subprocess.Popen([sys.executable, "-c", 502 'import sys; sys.stderr.write("strawberry")'], 503 stderr=subprocess.PIPE) 504 with p: 505 self.assertStderrEqual(p.stderr.read(), b"strawberry") 506 507 def test_stderr_filedes(self): 508 # stderr is set to open file descriptor 509 tf = tempfile.TemporaryFile() 510 self.addCleanup(tf.close) 511 d = tf.fileno() 512 p = subprocess.Popen([sys.executable, "-c", 513 'import sys; sys.stderr.write("strawberry")'], 514 stderr=d) 515 p.wait() 516 os.lseek(d, 0, 0) 517 self.assertStderrEqual(os.read(d, 1024), b"strawberry") 518 519 def test_stderr_fileobj(self): 520 # stderr is set to open file object 521 tf = tempfile.TemporaryFile() 522 self.addCleanup(tf.close) 523 p = subprocess.Popen([sys.executable, "-c", 524 'import sys; sys.stderr.write("strawberry")'], 525 stderr=tf) 526 p.wait() 527 tf.seek(0) 528 self.assertStderrEqual(tf.read(), b"strawberry") 529 530 def test_stderr_redirect_with_no_stdout_redirect(self): 531 # test stderr=STDOUT while stdout=None (not set) 532 533 # - grandchild prints to stderr 534 # - child redirects grandchild's stderr to its stdout 535 # - the parent should get grandchild's stderr in child's stdout 536 p = subprocess.Popen([sys.executable, "-c", 537 'import sys, subprocess;' 538 'rc = subprocess.call([sys.executable, "-c",' 539 ' "import sys;"' 540 ' "sys.stderr.write(\'42\')"],' 541 ' stderr=subprocess.STDOUT);' 542 'sys.exit(rc)'], 543 stdout=subprocess.PIPE, 544 stderr=subprocess.PIPE) 545 stdout, stderr = p.communicate() 546 #NOTE: stdout should get stderr from grandchild 547 self.assertStderrEqual(stdout, b'42') 548 self.assertStderrEqual(stderr, b'') # should be empty 549 self.assertEqual(p.returncode, 0) 550 551 def test_stdout_stderr_pipe(self): 552 # capture stdout and stderr to the same pipe 553 p = subprocess.Popen([sys.executable, "-c", 554 'import sys;' 555 'sys.stdout.write("apple");' 556 'sys.stdout.flush();' 557 'sys.stderr.write("orange")'], 558 stdout=subprocess.PIPE, 559 stderr=subprocess.STDOUT) 560 with p: 561 self.assertStderrEqual(p.stdout.read(), b"appleorange") 562 563 def test_stdout_stderr_file(self): 564 # capture stdout and stderr to the same open file 565 tf = tempfile.TemporaryFile() 566 self.addCleanup(tf.close) 567 p = subprocess.Popen([sys.executable, "-c", 568 'import sys;' 569 'sys.stdout.write("apple");' 570 'sys.stdout.flush();' 571 'sys.stderr.write("orange")'], 572 stdout=tf, 573 stderr=tf) 574 p.wait() 575 tf.seek(0) 576 self.assertStderrEqual(tf.read(), b"appleorange") 577 578 def test_stdout_filedes_of_stdout(self): 579 # stdout is set to 1 (#1531862). 580 # To avoid printing the text on stdout, we do something similar to 581 # test_stdout_none (see above). The parent subprocess calls the child 582 # subprocess passing stdout=1, and this test uses stdout=PIPE in 583 # order to capture and check the output of the parent. See #11963. 584 code = ('import sys, subprocess; ' 585 'rc = subprocess.call([sys.executable, "-c", ' 586 ' "import os, sys; sys.exit(os.write(sys.stdout.fileno(), ' 587 'b\'test with stdout=1\'))"], stdout=1); ' 588 'assert rc == 18') 589 p = subprocess.Popen([sys.executable, "-c", code], 590 stdout=subprocess.PIPE, stderr=subprocess.PIPE) 591 self.addCleanup(p.stdout.close) 592 self.addCleanup(p.stderr.close) 593 out, err = p.communicate() 594 self.assertEqual(p.returncode, 0, err) 595 self.assertEqual(out.rstrip(), b'test with stdout=1') 596 597 def test_stdout_devnull(self): 598 p = subprocess.Popen([sys.executable, "-c", 599 'for i in range(10240):' 600 'print("x" * 1024)'], 601 stdout=subprocess.DEVNULL) 602 p.wait() 603 self.assertEqual(p.stdout, None) 604 605 def test_stderr_devnull(self): 606 p = subprocess.Popen([sys.executable, "-c", 607 'import sys\n' 608 'for i in range(10240):' 609 'sys.stderr.write("x" * 1024)'], 610 stderr=subprocess.DEVNULL) 611 p.wait() 612 self.assertEqual(p.stderr, None) 613 614 def test_stdin_devnull(self): 615 p = subprocess.Popen([sys.executable, "-c", 616 'import sys;' 617 'sys.stdin.read(1)'], 618 stdin=subprocess.DEVNULL) 619 p.wait() 620 self.assertEqual(p.stdin, None) 621 622 def test_env(self): 623 newenv = os.environ.copy() 624 newenv["FRUIT"] = "orange" 625 with subprocess.Popen([sys.executable, "-c", 626 'import sys,os;' 627 'sys.stdout.write(os.getenv("FRUIT"))'], 628 stdout=subprocess.PIPE, 629 env=newenv) as p: 630 stdout, stderr = p.communicate() 631 self.assertEqual(stdout, b"orange") 632 633 # Windows requires at least the SYSTEMROOT environment variable to start 634 # Python 635 @unittest.skipIf(sys.platform == 'win32', 636 'cannot test an empty env on Windows') 637 @unittest.skipIf(sysconfig.get_config_var('Py_ENABLE_SHARED') == 1, 638 'The Python shared library cannot be loaded ' 639 'with an empty environment.') 640 def test_empty_env(self): 641 """Verify that env={} is as empty as possible.""" 642 643 def is_env_var_to_ignore(n): 644 """Determine if an environment variable is under our control.""" 645 # This excludes some __CF_* and VERSIONER_* keys MacOS insists 646 # on adding even when the environment in exec is empty. 647 # Gentoo sandboxes also force LD_PRELOAD and SANDBOX_* to exist. 648 return ('VERSIONER' in n or '__CF' in n or # MacOS 649 '__PYVENV_LAUNCHER__' in n or # MacOS framework build 650 n == 'LD_PRELOAD' or n.startswith('SANDBOX') or # Gentoo 651 n == 'LC_CTYPE') # Locale coercion triggered 652 653 with subprocess.Popen([sys.executable, "-c", 654 'import os; print(list(os.environ.keys()))'], 655 stdout=subprocess.PIPE, env={}) as p: 656 stdout, stderr = p.communicate() 657 child_env_names = eval(stdout.strip()) 658 self.assertIsInstance(child_env_names, list) 659 child_env_names = [k for k in child_env_names 660 if not is_env_var_to_ignore(k)] 661 self.assertEqual(child_env_names, []) 662 663 def test_invalid_cmd(self): 664 # null character in the command name 665 cmd = sys.executable + '\0' 666 with self.assertRaises(ValueError): 667 subprocess.Popen([cmd, "-c", "pass"]) 668 669 # null character in the command argument 670 with self.assertRaises(ValueError): 671 subprocess.Popen([sys.executable, "-c", "pass#\0"]) 672 673 def test_invalid_env(self): 674 # null character in the environment variable name 675 newenv = os.environ.copy() 676 newenv["FRUIT\0VEGETABLE"] = "cabbage" 677 with self.assertRaises(ValueError): 678 subprocess.Popen([sys.executable, "-c", "pass"], env=newenv) 679 680 # null character in the environment variable value 681 newenv = os.environ.copy() 682 newenv["FRUIT"] = "orange\0VEGETABLE=cabbage" 683 with self.assertRaises(ValueError): 684 subprocess.Popen([sys.executable, "-c", "pass"], env=newenv) 685 686 # equal character in the environment variable name 687 newenv = os.environ.copy() 688 newenv["FRUIT=ORANGE"] = "lemon" 689 with self.assertRaises(ValueError): 690 subprocess.Popen([sys.executable, "-c", "pass"], env=newenv) 691 692 # equal character in the environment variable value 693 newenv = os.environ.copy() 694 newenv["FRUIT"] = "orange=lemon" 695 with subprocess.Popen([sys.executable, "-c", 696 'import sys, os;' 697 'sys.stdout.write(os.getenv("FRUIT"))'], 698 stdout=subprocess.PIPE, 699 env=newenv) as p: 700 stdout, stderr = p.communicate() 701 self.assertEqual(stdout, b"orange=lemon") 702 703 def test_communicate_stdin(self): 704 p = subprocess.Popen([sys.executable, "-c", 705 'import sys;' 706 'sys.exit(sys.stdin.read() == "pear")'], 707 stdin=subprocess.PIPE) 708 p.communicate(b"pear") 709 self.assertEqual(p.returncode, 1) 710 711 def test_communicate_stdout(self): 712 p = subprocess.Popen([sys.executable, "-c", 713 'import sys; sys.stdout.write("pineapple")'], 714 stdout=subprocess.PIPE) 715 (stdout, stderr) = p.communicate() 716 self.assertEqual(stdout, b"pineapple") 717 self.assertEqual(stderr, None) 718 719 def test_communicate_stderr(self): 720 p = subprocess.Popen([sys.executable, "-c", 721 'import sys; sys.stderr.write("pineapple")'], 722 stderr=subprocess.PIPE) 723 (stdout, stderr) = p.communicate() 724 self.assertEqual(stdout, None) 725 self.assertStderrEqual(stderr, b"pineapple") 726 727 def test_communicate(self): 728 p = subprocess.Popen([sys.executable, "-c", 729 'import sys,os;' 730 'sys.stderr.write("pineapple");' 731 'sys.stdout.write(sys.stdin.read())'], 732 stdin=subprocess.PIPE, 733 stdout=subprocess.PIPE, 734 stderr=subprocess.PIPE) 735 self.addCleanup(p.stdout.close) 736 self.addCleanup(p.stderr.close) 737 self.addCleanup(p.stdin.close) 738 (stdout, stderr) = p.communicate(b"banana") 739 self.assertEqual(stdout, b"banana") 740 self.assertStderrEqual(stderr, b"pineapple") 741 742 def test_communicate_timeout(self): 743 p = subprocess.Popen([sys.executable, "-c", 744 'import sys,os,time;' 745 'sys.stderr.write("pineapple\\n");' 746 'time.sleep(1);' 747 'sys.stderr.write("pear\\n");' 748 'sys.stdout.write(sys.stdin.read())'], 749 universal_newlines=True, 750 stdin=subprocess.PIPE, 751 stdout=subprocess.PIPE, 752 stderr=subprocess.PIPE) 753 self.assertRaises(subprocess.TimeoutExpired, p.communicate, "banana", 754 timeout=0.3) 755 # Make sure we can keep waiting for it, and that we get the whole output 756 # after it completes. 757 (stdout, stderr) = p.communicate() 758 self.assertEqual(stdout, "banana") 759 self.assertStderrEqual(stderr.encode(), b"pineapple\npear\n") 760 761 def test_communicate_timeout_large_output(self): 762 # Test an expiring timeout while the child is outputting lots of data. 763 p = subprocess.Popen([sys.executable, "-c", 764 'import sys,os,time;' 765 'sys.stdout.write("a" * (64 * 1024));' 766 'time.sleep(0.2);' 767 'sys.stdout.write("a" * (64 * 1024));' 768 'time.sleep(0.2);' 769 'sys.stdout.write("a" * (64 * 1024));' 770 'time.sleep(0.2);' 771 'sys.stdout.write("a" * (64 * 1024));'], 772 stdout=subprocess.PIPE) 773 self.assertRaises(subprocess.TimeoutExpired, p.communicate, timeout=0.4) 774 (stdout, _) = p.communicate() 775 self.assertEqual(len(stdout), 4 * 64 * 1024) 776 777 # Test for the fd leak reported in http://bugs.python.org/issue2791. 778 def test_communicate_pipe_fd_leak(self): 779 for stdin_pipe in (False, True): 780 for stdout_pipe in (False, True): 781 for stderr_pipe in (False, True): 782 options = {} 783 if stdin_pipe: 784 options['stdin'] = subprocess.PIPE 785 if stdout_pipe: 786 options['stdout'] = subprocess.PIPE 787 if stderr_pipe: 788 options['stderr'] = subprocess.PIPE 789 if not options: 790 continue 791 p = subprocess.Popen((sys.executable, "-c", "pass"), **options) 792 p.communicate() 793 if p.stdin is not None: 794 self.assertTrue(p.stdin.closed) 795 if p.stdout is not None: 796 self.assertTrue(p.stdout.closed) 797 if p.stderr is not None: 798 self.assertTrue(p.stderr.closed) 799 800 def test_communicate_returns(self): 801 # communicate() should return None if no redirection is active 802 p = subprocess.Popen([sys.executable, "-c", 803 "import sys; sys.exit(47)"]) 804 (stdout, stderr) = p.communicate() 805 self.assertEqual(stdout, None) 806 self.assertEqual(stderr, None) 807 808 def test_communicate_pipe_buf(self): 809 # communicate() with writes larger than pipe_buf 810 # This test will probably deadlock rather than fail, if 811 # communicate() does not work properly. 812 x, y = os.pipe() 813 os.close(x) 814 os.close(y) 815 p = subprocess.Popen([sys.executable, "-c", 816 'import sys,os;' 817 'sys.stdout.write(sys.stdin.read(47));' 818 'sys.stderr.write("x" * %d);' 819 'sys.stdout.write(sys.stdin.read())' % 820 support.PIPE_MAX_SIZE], 821 stdin=subprocess.PIPE, 822 stdout=subprocess.PIPE, 823 stderr=subprocess.PIPE) 824 self.addCleanup(p.stdout.close) 825 self.addCleanup(p.stderr.close) 826 self.addCleanup(p.stdin.close) 827 string_to_write = b"a" * support.PIPE_MAX_SIZE 828 (stdout, stderr) = p.communicate(string_to_write) 829 self.assertEqual(stdout, string_to_write) 830 831 def test_writes_before_communicate(self): 832 # stdin.write before communicate() 833 p = subprocess.Popen([sys.executable, "-c", 834 'import sys,os;' 835 'sys.stdout.write(sys.stdin.read())'], 836 stdin=subprocess.PIPE, 837 stdout=subprocess.PIPE, 838 stderr=subprocess.PIPE) 839 self.addCleanup(p.stdout.close) 840 self.addCleanup(p.stderr.close) 841 self.addCleanup(p.stdin.close) 842 p.stdin.write(b"banana") 843 (stdout, stderr) = p.communicate(b"split") 844 self.assertEqual(stdout, b"bananasplit") 845 self.assertStderrEqual(stderr, b"") 846 847 def test_universal_newlines_and_text(self): 848 args = [ 849 sys.executable, "-c", 850 'import sys,os;' + SETBINARY + 851 'buf = sys.stdout.buffer;' 852 'buf.write(sys.stdin.readline().encode());' 853 'buf.flush();' 854 'buf.write(b"line2\\n");' 855 'buf.flush();' 856 'buf.write(sys.stdin.read().encode());' 857 'buf.flush();' 858 'buf.write(b"line4\\n");' 859 'buf.flush();' 860 'buf.write(b"line5\\r\\n");' 861 'buf.flush();' 862 'buf.write(b"line6\\r");' 863 'buf.flush();' 864 'buf.write(b"\\nline7");' 865 'buf.flush();' 866 'buf.write(b"\\nline8");'] 867 868 for extra_kwarg in ('universal_newlines', 'text'): 869 p = subprocess.Popen(args, **{'stdin': subprocess.PIPE, 870 'stdout': subprocess.PIPE, 871 extra_kwarg: True}) 872 with p: 873 p.stdin.write("line1\n") 874 p.stdin.flush() 875 self.assertEqual(p.stdout.readline(), "line1\n") 876 p.stdin.write("line3\n") 877 p.stdin.close() 878 self.addCleanup(p.stdout.close) 879 self.assertEqual(p.stdout.readline(), 880 "line2\n") 881 self.assertEqual(p.stdout.read(6), 882 "line3\n") 883 self.assertEqual(p.stdout.read(), 884 "line4\nline5\nline6\nline7\nline8") 885 886 def test_universal_newlines_communicate(self): 887 # universal newlines through communicate() 888 p = subprocess.Popen([sys.executable, "-c", 889 'import sys,os;' + SETBINARY + 890 'buf = sys.stdout.buffer;' 891 'buf.write(b"line2\\n");' 892 'buf.flush();' 893 'buf.write(b"line4\\n");' 894 'buf.flush();' 895 'buf.write(b"line5\\r\\n");' 896 'buf.flush();' 897 'buf.write(b"line6\\r");' 898 'buf.flush();' 899 'buf.write(b"\\nline7");' 900 'buf.flush();' 901 'buf.write(b"\\nline8");'], 902 stderr=subprocess.PIPE, 903 stdout=subprocess.PIPE, 904 universal_newlines=1) 905 self.addCleanup(p.stdout.close) 906 self.addCleanup(p.stderr.close) 907 (stdout, stderr) = p.communicate() 908 self.assertEqual(stdout, 909 "line2\nline4\nline5\nline6\nline7\nline8") 910 911 def test_universal_newlines_communicate_stdin(self): 912 # universal newlines through communicate(), with only stdin 913 p = subprocess.Popen([sys.executable, "-c", 914 'import sys,os;' + SETBINARY + textwrap.dedent(''' 915 s = sys.stdin.readline() 916 assert s == "line1\\n", repr(s) 917 s = sys.stdin.read() 918 assert s == "line3\\n", repr(s) 919 ''')], 920 stdin=subprocess.PIPE, 921 universal_newlines=1) 922 (stdout, stderr) = p.communicate("line1\nline3\n") 923 self.assertEqual(p.returncode, 0) 924 925 def test_universal_newlines_communicate_input_none(self): 926 # Test communicate(input=None) with universal newlines. 927 # 928 # We set stdout to PIPE because, as of this writing, a different 929 # code path is tested when the number of pipes is zero or one. 930 p = subprocess.Popen([sys.executable, "-c", "pass"], 931 stdin=subprocess.PIPE, 932 stdout=subprocess.PIPE, 933 universal_newlines=True) 934 p.communicate() 935 self.assertEqual(p.returncode, 0) 936 937 def test_universal_newlines_communicate_stdin_stdout_stderr(self): 938 # universal newlines through communicate(), with stdin, stdout, stderr 939 p = subprocess.Popen([sys.executable, "-c", 940 'import sys,os;' + SETBINARY + textwrap.dedent(''' 941 s = sys.stdin.buffer.readline() 942 sys.stdout.buffer.write(s) 943 sys.stdout.buffer.write(b"line2\\r") 944 sys.stderr.buffer.write(b"eline2\\n") 945 s = sys.stdin.buffer.read() 946 sys.stdout.buffer.write(s) 947 sys.stdout.buffer.write(b"line4\\n") 948 sys.stdout.buffer.write(b"line5\\r\\n") 949 sys.stderr.buffer.write(b"eline6\\r") 950 sys.stderr.buffer.write(b"eline7\\r\\nz") 951 ''')], 952 stdin=subprocess.PIPE, 953 stderr=subprocess.PIPE, 954 stdout=subprocess.PIPE, 955 universal_newlines=True) 956 self.addCleanup(p.stdout.close) 957 self.addCleanup(p.stderr.close) 958 (stdout, stderr) = p.communicate("line1\nline3\n") 959 self.assertEqual(p.returncode, 0) 960 self.assertEqual("line1\nline2\nline3\nline4\nline5\n", stdout) 961 # Python debug build push something like "[42442 refs]\n" 962 # to stderr at exit of subprocess. 963 # Don't use assertStderrEqual because it strips CR and LF from output. 964 self.assertTrue(stderr.startswith("eline2\neline6\neline7\n")) 965 966 def test_universal_newlines_communicate_encodings(self): 967 # Check that universal newlines mode works for various encodings, 968 # in particular for encodings in the UTF-16 and UTF-32 families. 969 # See issue #15595. 970 # 971 # UTF-16 and UTF-32-BE are sufficient to check both with BOM and 972 # without, and UTF-16 and UTF-32. 973 for encoding in ['utf-16', 'utf-32-be']: 974 code = ("import sys; " 975 r"sys.stdout.buffer.write('1\r\n2\r3\n4'.encode('%s'))" % 976 encoding) 977 args = [sys.executable, '-c', code] 978 # We set stdin to be non-None because, as of this writing, 979 # a different code path is used when the number of pipes is 980 # zero or one. 981 popen = subprocess.Popen(args, 982 stdin=subprocess.PIPE, 983 stdout=subprocess.PIPE, 984 encoding=encoding) 985 stdout, stderr = popen.communicate(input='') 986 self.assertEqual(stdout, '1\n2\n3\n4') 987 988 def test_communicate_errors(self): 989 for errors, expected in [ 990 ('ignore', ''), 991 ('replace', '\ufffd\ufffd'), 992 ('surrogateescape', '\udc80\udc80'), 993 ('backslashreplace', '\\x80\\x80'), 994 ]: 995 code = ("import sys; " 996 r"sys.stdout.buffer.write(b'[\x80\x80]')") 997 args = [sys.executable, '-c', code] 998 # We set stdin to be non-None because, as of this writing, 999 # a different code path is used when the number of pipes is 1000 # zero or one. 1001 popen = subprocess.Popen(args, 1002 stdin=subprocess.PIPE, 1003 stdout=subprocess.PIPE, 1004 encoding='utf-8', 1005 errors=errors) 1006 stdout, stderr = popen.communicate(input='') 1007 self.assertEqual(stdout, '[{}]'.format(expected)) 1008 1009 def test_no_leaking(self): 1010 # Make sure we leak no resources 1011 if not mswindows: 1012 max_handles = 1026 # too much for most UNIX systems 1013 else: 1014 max_handles = 2050 # too much for (at least some) Windows setups 1015 handles = [] 1016 tmpdir = tempfile.mkdtemp() 1017 try: 1018 for i in range(max_handles): 1019 try: 1020 tmpfile = os.path.join(tmpdir, support.TESTFN) 1021 handles.append(os.open(tmpfile, os.O_WRONLY|os.O_CREAT)) 1022 except OSError as e: 1023 if e.errno != errno.EMFILE: 1024 raise 1025 break 1026 else: 1027 self.skipTest("failed to reach the file descriptor limit " 1028 "(tried %d)" % max_handles) 1029 # Close a couple of them (should be enough for a subprocess) 1030 for i in range(10): 1031 os.close(handles.pop()) 1032 # Loop creating some subprocesses. If one of them leaks some fds, 1033 # the next loop iteration will fail by reaching the max fd limit. 1034 for i in range(15): 1035 p = subprocess.Popen([sys.executable, "-c", 1036 "import sys;" 1037 "sys.stdout.write(sys.stdin.read())"], 1038 stdin=subprocess.PIPE, 1039 stdout=subprocess.PIPE, 1040 stderr=subprocess.PIPE) 1041 data = p.communicate(b"lime")[0] 1042 self.assertEqual(data, b"lime") 1043 finally: 1044 for h in handles: 1045 os.close(h) 1046 shutil.rmtree(tmpdir) 1047 1048 def test_list2cmdline(self): 1049 self.assertEqual(subprocess.list2cmdline(['a b c', 'd', 'e']), 1050 '"a b c" d e') 1051 self.assertEqual(subprocess.list2cmdline(['ab"c', '\\', 'd']), 1052 'ab\\"c \\ d') 1053 self.assertEqual(subprocess.list2cmdline(['ab"c', ' \\', 'd']), 1054 'ab\\"c " \\\\" d') 1055 self.assertEqual(subprocess.list2cmdline(['a\\\\\\b', 'de fg', 'h']), 1056 'a\\\\\\b "de fg" h') 1057 self.assertEqual(subprocess.list2cmdline(['a\\"b', 'c', 'd']), 1058 'a\\\\\\"b c d') 1059 self.assertEqual(subprocess.list2cmdline(['a\\\\b c', 'd', 'e']), 1060 '"a\\\\b c" d e') 1061 self.assertEqual(subprocess.list2cmdline(['a\\\\b\\ c', 'd', 'e']), 1062 '"a\\\\b\\ c" d e') 1063 self.assertEqual(subprocess.list2cmdline(['ab', '']), 1064 'ab ""') 1065 1066 def test_poll(self): 1067 p = subprocess.Popen([sys.executable, "-c", 1068 "import os; os.read(0, 1)"], 1069 stdin=subprocess.PIPE) 1070 self.addCleanup(p.stdin.close) 1071 self.assertIsNone(p.poll()) 1072 os.write(p.stdin.fileno(), b'A') 1073 p.wait() 1074 # Subsequent invocations should just return the returncode 1075 self.assertEqual(p.poll(), 0) 1076 1077 def test_wait(self): 1078 p = subprocess.Popen([sys.executable, "-c", "pass"]) 1079 self.assertEqual(p.wait(), 0) 1080 # Subsequent invocations should just return the returncode 1081 self.assertEqual(p.wait(), 0) 1082 1083 def test_wait_timeout(self): 1084 p = subprocess.Popen([sys.executable, 1085 "-c", "import time; time.sleep(0.3)"]) 1086 with self.assertRaises(subprocess.TimeoutExpired) as c: 1087 p.wait(timeout=0.0001) 1088 self.assertIn("0.0001", str(c.exception)) # For coverage of __str__. 1089 # Some heavily loaded buildbots (sparc Debian 3.x) require this much 1090 # time to start. 1091 self.assertEqual(p.wait(timeout=3), 0) 1092 1093 def test_invalid_bufsize(self): 1094 # an invalid type of the bufsize argument should raise 1095 # TypeError. 1096 with self.assertRaises(TypeError): 1097 subprocess.Popen([sys.executable, "-c", "pass"], "orange") 1098 1099 def test_bufsize_is_none(self): 1100 # bufsize=None should be the same as bufsize=0. 1101 p = subprocess.Popen([sys.executable, "-c", "pass"], None) 1102 self.assertEqual(p.wait(), 0) 1103 # Again with keyword arg 1104 p = subprocess.Popen([sys.executable, "-c", "pass"], bufsize=None) 1105 self.assertEqual(p.wait(), 0) 1106 1107 def _test_bufsize_equal_one(self, line, expected, universal_newlines): 1108 # subprocess may deadlock with bufsize=1, see issue #21332 1109 with subprocess.Popen([sys.executable, "-c", "import sys;" 1110 "sys.stdout.write(sys.stdin.readline());" 1111 "sys.stdout.flush()"], 1112 stdin=subprocess.PIPE, 1113 stdout=subprocess.PIPE, 1114 stderr=subprocess.DEVNULL, 1115 bufsize=1, 1116 universal_newlines=universal_newlines) as p: 1117 p.stdin.write(line) # expect that it flushes the line in text mode 1118 os.close(p.stdin.fileno()) # close it without flushing the buffer 1119 read_line = p.stdout.readline() 1120 with support.SuppressCrashReport(): 1121 try: 1122 p.stdin.close() 1123 except OSError: 1124 pass 1125 p.stdin = None 1126 self.assertEqual(p.returncode, 0) 1127 self.assertEqual(read_line, expected) 1128 1129 def test_bufsize_equal_one_text_mode(self): 1130 # line is flushed in text mode with bufsize=1. 1131 # we should get the full line in return 1132 line = "line\n" 1133 self._test_bufsize_equal_one(line, line, universal_newlines=True) 1134 1135 def test_bufsize_equal_one_binary_mode(self): 1136 # line is not flushed in binary mode with bufsize=1. 1137 # we should get empty response 1138 line = b'line' + os.linesep.encode() # assume ascii-based locale 1139 self._test_bufsize_equal_one(line, b'', universal_newlines=False) 1140 1141 def test_leaking_fds_on_error(self): 1142 # see bug #5179: Popen leaks file descriptors to PIPEs if 1143 # the child fails to execute; this will eventually exhaust 1144 # the maximum number of open fds. 1024 seems a very common 1145 # value for that limit, but Windows has 2048, so we loop 1146 # 1024 times (each call leaked two fds). 1147 for i in range(1024): 1148 with self.assertRaises(NONEXISTING_ERRORS): 1149 subprocess.Popen(NONEXISTING_CMD, 1150 stdout=subprocess.PIPE, 1151 stderr=subprocess.PIPE) 1152 1153 def test_nonexisting_with_pipes(self): 1154 # bpo-30121: Popen with pipes must close properly pipes on error. 1155 # Previously, os.close() was called with a Windows handle which is not 1156 # a valid file descriptor. 1157 # 1158 # Run the test in a subprocess to control how the CRT reports errors 1159 # and to get stderr content. 1160 try: 1161 import msvcrt 1162 msvcrt.CrtSetReportMode 1163 except (AttributeError, ImportError): 1164 self.skipTest("need msvcrt.CrtSetReportMode") 1165 1166 code = textwrap.dedent(f""" 1167 import msvcrt 1168 import subprocess 1169 1170 cmd = {NONEXISTING_CMD!r} 1171 1172 for report_type in [msvcrt.CRT_WARN, 1173 msvcrt.CRT_ERROR, 1174 msvcrt.CRT_ASSERT]: 1175 msvcrt.CrtSetReportMode(report_type, msvcrt.CRTDBG_MODE_FILE) 1176 msvcrt.CrtSetReportFile(report_type, msvcrt.CRTDBG_FILE_STDERR) 1177 1178 try: 1179 subprocess.Popen(cmd, 1180 stdout=subprocess.PIPE, 1181 stderr=subprocess.PIPE) 1182 except OSError: 1183 pass 1184 """) 1185 cmd = [sys.executable, "-c", code] 1186 proc = subprocess.Popen(cmd, 1187 stderr=subprocess.PIPE, 1188 universal_newlines=True) 1189 with proc: 1190 stderr = proc.communicate()[1] 1191 self.assertEqual(stderr, "") 1192 self.assertEqual(proc.returncode, 0) 1193 1194 def test_double_close_on_error(self): 1195 # Issue #18851 1196 fds = [] 1197 def open_fds(): 1198 for i in range(20): 1199 fds.extend(os.pipe()) 1200 time.sleep(0.001) 1201 t = threading.Thread(target=open_fds) 1202 t.start() 1203 try: 1204 with self.assertRaises(EnvironmentError): 1205 subprocess.Popen(NONEXISTING_CMD, 1206 stdin=subprocess.PIPE, 1207 stdout=subprocess.PIPE, 1208 stderr=subprocess.PIPE) 1209 finally: 1210 t.join() 1211 exc = None 1212 for fd in fds: 1213 # If a double close occurred, some of those fds will 1214 # already have been closed by mistake, and os.close() 1215 # here will raise. 1216 try: 1217 os.close(fd) 1218 except OSError as e: 1219 exc = e 1220 if exc is not None: 1221 raise exc 1222 1223 def test_threadsafe_wait(self): 1224 """Issue21291: Popen.wait() needs to be threadsafe for returncode.""" 1225 proc = subprocess.Popen([sys.executable, '-c', 1226 'import time; time.sleep(12)']) 1227 self.assertEqual(proc.returncode, None) 1228 results = [] 1229 1230 def kill_proc_timer_thread(): 1231 results.append(('thread-start-poll-result', proc.poll())) 1232 # terminate it from the thread and wait for the result. 1233 proc.kill() 1234 proc.wait() 1235 results.append(('thread-after-kill-and-wait', proc.returncode)) 1236 # this wait should be a no-op given the above. 1237 proc.wait() 1238 results.append(('thread-after-second-wait', proc.returncode)) 1239 1240 # This is a timing sensitive test, the failure mode is 1241 # triggered when both the main thread and this thread are in 1242 # the wait() call at once. The delay here is to allow the 1243 # main thread to most likely be blocked in its wait() call. 1244 t = threading.Timer(0.2, kill_proc_timer_thread) 1245 t.start() 1246 1247 if mswindows: 1248 expected_errorcode = 1 1249 else: 1250 # Should be -9 because of the proc.kill() from the thread. 1251 expected_errorcode = -9 1252 1253 # Wait for the process to finish; the thread should kill it 1254 # long before it finishes on its own. Supplying a timeout 1255 # triggers a different code path for better coverage. 1256 proc.wait(timeout=20) 1257 self.assertEqual(proc.returncode, expected_errorcode, 1258 msg="unexpected result in wait from main thread") 1259 1260 # This should be a no-op with no change in returncode. 1261 proc.wait() 1262 self.assertEqual(proc.returncode, expected_errorcode, 1263 msg="unexpected result in second main wait.") 1264 1265 t.join() 1266 # Ensure that all of the thread results are as expected. 1267 # When a race condition occurs in wait(), the returncode could 1268 # be set by the wrong thread that doesn't actually have it 1269 # leading to an incorrect value. 1270 self.assertEqual([('thread-start-poll-result', None), 1271 ('thread-after-kill-and-wait', expected_errorcode), 1272 ('thread-after-second-wait', expected_errorcode)], 1273 results) 1274 1275 def test_issue8780(self): 1276 # Ensure that stdout is inherited from the parent 1277 # if stdout=PIPE is not used 1278 code = ';'.join(( 1279 'import subprocess, sys', 1280 'retcode = subprocess.call(' 1281 "[sys.executable, '-c', 'print(\"Hello World!\")'])", 1282 'assert retcode == 0')) 1283 output = subprocess.check_output([sys.executable, '-c', code]) 1284 self.assertTrue(output.startswith(b'Hello World!'), ascii(output)) 1285 1286 def test_handles_closed_on_exception(self): 1287 # If CreateProcess exits with an error, ensure the 1288 # duplicate output handles are released 1289 ifhandle, ifname = tempfile.mkstemp() 1290 ofhandle, ofname = tempfile.mkstemp() 1291 efhandle, efname = tempfile.mkstemp() 1292 try: 1293 subprocess.Popen (["*"], stdin=ifhandle, stdout=ofhandle, 1294 stderr=efhandle) 1295 except OSError: 1296 os.close(ifhandle) 1297 os.remove(ifname) 1298 os.close(ofhandle) 1299 os.remove(ofname) 1300 os.close(efhandle) 1301 os.remove(efname) 1302 self.assertFalse(os.path.exists(ifname)) 1303 self.assertFalse(os.path.exists(ofname)) 1304 self.assertFalse(os.path.exists(efname)) 1305 1306 def test_communicate_epipe(self): 1307 # Issue 10963: communicate() should hide EPIPE 1308 p = subprocess.Popen([sys.executable, "-c", 'pass'], 1309 stdin=subprocess.PIPE, 1310 stdout=subprocess.PIPE, 1311 stderr=subprocess.PIPE) 1312 self.addCleanup(p.stdout.close) 1313 self.addCleanup(p.stderr.close) 1314 self.addCleanup(p.stdin.close) 1315 p.communicate(b"x" * 2**20) 1316 1317 def test_communicate_epipe_only_stdin(self): 1318 # Issue 10963: communicate() should hide EPIPE 1319 p = subprocess.Popen([sys.executable, "-c", 'pass'], 1320 stdin=subprocess.PIPE) 1321 self.addCleanup(p.stdin.close) 1322 p.wait() 1323 p.communicate(b"x" * 2**20) 1324 1325 @unittest.skipUnless(hasattr(signal, 'SIGUSR1'), 1326 "Requires signal.SIGUSR1") 1327 @unittest.skipUnless(hasattr(os, 'kill'), 1328 "Requires os.kill") 1329 @unittest.skipUnless(hasattr(os, 'getppid'), 1330 "Requires os.getppid") 1331 def test_communicate_eintr(self): 1332 # Issue #12493: communicate() should handle EINTR 1333 def handler(signum, frame): 1334 pass 1335 old_handler = signal.signal(signal.SIGUSR1, handler) 1336 self.addCleanup(signal.signal, signal.SIGUSR1, old_handler) 1337 1338 args = [sys.executable, "-c", 1339 'import os, signal;' 1340 'os.kill(os.getppid(), signal.SIGUSR1)'] 1341 for stream in ('stdout', 'stderr'): 1342 kw = {stream: subprocess.PIPE} 1343 with subprocess.Popen(args, **kw) as process: 1344 # communicate() will be interrupted by SIGUSR1 1345 process.communicate() 1346 1347 1348 # This test is Linux-ish specific for simplicity to at least have 1349 # some coverage. It is not a platform specific bug. 1350 @unittest.skipUnless(os.path.isdir('/proc/%d/fd' % os.getpid()), 1351 "Linux specific") 1352 def test_failed_child_execute_fd_leak(self): 1353 """Test for the fork() failure fd leak reported in issue16327.""" 1354 fd_directory = '/proc/%d/fd' % os.getpid() 1355 fds_before_popen = os.listdir(fd_directory) 1356 with self.assertRaises(PopenTestException): 1357 PopenExecuteChildRaises( 1358 [sys.executable, '-c', 'pass'], stdin=subprocess.PIPE, 1359 stdout=subprocess.PIPE, stderr=subprocess.PIPE) 1360 1361 # NOTE: This test doesn't verify that the real _execute_child 1362 # does not close the file descriptors itself on the way out 1363 # during an exception. Code inspection has confirmed that. 1364 1365 fds_after_exception = os.listdir(fd_directory) 1366 self.assertEqual(fds_before_popen, fds_after_exception) 1367 1368 @unittest.skipIf(mswindows, "behavior currently not supported on Windows") 1369 def test_file_not_found_includes_filename(self): 1370 with self.assertRaises(FileNotFoundError) as c: 1371 subprocess.call(['/opt/nonexistent_binary', 'with', 'some', 'args']) 1372 self.assertEqual(c.exception.filename, '/opt/nonexistent_binary') 1373 1374 @unittest.skipIf(mswindows, "behavior currently not supported on Windows") 1375 def test_file_not_found_with_bad_cwd(self): 1376 with self.assertRaises(FileNotFoundError) as c: 1377 subprocess.Popen(['exit', '0'], cwd='/some/nonexistent/directory') 1378 self.assertEqual(c.exception.filename, '/some/nonexistent/directory') 1379 1380 1381 class RunFuncTestCase(BaseTestCase): 1382 def run_python(self, code, **kwargs): 1383 """Run Python code in a subprocess using subprocess.run""" 1384 argv = [sys.executable, "-c", code] 1385 return subprocess.run(argv, **kwargs) 1386 1387 def test_returncode(self): 1388 # call() function with sequence argument 1389 cp = self.run_python("import sys; sys.exit(47)") 1390 self.assertEqual(cp.returncode, 47) 1391 with self.assertRaises(subprocess.CalledProcessError): 1392 cp.check_returncode() 1393 1394 def test_check(self): 1395 with self.assertRaises(subprocess.CalledProcessError) as c: 1396 self.run_python("import sys; sys.exit(47)", check=True) 1397 self.assertEqual(c.exception.returncode, 47) 1398 1399 def test_check_zero(self): 1400 # check_returncode shouldn't raise when returncode is zero 1401 cp = self.run_python("import sys; sys.exit(0)", check=True) 1402 self.assertEqual(cp.returncode, 0) 1403 1404 def test_timeout(self): 1405 # run() function with timeout argument; we want to test that the child 1406 # process gets killed when the timeout expires. If the child isn't 1407 # killed, this call will deadlock since subprocess.run waits for the 1408 # child. 1409 with self.assertRaises(subprocess.TimeoutExpired): 1410 self.run_python("while True: pass", timeout=0.0001) 1411 1412 def test_capture_stdout(self): 1413 # capture stdout with zero return code 1414 cp = self.run_python("print('BDFL')", stdout=subprocess.PIPE) 1415 self.assertIn(b'BDFL', cp.stdout) 1416 1417 def test_capture_stderr(self): 1418 cp = self.run_python("import sys; sys.stderr.write('BDFL')", 1419 stderr=subprocess.PIPE) 1420 self.assertIn(b'BDFL', cp.stderr) 1421 1422 def test_check_output_stdin_arg(self): 1423 # run() can be called with stdin set to a file 1424 tf = tempfile.TemporaryFile() 1425 self.addCleanup(tf.close) 1426 tf.write(b'pear') 1427 tf.seek(0) 1428 cp = self.run_python( 1429 "import sys; sys.stdout.write(sys.stdin.read().upper())", 1430 stdin=tf, stdout=subprocess.PIPE) 1431 self.assertIn(b'PEAR', cp.stdout) 1432 1433 def test_check_output_input_arg(self): 1434 # check_output() can be called with input set to a string 1435 cp = self.run_python( 1436 "import sys; sys.stdout.write(sys.stdin.read().upper())", 1437 input=b'pear', stdout=subprocess.PIPE) 1438 self.assertIn(b'PEAR', cp.stdout) 1439 1440 def test_check_output_stdin_with_input_arg(self): 1441 # run() refuses to accept 'stdin' with 'input' 1442 tf = tempfile.TemporaryFile() 1443 self.addCleanup(tf.close) 1444 tf.write(b'pear') 1445 tf.seek(0) 1446 with self.assertRaises(ValueError, 1447 msg="Expected ValueError when stdin and input args supplied.") as c: 1448 output = self.run_python("print('will not be run')", 1449 stdin=tf, input=b'hare') 1450 self.assertIn('stdin', c.exception.args[0]) 1451 self.assertIn('input', c.exception.args[0]) 1452 1453 def test_check_output_timeout(self): 1454 with self.assertRaises(subprocess.TimeoutExpired) as c: 1455 cp = self.run_python(( 1456 "import sys, time\n" 1457 "sys.stdout.write('BDFL')\n" 1458 "sys.stdout.flush()\n" 1459 "time.sleep(3600)"), 1460 # Some heavily loaded buildbots (sparc Debian 3.x) require 1461 # this much time to start and print. 1462 timeout=3, stdout=subprocess.PIPE) 1463 self.assertEqual(c.exception.output, b'BDFL') 1464 # output is aliased to stdout 1465 self.assertEqual(c.exception.stdout, b'BDFL') 1466 1467 def test_run_kwargs(self): 1468 newenv = os.environ.copy() 1469 newenv["FRUIT"] = "banana" 1470 cp = self.run_python(('import sys, os;' 1471 'sys.exit(33 if os.getenv("FRUIT")=="banana" else 31)'), 1472 env=newenv) 1473 self.assertEqual(cp.returncode, 33) 1474 1475 def test_capture_output(self): 1476 cp = self.run_python(("import sys;" 1477 "sys.stdout.write('BDFL'); " 1478 "sys.stderr.write('FLUFL')"), 1479 capture_output=True) 1480 self.assertIn(b'BDFL', cp.stdout) 1481 self.assertIn(b'FLUFL', cp.stderr) 1482 1483 def test_stdout_with_capture_output_arg(self): 1484 # run() refuses to accept 'stdout' with 'capture_output' 1485 tf = tempfile.TemporaryFile() 1486 self.addCleanup(tf.close) 1487 with self.assertRaises(ValueError, 1488 msg=("Expected ValueError when stdout and capture_output " 1489 "args supplied.")) as c: 1490 output = self.run_python("print('will not be run')", 1491 capture_output=True, stdout=tf) 1492 self.assertIn('stdout', c.exception.args[0]) 1493 self.assertIn('capture_output', c.exception.args[0]) 1494 1495 def test_stderr_with_capture_output_arg(self): 1496 # run() refuses to accept 'stderr' with 'capture_output' 1497 tf = tempfile.TemporaryFile() 1498 self.addCleanup(tf.close) 1499 with self.assertRaises(ValueError, 1500 msg=("Expected ValueError when stderr and capture_output " 1501 "args supplied.")) as c: 1502 output = self.run_python("print('will not be run')", 1503 capture_output=True, stderr=tf) 1504 self.assertIn('stderr', c.exception.args[0]) 1505 self.assertIn('capture_output', c.exception.args[0]) 1506 1507 1508 @unittest.skipIf(mswindows, "POSIX specific tests") 1509 class POSIXProcessTestCase(BaseTestCase): 1510 1511 def setUp(self): 1512 super().setUp() 1513 self._nonexistent_dir = "/_this/pa.th/does/not/exist" 1514 1515 def _get_chdir_exception(self): 1516 try: 1517 os.chdir(self._nonexistent_dir) 1518 except OSError as e: 1519 # This avoids hard coding the errno value or the OS perror() 1520 # string and instead capture the exception that we want to see 1521 # below for comparison. 1522 desired_exception = e 1523 desired_exception.strerror += ': ' + repr(self._nonexistent_dir) 1524 else: 1525 self.fail("chdir to nonexistent directory %s succeeded." % 1526 self._nonexistent_dir) 1527 return desired_exception 1528 1529 def test_exception_cwd(self): 1530 """Test error in the child raised in the parent for a bad cwd.""" 1531 desired_exception = self._get_chdir_exception() 1532 try: 1533 p = subprocess.Popen([sys.executable, "-c", ""], 1534 cwd=self._nonexistent_dir) 1535 except OSError as e: 1536 # Test that the child process chdir failure actually makes 1537 # it up to the parent process as the correct exception. 1538 self.assertEqual(desired_exception.errno, e.errno) 1539 self.assertEqual(desired_exception.strerror, e.strerror) 1540 else: 1541 self.fail("Expected OSError: %s" % desired_exception) 1542 1543 def test_exception_bad_executable(self): 1544 """Test error in the child raised in the parent for a bad executable.""" 1545 desired_exception = self._get_chdir_exception() 1546 try: 1547 p = subprocess.Popen([sys.executable, "-c", ""], 1548 executable=self._nonexistent_dir) 1549 except OSError as e: 1550 # Test that the child process exec failure actually makes 1551 # it up to the parent process as the correct exception. 1552 self.assertEqual(desired_exception.errno, e.errno) 1553 self.assertEqual(desired_exception.strerror, e.strerror) 1554 else: 1555 self.fail("Expected OSError: %s" % desired_exception) 1556 1557 def test_exception_bad_args_0(self): 1558 """Test error in the child raised in the parent for a bad args[0].""" 1559 desired_exception = self._get_chdir_exception() 1560 try: 1561 p = subprocess.Popen([self._nonexistent_dir, "-c", ""]) 1562 except OSError as e: 1563 # Test that the child process exec failure actually makes 1564 # it up to the parent process as the correct exception. 1565 self.assertEqual(desired_exception.errno, e.errno) 1566 self.assertEqual(desired_exception.strerror, e.strerror) 1567 else: 1568 self.fail("Expected OSError: %s" % desired_exception) 1569 1570 # We mock the __del__ method for Popen in the next two tests 1571 # because it does cleanup based on the pid returned by fork_exec 1572 # along with issuing a resource warning if it still exists. Since 1573 # we don't actually spawn a process in these tests we can forego 1574 # the destructor. An alternative would be to set _child_created to 1575 # False before the destructor is called but there is no easy way 1576 # to do that 1577 class PopenNoDestructor(subprocess.Popen): 1578 def __del__(self): 1579 pass 1580 1581 @mock.patch("subprocess._posixsubprocess.fork_exec") 1582 def test_exception_errpipe_normal(self, fork_exec): 1583 """Test error passing done through errpipe_write in the good case""" 1584 def proper_error(*args): 1585 errpipe_write = args[13] 1586 # Write the hex for the error code EISDIR: 'is a directory' 1587 err_code = '{:x}'.format(errno.EISDIR).encode() 1588 os.write(errpipe_write, b"OSError:" + err_code + b":") 1589 return 0 1590 1591 fork_exec.side_effect = proper_error 1592 1593 with mock.patch("subprocess.os.waitpid", 1594 side_effect=ChildProcessError): 1595 with self.assertRaises(IsADirectoryError): 1596 self.PopenNoDestructor(["non_existent_command"]) 1597 1598 @mock.patch("subprocess._posixsubprocess.fork_exec") 1599 def test_exception_errpipe_bad_data(self, fork_exec): 1600 """Test error passing done through errpipe_write where its not 1601 in the expected format""" 1602 error_data = b"\xFF\x00\xDE\xAD" 1603 def bad_error(*args): 1604 errpipe_write = args[13] 1605 # Anything can be in the pipe, no assumptions should 1606 # be made about its encoding, so we'll write some 1607 # arbitrary hex bytes to test it out 1608 os.write(errpipe_write, error_data) 1609 return 0 1610 1611 fork_exec.side_effect = bad_error 1612 1613 with mock.patch("subprocess.os.waitpid", 1614 side_effect=ChildProcessError): 1615 with self.assertRaises(subprocess.SubprocessError) as e: 1616 self.PopenNoDestructor(["non_existent_command"]) 1617 1618 self.assertIn(repr(error_data), str(e.exception)) 1619 1620 @unittest.skipIf(not os.path.exists('/proc/self/status'), 1621 "need /proc/self/status") 1622 def test_restore_signals(self): 1623 # Blindly assume that cat exists on systems with /proc/self/status... 1624 default_proc_status = subprocess.check_output( 1625 ['cat', '/proc/self/status'], 1626 restore_signals=False) 1627 for line in default_proc_status.splitlines(): 1628 if line.startswith(b'SigIgn'): 1629 default_sig_ign_mask = line 1630 break 1631 else: 1632 self.skipTest("SigIgn not found in /proc/self/status.") 1633 restored_proc_status = subprocess.check_output( 1634 ['cat', '/proc/self/status'], 1635 restore_signals=True) 1636 for line in restored_proc_status.splitlines(): 1637 if line.startswith(b'SigIgn'): 1638 restored_sig_ign_mask = line 1639 break 1640 self.assertNotEqual(default_sig_ign_mask, restored_sig_ign_mask, 1641 msg="restore_signals=True should've unblocked " 1642 "SIGPIPE and friends.") 1643 1644 def test_start_new_session(self): 1645 # For code coverage of calling setsid(). We don't care if we get an 1646 # EPERM error from it depending on the test execution environment, that 1647 # still indicates that it was called. 1648 try: 1649 output = subprocess.check_output( 1650 [sys.executable, "-c", 1651 "import os; print(os.getpgid(os.getpid()))"], 1652 start_new_session=True) 1653 except OSError as e: 1654 if e.errno != errno.EPERM: 1655 raise 1656 else: 1657 parent_pgid = os.getpgid(os.getpid()) 1658 child_pgid = int(output) 1659 self.assertNotEqual(parent_pgid, child_pgid) 1660 1661 def test_run_abort(self): 1662 # returncode handles signal termination 1663 with support.SuppressCrashReport(): 1664 p = subprocess.Popen([sys.executable, "-c", 1665 'import os; os.abort()']) 1666 p.wait() 1667 self.assertEqual(-p.returncode, signal.SIGABRT) 1668 1669 def test_CalledProcessError_str_signal(self): 1670 err = subprocess.CalledProcessError(-int(signal.SIGABRT), "fake cmd") 1671 error_string = str(err) 1672 # We're relying on the repr() of the signal.Signals intenum to provide 1673 # the word signal, the signal name and the numeric value. 1674 self.assertIn("signal", error_string.lower()) 1675 # We're not being specific about the signal name as some signals have 1676 # multiple names and which name is revealed can vary. 1677 self.assertIn("SIG", error_string) 1678 self.assertIn(str(signal.SIGABRT), error_string) 1679 1680 def test_CalledProcessError_str_unknown_signal(self): 1681 err = subprocess.CalledProcessError(-9876543, "fake cmd") 1682 error_string = str(err) 1683 self.assertIn("unknown signal 9876543.", error_string) 1684 1685 def test_CalledProcessError_str_non_zero(self): 1686 err = subprocess.CalledProcessError(2, "fake cmd") 1687 error_string = str(err) 1688 self.assertIn("non-zero exit status 2.", error_string) 1689 1690 def test_preexec(self): 1691 # DISCLAIMER: Setting environment variables is *not* a good use 1692 # of a preexec_fn. This is merely a test. 1693 p = subprocess.Popen([sys.executable, "-c", 1694 'import sys,os;' 1695 'sys.stdout.write(os.getenv("FRUIT"))'], 1696 stdout=subprocess.PIPE, 1697 preexec_fn=lambda: os.putenv("FRUIT", "apple")) 1698 with p: 1699 self.assertEqual(p.stdout.read(), b"apple") 1700 1701 def test_preexec_exception(self): 1702 def raise_it(): 1703 raise ValueError("What if two swallows carried a coconut?") 1704 try: 1705 p = subprocess.Popen([sys.executable, "-c", ""], 1706 preexec_fn=raise_it) 1707 except subprocess.SubprocessError as e: 1708 self.assertTrue( 1709 subprocess._posixsubprocess, 1710 "Expected a ValueError from the preexec_fn") 1711 except ValueError as e: 1712 self.assertIn("coconut", e.args[0]) 1713 else: 1714 self.fail("Exception raised by preexec_fn did not make it " 1715 "to the parent process.") 1716 1717 class _TestExecuteChildPopen(subprocess.Popen): 1718 """Used to test behavior at the end of _execute_child.""" 1719 def __init__(self, testcase, *args, **kwargs): 1720 self._testcase = testcase 1721 subprocess.Popen.__init__(self, *args, **kwargs) 1722 1723 def _execute_child(self, *args, **kwargs): 1724 try: 1725 subprocess.Popen._execute_child(self, *args, **kwargs) 1726 finally: 1727 # Open a bunch of file descriptors and verify that 1728 # none of them are the same as the ones the Popen 1729 # instance is using for stdin/stdout/stderr. 1730 devzero_fds = [os.open("/dev/zero", os.O_RDONLY) 1731 for _ in range(8)] 1732 try: 1733 for fd in devzero_fds: 1734 self._testcase.assertNotIn( 1735 fd, (self.stdin.fileno(), self.stdout.fileno(), 1736 self.stderr.fileno()), 1737 msg="At least one fd was closed early.") 1738 finally: 1739 for fd in devzero_fds: 1740 os.close(fd) 1741 1742 @unittest.skipIf(not os.path.exists("/dev/zero"), "/dev/zero required.") 1743 def test_preexec_errpipe_does_not_double_close_pipes(self): 1744 """Issue16140: Don't double close pipes on preexec error.""" 1745 1746 def raise_it(): 1747 raise subprocess.SubprocessError( 1748 "force the _execute_child() errpipe_data path.") 1749 1750 with self.assertRaises(subprocess.SubprocessError): 1751 self._TestExecuteChildPopen( 1752 self, [sys.executable, "-c", "pass"], 1753 stdin=subprocess.PIPE, stdout=subprocess.PIPE, 1754 stderr=subprocess.PIPE, preexec_fn=raise_it) 1755 1756 def test_preexec_gc_module_failure(self): 1757 # This tests the code that disables garbage collection if the child 1758 # process will execute any Python. 1759 def raise_runtime_error(): 1760 raise RuntimeError("this shouldn't escape") 1761 enabled = gc.isenabled() 1762 orig_gc_disable = gc.disable 1763 orig_gc_isenabled = gc.isenabled 1764 try: 1765 gc.disable() 1766 self.assertFalse(gc.isenabled()) 1767 subprocess.call([sys.executable, '-c', ''], 1768 preexec_fn=lambda: None) 1769 self.assertFalse(gc.isenabled(), 1770 "Popen enabled gc when it shouldn't.") 1771 1772 gc.enable() 1773 self.assertTrue(gc.isenabled()) 1774 subprocess.call([sys.executable, '-c', ''], 1775 preexec_fn=lambda: None) 1776 self.assertTrue(gc.isenabled(), "Popen left gc disabled.") 1777 1778 gc.disable = raise_runtime_error 1779 self.assertRaises(RuntimeError, subprocess.Popen, 1780 [sys.executable, '-c', ''], 1781 preexec_fn=lambda: None) 1782 1783 del gc.isenabled # force an AttributeError 1784 self.assertRaises(AttributeError, subprocess.Popen, 1785 [sys.executable, '-c', ''], 1786 preexec_fn=lambda: None) 1787 finally: 1788 gc.disable = orig_gc_disable 1789 gc.isenabled = orig_gc_isenabled 1790 if not enabled: 1791 gc.disable() 1792 1793 @unittest.skipIf( 1794 sys.platform == 'darwin', 'setrlimit() seems to fail on OS X') 1795 def test_preexec_fork_failure(self): 1796 # The internal code did not preserve the previous exception when 1797 # re-enabling garbage collection 1798 try: 1799 from resource import getrlimit, setrlimit, RLIMIT_NPROC 1800 except ImportError as err: 1801 self.skipTest(err) # RLIMIT_NPROC is specific to Linux and BSD 1802 limits = getrlimit(RLIMIT_NPROC) 1803 [_, hard] = limits 1804 setrlimit(RLIMIT_NPROC, (0, hard)) 1805 self.addCleanup(setrlimit, RLIMIT_NPROC, limits) 1806 try: 1807 subprocess.call([sys.executable, '-c', ''], 1808 preexec_fn=lambda: None) 1809 except BlockingIOError: 1810 # Forking should raise EAGAIN, translated to BlockingIOError 1811 pass 1812 else: 1813 self.skipTest('RLIMIT_NPROC had no effect; probably superuser') 1814 1815 def test_args_string(self): 1816 # args is a string 1817 fd, fname = tempfile.mkstemp() 1818 # reopen in text mode 1819 with open(fd, "w", errors="surrogateescape") as fobj: 1820 fobj.write("#!%s\n" % support.unix_shell) 1821 fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" % 1822 sys.executable) 1823 os.chmod(fname, 0o700) 1824 p = subprocess.Popen(fname) 1825 p.wait() 1826 os.remove(fname) 1827 self.assertEqual(p.returncode, 47) 1828 1829 def test_invalid_args(self): 1830 # invalid arguments should raise ValueError 1831 self.assertRaises(ValueError, subprocess.call, 1832 [sys.executable, "-c", 1833 "import sys; sys.exit(47)"], 1834 startupinfo=47) 1835 self.assertRaises(ValueError, subprocess.call, 1836 [sys.executable, "-c", 1837 "import sys; sys.exit(47)"], 1838 creationflags=47) 1839 1840 def test_shell_sequence(self): 1841 # Run command through the shell (sequence) 1842 newenv = os.environ.copy() 1843 newenv["FRUIT"] = "apple" 1844 p = subprocess.Popen(["echo $FRUIT"], shell=1, 1845 stdout=subprocess.PIPE, 1846 env=newenv) 1847 with p: 1848 self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple") 1849 1850 def test_shell_string(self): 1851 # Run command through the shell (string) 1852 newenv = os.environ.copy() 1853 newenv["FRUIT"] = "apple" 1854 p = subprocess.Popen("echo $FRUIT", shell=1, 1855 stdout=subprocess.PIPE, 1856 env=newenv) 1857 with p: 1858 self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple") 1859 1860 def test_call_string(self): 1861 # call() function with string argument on UNIX 1862 fd, fname = tempfile.mkstemp() 1863 # reopen in text mode 1864 with open(fd, "w", errors="surrogateescape") as fobj: 1865 fobj.write("#!%s\n" % support.unix_shell) 1866 fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" % 1867 sys.executable) 1868 os.chmod(fname, 0o700) 1869 rc = subprocess.call(fname) 1870 os.remove(fname) 1871 self.assertEqual(rc, 47) 1872 1873 def test_specific_shell(self): 1874 # Issue #9265: Incorrect name passed as arg[0]. 1875 shells = [] 1876 for prefix in ['/bin', '/usr/bin/', '/usr/local/bin']: 1877 for name in ['bash', 'ksh']: 1878 sh = os.path.join(prefix, name) 1879 if os.path.isfile(sh): 1880 shells.append(sh) 1881 if not shells: # Will probably work for any shell but csh. 1882 self.skipTest("bash or ksh required for this test") 1883 sh = '/bin/sh' 1884 if os.path.isfile(sh) and not os.path.islink(sh): 1885 # Test will fail if /bin/sh is a symlink to csh. 1886 shells.append(sh) 1887 for sh in shells: 1888 p = subprocess.Popen("echo $0", executable=sh, shell=True, 1889 stdout=subprocess.PIPE) 1890 with p: 1891 self.assertEqual(p.stdout.read().strip(), bytes(sh, 'ascii')) 1892 1893 def _kill_process(self, method, *args): 1894 # Do not inherit file handles from the parent. 1895 # It should fix failures on some platforms. 1896 # Also set the SIGINT handler to the default to make sure it's not 1897 # being ignored (some tests rely on that.) 1898 old_handler = signal.signal(signal.SIGINT, signal.default_int_handler) 1899 try: 1900 p = subprocess.Popen([sys.executable, "-c", """if 1: 1901 import sys, time 1902 sys.stdout.write('x\\n') 1903 sys.stdout.flush() 1904 time.sleep(30) 1905 """], 1906 close_fds=True, 1907 stdin=subprocess.PIPE, 1908 stdout=subprocess.PIPE, 1909 stderr=subprocess.PIPE) 1910 finally: 1911 signal.signal(signal.SIGINT, old_handler) 1912 # Wait for the interpreter to be completely initialized before 1913 # sending any signal. 1914 p.stdout.read(1) 1915 getattr(p, method)(*args) 1916 return p 1917 1918 @unittest.skipIf(sys.platform.startswith(('netbsd', 'openbsd')), 1919 "Due to known OS bug (issue #16762)") 1920 def _kill_dead_process(self, method, *args): 1921 # Do not inherit file handles from the parent. 1922 # It should fix failures on some platforms. 1923 p = subprocess.Popen([sys.executable, "-c", """if 1: 1924 import sys, time 1925 sys.stdout.write('x\\n') 1926 sys.stdout.flush() 1927 """], 1928 close_fds=True, 1929 stdin=subprocess.PIPE, 1930 stdout=subprocess.PIPE, 1931 stderr=subprocess.PIPE) 1932 # Wait for the interpreter to be completely initialized before 1933 # sending any signal. 1934 p.stdout.read(1) 1935 # The process should end after this 1936 time.sleep(1) 1937 # This shouldn't raise even though the child is now dead 1938 getattr(p, method)(*args) 1939 p.communicate() 1940 1941 def test_send_signal(self): 1942 p = self._kill_process('send_signal', signal.SIGINT) 1943 _, stderr = p.communicate() 1944 self.assertIn(b'KeyboardInterrupt', stderr) 1945 self.assertNotEqual(p.wait(), 0) 1946 1947 def test_kill(self): 1948 p = self._kill_process('kill') 1949 _, stderr = p.communicate() 1950 self.assertStderrEqual(stderr, b'') 1951 self.assertEqual(p.wait(), -signal.SIGKILL) 1952 1953 def test_terminate(self): 1954 p = self._kill_process('terminate') 1955 _, stderr = p.communicate() 1956 self.assertStderrEqual(stderr, b'') 1957 self.assertEqual(p.wait(), -signal.SIGTERM) 1958 1959 def test_send_signal_dead(self): 1960 # Sending a signal to a dead process 1961 self._kill_dead_process('send_signal', signal.SIGINT) 1962 1963 def test_kill_dead(self): 1964 # Killing a dead process 1965 self._kill_dead_process('kill') 1966 1967 def test_terminate_dead(self): 1968 # Terminating a dead process 1969 self._kill_dead_process('terminate') 1970 1971 def _save_fds(self, save_fds): 1972 fds = [] 1973 for fd in save_fds: 1974 inheritable = os.get_inheritable(fd) 1975 saved = os.dup(fd) 1976 fds.append((fd, saved, inheritable)) 1977 return fds 1978 1979 def _restore_fds(self, fds): 1980 for fd, saved, inheritable in fds: 1981 os.dup2(saved, fd, inheritable=inheritable) 1982 os.close(saved) 1983 1984 def check_close_std_fds(self, fds): 1985 # Issue #9905: test that subprocess pipes still work properly with 1986 # some standard fds closed 1987 stdin = 0 1988 saved_fds = self._save_fds(fds) 1989 for fd, saved, inheritable in saved_fds: 1990 if fd == 0: 1991 stdin = saved 1992 break 1993 try: 1994 for fd in fds: 1995 os.close(fd) 1996 out, err = subprocess.Popen([sys.executable, "-c", 1997 'import sys;' 1998 'sys.stdout.write("apple");' 1999 'sys.stdout.flush();' 2000 'sys.stderr.write("orange")'], 2001 stdin=stdin, 2002 stdout=subprocess.PIPE, 2003 stderr=subprocess.PIPE).communicate() 2004 err = support.strip_python_stderr(err) 2005 self.assertEqual((out, err), (b'apple', b'orange')) 2006 finally: 2007 self._restore_fds(saved_fds) 2008 2009 def test_close_fd_0(self): 2010 self.check_close_std_fds([0]) 2011 2012 def test_close_fd_1(self): 2013 self.check_close_std_fds([1]) 2014 2015 def test_close_fd_2(self): 2016 self.check_close_std_fds([2]) 2017 2018 def test_close_fds_0_1(self): 2019 self.check_close_std_fds([0, 1]) 2020 2021 def test_close_fds_0_2(self): 2022 self.check_close_std_fds([0, 2]) 2023 2024 def test_close_fds_1_2(self): 2025 self.check_close_std_fds([1, 2]) 2026 2027 def test_close_fds_0_1_2(self): 2028 # Issue #10806: test that subprocess pipes still work properly with 2029 # all standard fds closed. 2030 self.check_close_std_fds([0, 1, 2]) 2031 2032 def test_small_errpipe_write_fd(self): 2033 """Issue #15798: Popen should work when stdio fds are available.""" 2034 new_stdin = os.dup(0) 2035 new_stdout = os.dup(1) 2036 try: 2037 os.close(0) 2038 os.close(1) 2039 2040 # Side test: if errpipe_write fails to have its CLOEXEC 2041 # flag set this should cause the parent to think the exec 2042 # failed. Extremely unlikely: everyone supports CLOEXEC. 2043 subprocess.Popen([ 2044 sys.executable, "-c", 2045 "print('AssertionError:0:CLOEXEC failure.')"]).wait() 2046 finally: 2047 # Restore original stdin and stdout 2048 os.dup2(new_stdin, 0) 2049 os.dup2(new_stdout, 1) 2050 os.close(new_stdin) 2051 os.close(new_stdout) 2052 2053 def test_remapping_std_fds(self): 2054 # open up some temporary files 2055 temps = [tempfile.mkstemp() for i in range(3)] 2056 try: 2057 temp_fds = [fd for fd, fname in temps] 2058 2059 # unlink the files -- we won't need to reopen them 2060 for fd, fname in temps: 2061 os.unlink(fname) 2062 2063 # write some data to what will become stdin, and rewind 2064 os.write(temp_fds[1], b"STDIN") 2065 os.lseek(temp_fds[1], 0, 0) 2066 2067 # move the standard file descriptors out of the way 2068 saved_fds = self._save_fds(range(3)) 2069 try: 2070 # duplicate the file objects over the standard fd's 2071 for fd, temp_fd in enumerate(temp_fds): 2072 os.dup2(temp_fd, fd) 2073 2074 # now use those files in the "wrong" order, so that subprocess 2075 # has to rearrange them in the child 2076 p = subprocess.Popen([sys.executable, "-c", 2077 'import sys; got = sys.stdin.read();' 2078 'sys.stdout.write("got %s"%got); sys.stderr.write("err")'], 2079 stdin=temp_fds[1], 2080 stdout=temp_fds[2], 2081 stderr=temp_fds[0]) 2082 p.wait() 2083 finally: 2084 self._restore_fds(saved_fds) 2085 2086 for fd in temp_fds: 2087 os.lseek(fd, 0, 0) 2088 2089 out = os.read(temp_fds[2], 1024) 2090 err = support.strip_python_stderr(os.read(temp_fds[0], 1024)) 2091 self.assertEqual(out, b"got STDIN") 2092 self.assertEqual(err, b"err") 2093 2094 finally: 2095 for fd in temp_fds: 2096 os.close(fd) 2097 2098 def check_swap_fds(self, stdin_no, stdout_no, stderr_no): 2099 # open up some temporary files 2100 temps = [tempfile.mkstemp() for i in range(3)] 2101 temp_fds = [fd for fd, fname in temps] 2102 try: 2103 # unlink the files -- we won't need to reopen them 2104 for fd, fname in temps: 2105 os.unlink(fname) 2106 2107 # save a copy of the standard file descriptors 2108 saved_fds = self._save_fds(range(3)) 2109 try: 2110 # duplicate the temp files over the standard fd's 0, 1, 2 2111 for fd, temp_fd in enumerate(temp_fds): 2112 os.dup2(temp_fd, fd) 2113 2114 # write some data to what will become stdin, and rewind 2115 os.write(stdin_no, b"STDIN") 2116 os.lseek(stdin_no, 0, 0) 2117 2118 # now use those files in the given order, so that subprocess 2119 # has to rearrange them in the child 2120 p = subprocess.Popen([sys.executable, "-c", 2121 'import sys; got = sys.stdin.read();' 2122 'sys.stdout.write("got %s"%got); sys.stderr.write("err")'], 2123 stdin=stdin_no, 2124 stdout=stdout_no, 2125 stderr=stderr_no) 2126 p.wait() 2127 2128 for fd in temp_fds: 2129 os.lseek(fd, 0, 0) 2130 2131 out = os.read(stdout_no, 1024) 2132 err = support.strip_python_stderr(os.read(stderr_no, 1024)) 2133 finally: 2134 self._restore_fds(saved_fds) 2135 2136 self.assertEqual(out, b"got STDIN") 2137 self.assertEqual(err, b"err") 2138 2139 finally: 2140 for fd in temp_fds: 2141 os.close(fd) 2142 2143 # When duping fds, if there arises a situation where one of the fds is 2144 # either 0, 1 or 2, it is possible that it is overwritten (#12607). 2145 # This tests all combinations of this. 2146 def test_swap_fds(self): 2147 self.check_swap_fds(0, 1, 2) 2148 self.check_swap_fds(0, 2, 1) 2149 self.check_swap_fds(1, 0, 2) 2150 self.check_swap_fds(1, 2, 0) 2151 self.check_swap_fds(2, 0, 1) 2152 self.check_swap_fds(2, 1, 0) 2153 2154 def _check_swap_std_fds_with_one_closed(self, from_fds, to_fds): 2155 saved_fds = self._save_fds(range(3)) 2156 try: 2157 for from_fd in from_fds: 2158 with tempfile.TemporaryFile() as f: 2159 os.dup2(f.fileno(), from_fd) 2160 2161 fd_to_close = (set(range(3)) - set(from_fds)).pop() 2162 os.close(fd_to_close) 2163 2164 arg_names = ['stdin', 'stdout', 'stderr'] 2165 kwargs = {} 2166 for from_fd, to_fd in zip(from_fds, to_fds): 2167 kwargs[arg_names[to_fd]] = from_fd 2168 2169 code = textwrap.dedent(r''' 2170 import os, sys 2171 skipped_fd = int(sys.argv[1]) 2172 for fd in range(3): 2173 if fd != skipped_fd: 2174 os.write(fd, str(fd).encode('ascii')) 2175 ''') 2176 2177 skipped_fd = (set(range(3)) - set(to_fds)).pop() 2178 2179 rc = subprocess.call([sys.executable, '-c', code, str(skipped_fd)], 2180 **kwargs) 2181 self.assertEqual(rc, 0) 2182 2183 for from_fd, to_fd in zip(from_fds, to_fds): 2184 os.lseek(from_fd, 0, os.SEEK_SET) 2185 read_bytes = os.read(from_fd, 1024) 2186 read_fds = list(map(int, read_bytes.decode('ascii'))) 2187 msg = textwrap.dedent(f""" 2188 When testing {from_fds} to {to_fds} redirection, 2189 parent descriptor {from_fd} got redirected 2190 to descriptor(s) {read_fds} instead of descriptor {to_fd}. 2191 """) 2192 self.assertEqual([to_fd], read_fds, msg) 2193 finally: 2194 self._restore_fds(saved_fds) 2195 2196 # Check that subprocess can remap std fds correctly even 2197 # if one of them is closed (#32844). 2198 def test_swap_std_fds_with_one_closed(self): 2199 for from_fds in itertools.combinations(range(3), 2): 2200 for to_fds in itertools.permutations(range(3), 2): 2201 self._check_swap_std_fds_with_one_closed(from_fds, to_fds) 2202 2203 def test_surrogates_error_message(self): 2204 def prepare(): 2205 raise ValueError("surrogate:\uDCff") 2206 2207 try: 2208 subprocess.call( 2209 [sys.executable, "-c", "pass"], 2210 preexec_fn=prepare) 2211 except ValueError as err: 2212 # Pure Python implementations keeps the message 2213 self.assertIsNone(subprocess._posixsubprocess) 2214 self.assertEqual(str(err), "surrogate:\uDCff") 2215 except subprocess.SubprocessError as err: 2216 # _posixsubprocess uses a default message 2217 self.assertIsNotNone(subprocess._posixsubprocess) 2218 self.assertEqual(str(err), "Exception occurred in preexec_fn.") 2219 else: 2220 self.fail("Expected ValueError or subprocess.SubprocessError") 2221 2222 def test_undecodable_env(self): 2223 for key, value in (('test', 'abc\uDCFF'), ('test\uDCFF', '42')): 2224 encoded_value = value.encode("ascii", "surrogateescape") 2225 2226 # test str with surrogates 2227 script = "import os; print(ascii(os.getenv(%s)))" % repr(key) 2228 env = os.environ.copy() 2229 env[key] = value 2230 # Use C locale to get ASCII for the locale encoding to force 2231 # surrogate-escaping of \xFF in the child process 2232 env['LC_ALL'] = 'C' 2233 decoded_value = value 2234 stdout = subprocess.check_output( 2235 [sys.executable, "-c", script], 2236 env=env) 2237 stdout = stdout.rstrip(b'\n\r') 2238 self.assertEqual(stdout.decode('ascii'), ascii(decoded_value)) 2239 2240 # test bytes 2241 key = key.encode("ascii", "surrogateescape") 2242 script = "import os; print(ascii(os.getenvb(%s)))" % repr(key) 2243 env = os.environ.copy() 2244 env[key] = encoded_value 2245 stdout = subprocess.check_output( 2246 [sys.executable, "-c", script], 2247 env=env) 2248 stdout = stdout.rstrip(b'\n\r') 2249 self.assertEqual(stdout.decode('ascii'), ascii(encoded_value)) 2250 2251 def test_bytes_program(self): 2252 abs_program = os.fsencode(sys.executable) 2253 path, program = os.path.split(sys.executable) 2254 program = os.fsencode(program) 2255 2256 # absolute bytes path 2257 exitcode = subprocess.call([abs_program, "-c", "pass"]) 2258 self.assertEqual(exitcode, 0) 2259 2260 # absolute bytes path as a string 2261 cmd = b"'" + abs_program + b"' -c pass" 2262 exitcode = subprocess.call(cmd, shell=True) 2263 self.assertEqual(exitcode, 0) 2264 2265 # bytes program, unicode PATH 2266 env = os.environ.copy() 2267 env["PATH"] = path 2268 exitcode = subprocess.call([program, "-c", "pass"], env=env) 2269 self.assertEqual(exitcode, 0) 2270 2271 # bytes program, bytes PATH 2272 envb = os.environb.copy() 2273 envb[b"PATH"] = os.fsencode(path) 2274 exitcode = subprocess.call([program, "-c", "pass"], env=envb) 2275 self.assertEqual(exitcode, 0) 2276 2277 def test_pipe_cloexec(self): 2278 sleeper = support.findfile("input_reader.py", subdir="subprocessdata") 2279 fd_status = support.findfile("fd_status.py", subdir="subprocessdata") 2280 2281 p1 = subprocess.Popen([sys.executable, sleeper], 2282 stdin=subprocess.PIPE, stdout=subprocess.PIPE, 2283 stderr=subprocess.PIPE, close_fds=False) 2284 2285 self.addCleanup(p1.communicate, b'') 2286 2287 p2 = subprocess.Popen([sys.executable, fd_status], 2288 stdout=subprocess.PIPE, close_fds=False) 2289 2290 output, error = p2.communicate() 2291 result_fds = set(map(int, output.split(b','))) 2292 unwanted_fds = set([p1.stdin.fileno(), p1.stdout.fileno(), 2293 p1.stderr.fileno()]) 2294 2295 self.assertFalse(result_fds & unwanted_fds, 2296 "Expected no fds from %r to be open in child, " 2297 "found %r" % 2298 (unwanted_fds, result_fds & unwanted_fds)) 2299 2300 def test_pipe_cloexec_real_tools(self): 2301 qcat = support.findfile("qcat.py", subdir="subprocessdata") 2302 qgrep = support.findfile("qgrep.py", subdir="subprocessdata") 2303 2304 subdata = b'zxcvbn' 2305 data = subdata * 4 + b'\n' 2306 2307 p1 = subprocess.Popen([sys.executable, qcat], 2308 stdin=subprocess.PIPE, stdout=subprocess.PIPE, 2309 close_fds=False) 2310 2311 p2 = subprocess.Popen([sys.executable, qgrep, subdata], 2312 stdin=p1.stdout, stdout=subprocess.PIPE, 2313 close_fds=False) 2314 2315 self.addCleanup(p1.wait) 2316 self.addCleanup(p2.wait) 2317 def kill_p1(): 2318 try: 2319 p1.terminate() 2320 except ProcessLookupError: 2321 pass 2322 def kill_p2(): 2323 try: 2324 p2.terminate() 2325 except ProcessLookupError: 2326 pass 2327 self.addCleanup(kill_p1) 2328 self.addCleanup(kill_p2) 2329 2330 p1.stdin.write(data) 2331 p1.stdin.close() 2332 2333 readfiles, ignored1, ignored2 = select.select([p2.stdout], [], [], 10) 2334 2335 self.assertTrue(readfiles, "The child hung") 2336 self.assertEqual(p2.stdout.read(), data) 2337 2338 p1.stdout.close() 2339 p2.stdout.close() 2340 2341 def test_close_fds(self): 2342 fd_status = support.findfile("fd_status.py", subdir="subprocessdata") 2343 2344 fds = os.pipe() 2345 self.addCleanup(os.close, fds[0]) 2346 self.addCleanup(os.close, fds[1]) 2347 2348 open_fds = set(fds) 2349 # add a bunch more fds 2350 for _ in range(9): 2351 fd = os.open(os.devnull, os.O_RDONLY) 2352 self.addCleanup(os.close, fd) 2353 open_fds.add(fd) 2354 2355 for fd in open_fds: 2356 os.set_inheritable(fd, True) 2357 2358 p = subprocess.Popen([sys.executable, fd_status], 2359 stdout=subprocess.PIPE, close_fds=False) 2360 output, ignored = p.communicate() 2361 remaining_fds = set(map(int, output.split(b','))) 2362 2363 self.assertEqual(remaining_fds & open_fds, open_fds, 2364 "Some fds were closed") 2365 2366 p = subprocess.Popen([sys.executable, fd_status], 2367 stdout=subprocess.PIPE, close_fds=True) 2368 output, ignored = p.communicate() 2369 remaining_fds = set(map(int, output.split(b','))) 2370 2371 self.assertFalse(remaining_fds & open_fds, 2372 "Some fds were left open") 2373 self.assertIn(1, remaining_fds, "Subprocess failed") 2374 2375 # Keep some of the fd's we opened open in the subprocess. 2376 # This tests _posixsubprocess.c's proper handling of fds_to_keep. 2377 fds_to_keep = set(open_fds.pop() for _ in range(8)) 2378 p = subprocess.Popen([sys.executable, fd_status], 2379 stdout=subprocess.PIPE, close_fds=True, 2380 pass_fds=fds_to_keep) 2381 output, ignored = p.communicate() 2382 remaining_fds = set(map(int, output.split(b','))) 2383 2384 self.assertFalse((remaining_fds - fds_to_keep) & open_fds, 2385 "Some fds not in pass_fds were left open") 2386 self.assertIn(1, remaining_fds, "Subprocess failed") 2387 2388 2389 @unittest.skipIf(sys.platform.startswith("freebsd") and 2390 os.stat("/dev").st_dev == os.stat("/dev/fd").st_dev, 2391 "Requires fdescfs mounted on /dev/fd on FreeBSD.") 2392 def test_close_fds_when_max_fd_is_lowered(self): 2393 """Confirm that issue21618 is fixed (may fail under valgrind).""" 2394 fd_status = support.findfile("fd_status.py", subdir="subprocessdata") 2395 2396 # This launches the meat of the test in a child process to 2397 # avoid messing with the larger unittest processes maximum 2398 # number of file descriptors. 2399 # This process launches: 2400 # +--> Process that lowers its RLIMIT_NOFILE aftr setting up 2401 # a bunch of high open fds above the new lower rlimit. 2402 # Those are reported via stdout before launching a new 2403 # process with close_fds=False to run the actual test: 2404 # +--> The TEST: This one launches a fd_status.py 2405 # subprocess with close_fds=True so we can find out if 2406 # any of the fds above the lowered rlimit are still open. 2407 p = subprocess.Popen([sys.executable, '-c', textwrap.dedent( 2408 ''' 2409 import os, resource, subprocess, sys, textwrap 2410 open_fds = set() 2411 # Add a bunch more fds to pass down. 2412 for _ in range(40): 2413 fd = os.open(os.devnull, os.O_RDONLY) 2414 open_fds.add(fd) 2415 2416 # Leave a two pairs of low ones available for use by the 2417 # internal child error pipe and the stdout pipe. 2418 # We also leave 10 more open as some Python buildbots run into 2419 # "too many open files" errors during the test if we do not. 2420 for fd in sorted(open_fds)[:14]: 2421 os.close(fd) 2422 open_fds.remove(fd) 2423 2424 for fd in open_fds: 2425 #self.addCleanup(os.close, fd) 2426 os.set_inheritable(fd, True) 2427 2428 max_fd_open = max(open_fds) 2429 2430 # Communicate the open_fds to the parent unittest.TestCase process. 2431 print(','.join(map(str, sorted(open_fds)))) 2432 sys.stdout.flush() 2433 2434 rlim_cur, rlim_max = resource.getrlimit(resource.RLIMIT_NOFILE) 2435 try: 2436 # 29 is lower than the highest fds we are leaving open. 2437 resource.setrlimit(resource.RLIMIT_NOFILE, (29, rlim_max)) 2438 # Launch a new Python interpreter with our low fd rlim_cur that 2439 # inherits open fds above that limit. It then uses subprocess 2440 # with close_fds=True to get a report of open fds in the child. 2441 # An explicit list of fds to check is passed to fd_status.py as 2442 # letting fd_status rely on its default logic would miss the 2443 # fds above rlim_cur as it normally only checks up to that limit. 2444 subprocess.Popen( 2445 [sys.executable, '-c', 2446 textwrap.dedent(""" 2447 import subprocess, sys 2448 subprocess.Popen([sys.executable, %r] + 2449 [str(x) for x in range({max_fd})], 2450 close_fds=True).wait() 2451 """.format(max_fd=max_fd_open+1))], 2452 close_fds=False).wait() 2453 finally: 2454 resource.setrlimit(resource.RLIMIT_NOFILE, (rlim_cur, rlim_max)) 2455 ''' % fd_status)], stdout=subprocess.PIPE) 2456 2457 output, unused_stderr = p.communicate() 2458 output_lines = output.splitlines() 2459 self.assertEqual(len(output_lines), 2, 2460 msg="expected exactly two lines of output:\n%r" % output) 2461 opened_fds = set(map(int, output_lines[0].strip().split(b','))) 2462 remaining_fds = set(map(int, output_lines[1].strip().split(b','))) 2463 2464 self.assertFalse(remaining_fds & opened_fds, 2465 msg="Some fds were left open.") 2466 2467 2468 # Mac OS X Tiger (10.4) has a kernel bug: sometimes, the file 2469 # descriptor of a pipe closed in the parent process is valid in the 2470 # child process according to fstat(), but the mode of the file 2471 # descriptor is invalid, and read or write raise an error. 2472 @support.requires_mac_ver(10, 5) 2473 def test_pass_fds(self): 2474 fd_status = support.findfile("fd_status.py", subdir="subprocessdata") 2475 2476 open_fds = set() 2477 2478 for x in range(5): 2479 fds = os.pipe() 2480 self.addCleanup(os.close, fds[0]) 2481 self.addCleanup(os.close, fds[1]) 2482 os.set_inheritable(fds[0], True) 2483 os.set_inheritable(fds[1], True) 2484 open_fds.update(fds) 2485 2486 for fd in open_fds: 2487 p = subprocess.Popen([sys.executable, fd_status], 2488 stdout=subprocess.PIPE, close_fds=True, 2489 pass_fds=(fd, )) 2490 output, ignored = p.communicate() 2491 2492 remaining_fds = set(map(int, output.split(b','))) 2493 to_be_closed = open_fds - {fd} 2494 2495 self.assertIn(fd, remaining_fds, "fd to be passed not passed") 2496 self.assertFalse(remaining_fds & to_be_closed, 2497 "fd to be closed passed") 2498 2499 # pass_fds overrides close_fds with a warning. 2500 with self.assertWarns(RuntimeWarning) as context: 2501 self.assertFalse(subprocess.call( 2502 [sys.executable, "-c", "import sys; sys.exit(0)"], 2503 close_fds=False, pass_fds=(fd, ))) 2504 self.assertIn('overriding close_fds', str(context.warning)) 2505 2506 def test_pass_fds_inheritable(self): 2507 script = support.findfile("fd_status.py", subdir="subprocessdata") 2508 2509 inheritable, non_inheritable = os.pipe() 2510 self.addCleanup(os.close, inheritable) 2511 self.addCleanup(os.close, non_inheritable) 2512 os.set_inheritable(inheritable, True) 2513 os.set_inheritable(non_inheritable, False) 2514 pass_fds = (inheritable, non_inheritable) 2515 args = [sys.executable, script] 2516 args += list(map(str, pass_fds)) 2517 2518 p = subprocess.Popen(args, 2519 stdout=subprocess.PIPE, close_fds=True, 2520 pass_fds=pass_fds) 2521 output, ignored = p.communicate() 2522 fds = set(map(int, output.split(b','))) 2523 2524 # the inheritable file descriptor must be inherited, so its inheritable 2525 # flag must be set in the child process after fork() and before exec() 2526 self.assertEqual(fds, set(pass_fds), "output=%a" % output) 2527 2528 # inheritable flag must not be changed in the parent process 2529 self.assertEqual(os.get_inheritable(inheritable), True) 2530 self.assertEqual(os.get_inheritable(non_inheritable), False) 2531 2532 2533 # bpo-32270: Ensure that descriptors specified in pass_fds 2534 # are inherited even if they are used in redirections. 2535 # Contributed by @izbyshev. 2536 def test_pass_fds_redirected(self): 2537 """Regression test for https://bugs.python.org/issue32270.""" 2538 fd_status = support.findfile("fd_status.py", subdir="subprocessdata") 2539 pass_fds = [] 2540 for _ in range(2): 2541 fd = os.open(os.devnull, os.O_RDWR) 2542 self.addCleanup(os.close, fd) 2543 pass_fds.append(fd) 2544 2545 stdout_r, stdout_w = os.pipe() 2546 self.addCleanup(os.close, stdout_r) 2547 self.addCleanup(os.close, stdout_w) 2548 pass_fds.insert(1, stdout_w) 2549 2550 with subprocess.Popen([sys.executable, fd_status], 2551 stdin=pass_fds[0], 2552 stdout=pass_fds[1], 2553 stderr=pass_fds[2], 2554 close_fds=True, 2555 pass_fds=pass_fds): 2556 output = os.read(stdout_r, 1024) 2557 fds = {int(num) for num in output.split(b',')} 2558 2559 self.assertEqual(fds, {0, 1, 2} | frozenset(pass_fds), f"output={output!a}") 2560 2561 2562 def test_stdout_stdin_are_single_inout_fd(self): 2563 with io.open(os.devnull, "r+") as inout: 2564 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"], 2565 stdout=inout, stdin=inout) 2566 p.wait() 2567 2568 def test_stdout_stderr_are_single_inout_fd(self): 2569 with io.open(os.devnull, "r+") as inout: 2570 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"], 2571 stdout=inout, stderr=inout) 2572 p.wait() 2573 2574 def test_stderr_stdin_are_single_inout_fd(self): 2575 with io.open(os.devnull, "r+") as inout: 2576 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"], 2577 stderr=inout, stdin=inout) 2578 p.wait() 2579 2580 def test_wait_when_sigchild_ignored(self): 2581 # NOTE: sigchild_ignore.py may not be an effective test on all OSes. 2582 sigchild_ignore = support.findfile("sigchild_ignore.py", 2583 subdir="subprocessdata") 2584 p = subprocess.Popen([sys.executable, sigchild_ignore], 2585 stdout=subprocess.PIPE, stderr=subprocess.PIPE) 2586 stdout, stderr = p.communicate() 2587 self.assertEqual(0, p.returncode, "sigchild_ignore.py exited" 2588 " non-zero with this error:\n%s" % 2589 stderr.decode('utf-8')) 2590 2591 def test_select_unbuffered(self): 2592 # Issue #11459: bufsize=0 should really set the pipes as 2593 # unbuffered (and therefore let select() work properly). 2594 select = support.import_module("select") 2595 p = subprocess.Popen([sys.executable, "-c", 2596 'import sys;' 2597 'sys.stdout.write("apple")'], 2598 stdout=subprocess.PIPE, 2599 bufsize=0) 2600 f = p.stdout 2601 self.addCleanup(f.close) 2602 try: 2603 self.assertEqual(f.read(4), b"appl") 2604 self.assertIn(f, select.select([f], [], [], 0.0)[0]) 2605 finally: 2606 p.wait() 2607 2608 def test_zombie_fast_process_del(self): 2609 # Issue #12650: on Unix, if Popen.__del__() was called before the 2610 # process exited, it wouldn't be added to subprocess._active, and would 2611 # remain a zombie. 2612 # spawn a Popen, and delete its reference before it exits 2613 p = subprocess.Popen([sys.executable, "-c", 2614 'import sys, time;' 2615 'time.sleep(0.2)'], 2616 stdout=subprocess.PIPE, 2617 stderr=subprocess.PIPE) 2618 self.addCleanup(p.stdout.close) 2619 self.addCleanup(p.stderr.close) 2620 ident = id(p) 2621 pid = p.pid 2622 with support.check_warnings(('', ResourceWarning)): 2623 p = None 2624 2625 # check that p is in the active processes list 2626 self.assertIn(ident, [id(o) for o in subprocess._active]) 2627 2628 def test_leak_fast_process_del_killed(self): 2629 # Issue #12650: on Unix, if Popen.__del__() was called before the 2630 # process exited, and the process got killed by a signal, it would never 2631 # be removed from subprocess._active, which triggered a FD and memory 2632 # leak. 2633 # spawn a Popen, delete its reference and kill it 2634 p = subprocess.Popen([sys.executable, "-c", 2635 'import time;' 2636 'time.sleep(3)'], 2637 stdout=subprocess.PIPE, 2638 stderr=subprocess.PIPE) 2639 self.addCleanup(p.stdout.close) 2640 self.addCleanup(p.stderr.close) 2641 ident = id(p) 2642 pid = p.pid 2643 with support.check_warnings(('', ResourceWarning)): 2644 p = None 2645 2646 os.kill(pid, signal.SIGKILL) 2647 # check that p is in the active processes list 2648 self.assertIn(ident, [id(o) for o in subprocess._active]) 2649 2650 # let some time for the process to exit, and create a new Popen: this 2651 # should trigger the wait() of p 2652 time.sleep(0.2) 2653 with self.assertRaises(OSError): 2654 with subprocess.Popen(NONEXISTING_CMD, 2655 stdout=subprocess.PIPE, 2656 stderr=subprocess.PIPE) as proc: 2657 pass 2658 # p should have been wait()ed on, and removed from the _active list 2659 self.assertRaises(OSError, os.waitpid, pid, 0) 2660 self.assertNotIn(ident, [id(o) for o in subprocess._active]) 2661 2662 def test_close_fds_after_preexec(self): 2663 fd_status = support.findfile("fd_status.py", subdir="subprocessdata") 2664 2665 # this FD is used as dup2() target by preexec_fn, and should be closed 2666 # in the child process 2667 fd = os.dup(1) 2668 self.addCleanup(os.close, fd) 2669 2670 p = subprocess.Popen([sys.executable, fd_status], 2671 stdout=subprocess.PIPE, close_fds=True, 2672 preexec_fn=lambda: os.dup2(1, fd)) 2673 output, ignored = p.communicate() 2674 2675 remaining_fds = set(map(int, output.split(b','))) 2676 2677 self.assertNotIn(fd, remaining_fds) 2678 2679 @support.cpython_only 2680 def test_fork_exec(self): 2681 # Issue #22290: fork_exec() must not crash on memory allocation failure 2682 # or other errors 2683 import _posixsubprocess 2684 gc_enabled = gc.isenabled() 2685 try: 2686 # Use a preexec function and enable the garbage collector 2687 # to force fork_exec() to re-enable the garbage collector 2688 # on error. 2689 func = lambda: None 2690 gc.enable() 2691 2692 for args, exe_list, cwd, env_list in ( 2693 (123, [b"exe"], None, [b"env"]), 2694 ([b"arg"], 123, None, [b"env"]), 2695 ([b"arg"], [b"exe"], 123, [b"env"]), 2696 ([b"arg"], [b"exe"], None, 123), 2697 ): 2698 with self.assertRaises(TypeError): 2699 _posixsubprocess.fork_exec( 2700 args, exe_list, 2701 True, (), cwd, env_list, 2702 -1, -1, -1, -1, 2703 1, 2, 3, 4, 2704 True, True, func) 2705 finally: 2706 if not gc_enabled: 2707 gc.disable() 2708 2709 @support.cpython_only 2710 def test_fork_exec_sorted_fd_sanity_check(self): 2711 # Issue #23564: sanity check the fork_exec() fds_to_keep sanity check. 2712 import _posixsubprocess 2713 class BadInt: 2714 first = True 2715 def __init__(self, value): 2716 self.value = value 2717 def __int__(self): 2718 if self.first: 2719 self.first = False 2720 return self.value 2721 raise ValueError 2722 2723 gc_enabled = gc.isenabled() 2724 try: 2725 gc.enable() 2726 2727 for fds_to_keep in ( 2728 (-1, 2, 3, 4, 5), # Negative number. 2729 ('str', 4), # Not an int. 2730 (18, 23, 42, 2**63), # Out of range. 2731 (5, 4), # Not sorted. 2732 (6, 7, 7, 8), # Duplicate. 2733 (BadInt(1), BadInt(2)), 2734 ): 2735 with self.assertRaises( 2736 ValueError, 2737 msg='fds_to_keep={}'.format(fds_to_keep)) as c: 2738 _posixsubprocess.fork_exec( 2739 [b"false"], [b"false"], 2740 True, fds_to_keep, None, [b"env"], 2741 -1, -1, -1, -1, 2742 1, 2, 3, 4, 2743 True, True, None) 2744 self.assertIn('fds_to_keep', str(c.exception)) 2745 finally: 2746 if not gc_enabled: 2747 gc.disable() 2748 2749 def test_communicate_BrokenPipeError_stdin_close(self): 2750 # By not setting stdout or stderr or a timeout we force the fast path 2751 # that just calls _stdin_write() internally due to our mock. 2752 proc = subprocess.Popen([sys.executable, '-c', 'pass']) 2753 with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin: 2754 mock_proc_stdin.close.side_effect = BrokenPipeError 2755 proc.communicate() # Should swallow BrokenPipeError from close. 2756 mock_proc_stdin.close.assert_called_with() 2757 2758 def test_communicate_BrokenPipeError_stdin_write(self): 2759 # By not setting stdout or stderr or a timeout we force the fast path 2760 # that just calls _stdin_write() internally due to our mock. 2761 proc = subprocess.Popen([sys.executable, '-c', 'pass']) 2762 with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin: 2763 mock_proc_stdin.write.side_effect = BrokenPipeError 2764 proc.communicate(b'stuff') # Should swallow the BrokenPipeError. 2765 mock_proc_stdin.write.assert_called_once_with(b'stuff') 2766 mock_proc_stdin.close.assert_called_once_with() 2767 2768 def test_communicate_BrokenPipeError_stdin_flush(self): 2769 # Setting stdin and stdout forces the ._communicate() code path. 2770 # python -h exits faster than python -c pass (but spams stdout). 2771 proc = subprocess.Popen([sys.executable, '-h'], 2772 stdin=subprocess.PIPE, 2773 stdout=subprocess.PIPE) 2774 with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin, \ 2775 open(os.devnull, 'wb') as dev_null: 2776 mock_proc_stdin.flush.side_effect = BrokenPipeError 2777 # because _communicate registers a selector using proc.stdin... 2778 mock_proc_stdin.fileno.return_value = dev_null.fileno() 2779 # _communicate() should swallow BrokenPipeError from flush. 2780 proc.communicate(b'stuff') 2781 mock_proc_stdin.flush.assert_called_once_with() 2782 2783 def test_communicate_BrokenPipeError_stdin_close_with_timeout(self): 2784 # Setting stdin and stdout forces the ._communicate() code path. 2785 # python -h exits faster than python -c pass (but spams stdout). 2786 proc = subprocess.Popen([sys.executable, '-h'], 2787 stdin=subprocess.PIPE, 2788 stdout=subprocess.PIPE) 2789 with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin: 2790 mock_proc_stdin.close.side_effect = BrokenPipeError 2791 # _communicate() should swallow BrokenPipeError from close. 2792 proc.communicate(timeout=999) 2793 mock_proc_stdin.close.assert_called_once_with() 2794 2795 @unittest.skipUnless(_testcapi is not None 2796 and hasattr(_testcapi, 'W_STOPCODE'), 2797 'need _testcapi.W_STOPCODE') 2798 def test_stopped(self): 2799 """Test wait() behavior when waitpid returns WIFSTOPPED; issue29335.""" 2800 args = [sys.executable, '-c', 'pass'] 2801 proc = subprocess.Popen(args) 2802 2803 # Wait until the real process completes to avoid zombie process 2804 pid = proc.pid 2805 pid, status = os.waitpid(pid, 0) 2806 self.assertEqual(status, 0) 2807 2808 status = _testcapi.W_STOPCODE(3) 2809 with mock.patch('subprocess.os.waitpid', return_value=(pid, status)): 2810 returncode = proc.wait() 2811 2812 self.assertEqual(returncode, -3) 2813 2814 2815 @unittest.skipUnless(mswindows, "Windows specific tests") 2816 class Win32ProcessTestCase(BaseTestCase): 2817 2818 def test_startupinfo(self): 2819 # startupinfo argument 2820 # We uses hardcoded constants, because we do not want to 2821 # depend on win32all. 2822 STARTF_USESHOWWINDOW = 1 2823 SW_MAXIMIZE = 3 2824 startupinfo = subprocess.STARTUPINFO() 2825 startupinfo.dwFlags = STARTF_USESHOWWINDOW 2826 startupinfo.wShowWindow = SW_MAXIMIZE 2827 # Since Python is a console process, it won't be affected 2828 # by wShowWindow, but the argument should be silently 2829 # ignored 2830 subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"], 2831 startupinfo=startupinfo) 2832 2833 def test_startupinfo_keywords(self): 2834 # startupinfo argument 2835 # We use hardcoded constants, because we do not want to 2836 # depend on win32all. 2837 STARTF_USERSHOWWINDOW = 1 2838 SW_MAXIMIZE = 3 2839 startupinfo = subprocess.STARTUPINFO( 2840 dwFlags=STARTF_USERSHOWWINDOW, 2841 wShowWindow=SW_MAXIMIZE 2842 ) 2843 # Since Python is a console process, it won't be affected 2844 # by wShowWindow, but the argument should be silently 2845 # ignored 2846 subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"], 2847 startupinfo=startupinfo) 2848 2849 def test_startupinfo_copy(self): 2850 # bpo-34044: Popen must not modify input STARTUPINFO structure 2851 startupinfo = subprocess.STARTUPINFO() 2852 startupinfo.dwFlags = subprocess.STARTF_USESHOWWINDOW 2853 startupinfo.wShowWindow = subprocess.SW_HIDE 2854 2855 # Call Popen() twice with the same startupinfo object to make sure 2856 # that it's not modified 2857 for _ in range(2): 2858 cmd = [sys.executable, "-c", "pass"] 2859 with open(os.devnull, 'w') as null: 2860 proc = subprocess.Popen(cmd, 2861 stdout=null, 2862 stderr=subprocess.STDOUT, 2863 startupinfo=startupinfo) 2864 with proc: 2865 proc.communicate() 2866 self.assertEqual(proc.returncode, 0) 2867 2868 self.assertEqual(startupinfo.dwFlags, 2869 subprocess.STARTF_USESHOWWINDOW) 2870 self.assertIsNone(startupinfo.hStdInput) 2871 self.assertIsNone(startupinfo.hStdOutput) 2872 self.assertIsNone(startupinfo.hStdError) 2873 self.assertEqual(startupinfo.wShowWindow, subprocess.SW_HIDE) 2874 self.assertEqual(startupinfo.lpAttributeList, {"handle_list": []}) 2875 2876 def test_creationflags(self): 2877 # creationflags argument 2878 CREATE_NEW_CONSOLE = 16 2879 sys.stderr.write(" a DOS box should flash briefly ...\n") 2880 subprocess.call(sys.executable + 2881 ' -c "import time; time.sleep(0.25)"', 2882 creationflags=CREATE_NEW_CONSOLE) 2883 2884 def test_invalid_args(self): 2885 # invalid arguments should raise ValueError 2886 self.assertRaises(ValueError, subprocess.call, 2887 [sys.executable, "-c", 2888 "import sys; sys.exit(47)"], 2889 preexec_fn=lambda: 1) 2890 2891 @support.cpython_only 2892 def test_issue31471(self): 2893 # There shouldn't be an assertion failure in Popen() in case the env 2894 # argument has a bad keys() method. 2895 class BadEnv(dict): 2896 keys = None 2897 with self.assertRaises(TypeError): 2898 subprocess.Popen([sys.executable, "-c", "pass"], env=BadEnv()) 2899 2900 def test_close_fds(self): 2901 # close file descriptors 2902 rc = subprocess.call([sys.executable, "-c", 2903 "import sys; sys.exit(47)"], 2904 close_fds=True) 2905 self.assertEqual(rc, 47) 2906 2907 def test_close_fds_with_stdio(self): 2908 import msvcrt 2909 2910 fds = os.pipe() 2911 self.addCleanup(os.close, fds[0]) 2912 self.addCleanup(os.close, fds[1]) 2913 2914 handles = [] 2915 for fd in fds: 2916 os.set_inheritable(fd, True) 2917 handles.append(msvcrt.get_osfhandle(fd)) 2918 2919 p = subprocess.Popen([sys.executable, "-c", 2920 "import msvcrt; print(msvcrt.open_osfhandle({}, 0))".format(handles[0])], 2921 stdout=subprocess.PIPE, close_fds=False) 2922 stdout, stderr = p.communicate() 2923 self.assertEqual(p.returncode, 0) 2924 int(stdout.strip()) # Check that stdout is an integer 2925 2926 p = subprocess.Popen([sys.executable, "-c", 2927 "import msvcrt; print(msvcrt.open_osfhandle({}, 0))".format(handles[0])], 2928 stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True) 2929 stdout, stderr = p.communicate() 2930 self.assertEqual(p.returncode, 1) 2931 self.assertIn(b"OSError", stderr) 2932 2933 # The same as the previous call, but with an empty handle_list 2934 handle_list = [] 2935 startupinfo = subprocess.STARTUPINFO() 2936 startupinfo.lpAttributeList = {"handle_list": handle_list} 2937 p = subprocess.Popen([sys.executable, "-c", 2938 "import msvcrt; print(msvcrt.open_osfhandle({}, 0))".format(handles[0])], 2939 stdout=subprocess.PIPE, stderr=subprocess.PIPE, 2940 startupinfo=startupinfo, close_fds=True) 2941 stdout, stderr = p.communicate() 2942 self.assertEqual(p.returncode, 1) 2943 self.assertIn(b"OSError", stderr) 2944 2945 # Check for a warning due to using handle_list and close_fds=False 2946 with support.check_warnings((".*overriding close_fds", RuntimeWarning)): 2947 startupinfo = subprocess.STARTUPINFO() 2948 startupinfo.lpAttributeList = {"handle_list": handles[:]} 2949 p = subprocess.Popen([sys.executable, "-c", 2950 "import msvcrt; print(msvcrt.open_osfhandle({}, 0))".format(handles[0])], 2951 stdout=subprocess.PIPE, stderr=subprocess.PIPE, 2952 startupinfo=startupinfo, close_fds=False) 2953 stdout, stderr = p.communicate() 2954 self.assertEqual(p.returncode, 0) 2955 2956 def test_empty_attribute_list(self): 2957 startupinfo = subprocess.STARTUPINFO() 2958 startupinfo.lpAttributeList = {} 2959 subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"], 2960 startupinfo=startupinfo) 2961 2962 def test_empty_handle_list(self): 2963 startupinfo = subprocess.STARTUPINFO() 2964 startupinfo.lpAttributeList = {"handle_list": []} 2965 subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"], 2966 startupinfo=startupinfo) 2967 2968 def test_shell_sequence(self): 2969 # Run command through the shell (sequence) 2970 newenv = os.environ.copy() 2971 newenv["FRUIT"] = "physalis" 2972 p = subprocess.Popen(["set"], shell=1, 2973 stdout=subprocess.PIPE, 2974 env=newenv) 2975 with p: 2976 self.assertIn(b"physalis", p.stdout.read()) 2977 2978 def test_shell_string(self): 2979 # Run command through the shell (string) 2980 newenv = os.environ.copy() 2981 newenv["FRUIT"] = "physalis" 2982 p = subprocess.Popen("set", shell=1, 2983 stdout=subprocess.PIPE, 2984 env=newenv) 2985 with p: 2986 self.assertIn(b"physalis", p.stdout.read()) 2987 2988 def test_shell_encodings(self): 2989 # Run command through the shell (string) 2990 for enc in ['ansi', 'oem']: 2991 newenv = os.environ.copy() 2992 newenv["FRUIT"] = "physalis" 2993 p = subprocess.Popen("set", shell=1, 2994 stdout=subprocess.PIPE, 2995 env=newenv, 2996 encoding=enc) 2997 with p: 2998 self.assertIn("physalis", p.stdout.read(), enc) 2999 3000 def test_call_string(self): 3001 # call() function with string argument on Windows 3002 rc = subprocess.call(sys.executable + 3003 ' -c "import sys; sys.exit(47)"') 3004 self.assertEqual(rc, 47) 3005 3006 def _kill_process(self, method, *args): 3007 # Some win32 buildbot raises EOFError if stdin is inherited 3008 p = subprocess.Popen([sys.executable, "-c", """if 1: 3009 import sys, time 3010 sys.stdout.write('x\\n') 3011 sys.stdout.flush() 3012 time.sleep(30) 3013 """], 3014 stdin=subprocess.PIPE, 3015 stdout=subprocess.PIPE, 3016 stderr=subprocess.PIPE) 3017 with p: 3018 # Wait for the interpreter to be completely initialized before 3019 # sending any signal. 3020 p.stdout.read(1) 3021 getattr(p, method)(*args) 3022 _, stderr = p.communicate() 3023 self.assertStderrEqual(stderr, b'') 3024 returncode = p.wait() 3025 self.assertNotEqual(returncode, 0) 3026 3027 def _kill_dead_process(self, method, *args): 3028 p = subprocess.Popen([sys.executable, "-c", """if 1: 3029 import sys, time 3030 sys.stdout.write('x\\n') 3031 sys.stdout.flush() 3032 sys.exit(42) 3033 """], 3034 stdin=subprocess.PIPE, 3035 stdout=subprocess.PIPE, 3036 stderr=subprocess.PIPE) 3037 with p: 3038 # Wait for the interpreter to be completely initialized before 3039 # sending any signal. 3040 p.stdout.read(1) 3041 # The process should end after this 3042 time.sleep(1) 3043 # This shouldn't raise even though the child is now dead 3044 getattr(p, method)(*args) 3045 _, stderr = p.communicate() 3046 self.assertStderrEqual(stderr, b'') 3047 rc = p.wait() 3048 self.assertEqual(rc, 42) 3049 3050 def test_send_signal(self): 3051 self._kill_process('send_signal', signal.SIGTERM) 3052 3053 def test_kill(self): 3054 self._kill_process('kill') 3055 3056 def test_terminate(self): 3057 self._kill_process('terminate') 3058 3059 def test_send_signal_dead(self): 3060 self._kill_dead_process('send_signal', signal.SIGTERM) 3061 3062 def test_kill_dead(self): 3063 self._kill_dead_process('kill') 3064 3065 def test_terminate_dead(self): 3066 self._kill_dead_process('terminate') 3067 3068 class MiscTests(unittest.TestCase): 3069 3070 class RecordingPopen(subprocess.Popen): 3071 """A Popen that saves a reference to each instance for testing.""" 3072 instances_created = [] 3073 3074 def __init__(self, *args, **kwargs): 3075 super().__init__(*args, **kwargs) 3076 self.instances_created.append(self) 3077 3078 @mock.patch.object(subprocess.Popen, "_communicate") 3079 def _test_keyboardinterrupt_no_kill(self, popener, mock__communicate, 3080 **kwargs): 3081 """Fake a SIGINT happening during Popen._communicate() and ._wait(). 3082 3083 This avoids the need to actually try and get test environments to send 3084 and receive signals reliably across platforms. The net effect of a ^C 3085 happening during a blocking subprocess execution which we want to clean 3086 up from is a KeyboardInterrupt coming out of communicate() or wait(). 3087 """ 3088 3089 mock__communicate.side_effect = KeyboardInterrupt 3090 try: 3091 with mock.patch.object(subprocess.Popen, "_wait") as mock__wait: 3092 # We patch out _wait() as no signal was involved so the 3093 # child process isn't actually going to exit rapidly. 3094 mock__wait.side_effect = KeyboardInterrupt 3095 with mock.patch.object(subprocess, "Popen", 3096 self.RecordingPopen): 3097 with self.assertRaises(KeyboardInterrupt): 3098 popener([sys.executable, "-c", 3099 "import time\ntime.sleep(9)\nimport sys\n" 3100 "sys.stderr.write('\\n!runaway child!\\n')"], 3101 stdout=subprocess.DEVNULL, **kwargs) 3102 for call in mock__wait.call_args_list[1:]: 3103 self.assertNotEqual( 3104 call, mock.call(timeout=None), 3105 "no open-ended wait() after the first allowed: " 3106 f"{mock__wait.call_args_list}") 3107 sigint_calls = [] 3108 for call in mock__wait.call_args_list: 3109 if call == mock.call(timeout=0.25): # from Popen.__init__ 3110 sigint_calls.append(call) 3111 self.assertLessEqual(mock__wait.call_count, 2, 3112 msg=mock__wait.call_args_list) 3113 self.assertEqual(len(sigint_calls), 1, 3114 msg=mock__wait.call_args_list) 3115 finally: 3116 # cleanup the forgotten (due to our mocks) child process 3117 process = self.RecordingPopen.instances_created.pop() 3118 process.kill() 3119 process.wait() 3120 self.assertEqual([], self.RecordingPopen.instances_created) 3121 3122 def test_call_keyboardinterrupt_no_kill(self): 3123 self._test_keyboardinterrupt_no_kill(subprocess.call, timeout=6.282) 3124 3125 def test_run_keyboardinterrupt_no_kill(self): 3126 self._test_keyboardinterrupt_no_kill(subprocess.run, timeout=6.282) 3127 3128 def test_context_manager_keyboardinterrupt_no_kill(self): 3129 def popen_via_context_manager(*args, **kwargs): 3130 with subprocess.Popen(*args, **kwargs) as unused_process: 3131 raise KeyboardInterrupt # Test how __exit__ handles ^C. 3132 self._test_keyboardinterrupt_no_kill(popen_via_context_manager) 3133 3134 def test_getoutput(self): 3135 self.assertEqual(subprocess.getoutput('echo xyzzy'), 'xyzzy') 3136 self.assertEqual(subprocess.getstatusoutput('echo xyzzy'), 3137 (0, 'xyzzy')) 3138 3139 # we use mkdtemp in the next line to create an empty directory 3140 # under our exclusive control; from that, we can invent a pathname 3141 # that we _know_ won't exist. This is guaranteed to fail. 3142 dir = None 3143 try: 3144 dir = tempfile.mkdtemp() 3145 name = os.path.join(dir, "foo") 3146 status, output = subprocess.getstatusoutput( 3147 ("type " if mswindows else "cat ") + name) 3148 self.assertNotEqual(status, 0) 3149 finally: 3150 if dir is not None: 3151 os.rmdir(dir) 3152 3153 def test__all__(self): 3154 """Ensure that __all__ is populated properly.""" 3155 intentionally_excluded = {"list2cmdline", "Handle"} 3156 exported = set(subprocess.__all__) 3157 possible_exports = set() 3158 import types 3159 for name, value in subprocess.__dict__.items(): 3160 if name.startswith('_'): 3161 continue 3162 if isinstance(value, (types.ModuleType,)): 3163 continue 3164 possible_exports.add(name) 3165 self.assertEqual(exported, possible_exports - intentionally_excluded) 3166 3167 3168 @unittest.skipUnless(hasattr(selectors, 'PollSelector'), 3169 "Test needs selectors.PollSelector") 3170 class ProcessTestCaseNoPoll(ProcessTestCase): 3171 def setUp(self): 3172 self.orig_selector = subprocess._PopenSelector 3173 subprocess._PopenSelector = selectors.SelectSelector 3174 ProcessTestCase.setUp(self) 3175 3176 def tearDown(self): 3177 subprocess._PopenSelector = self.orig_selector 3178 ProcessTestCase.tearDown(self) 3179 3180 3181 @unittest.skipUnless(mswindows, "Windows-specific tests") 3182 class CommandsWithSpaces (BaseTestCase): 3183 3184 def setUp(self): 3185 super().setUp() 3186 f, fname = tempfile.mkstemp(".py", "te st") 3187 self.fname = fname.lower () 3188 os.write(f, b"import sys;" 3189 b"sys.stdout.write('%d %s' % (len(sys.argv), [a.lower () for a in sys.argv]))" 3190 ) 3191 os.close(f) 3192 3193 def tearDown(self): 3194 os.remove(self.fname) 3195 super().tearDown() 3196 3197 def with_spaces(self, *args, **kwargs): 3198 kwargs['stdout'] = subprocess.PIPE 3199 p = subprocess.Popen(*args, **kwargs) 3200 with p: 3201 self.assertEqual( 3202 p.stdout.read ().decode("mbcs"), 3203 "2 [%r, 'ab cd']" % self.fname 3204 ) 3205 3206 def test_shell_string_with_spaces(self): 3207 # call() function with string argument with spaces on Windows 3208 self.with_spaces('"%s" "%s" "%s"' % (sys.executable, self.fname, 3209 "ab cd"), shell=1) 3210 3211 def test_shell_sequence_with_spaces(self): 3212 # call() function with sequence argument with spaces on Windows 3213 self.with_spaces([sys.executable, self.fname, "ab cd"], shell=1) 3214 3215 def test_noshell_string_with_spaces(self): 3216 # call() function with string argument with spaces on Windows 3217 self.with_spaces('"%s" "%s" "%s"' % (sys.executable, self.fname, 3218 "ab cd")) 3219 3220 def test_noshell_sequence_with_spaces(self): 3221 # call() function with sequence argument with spaces on Windows 3222 self.with_spaces([sys.executable, self.fname, "ab cd"]) 3223 3224 3225 class ContextManagerTests(BaseTestCase): 3226 3227 def test_pipe(self): 3228 with subprocess.Popen([sys.executable, "-c", 3229 "import sys;" 3230 "sys.stdout.write('stdout');" 3231 "sys.stderr.write('stderr');"], 3232 stdout=subprocess.PIPE, 3233 stderr=subprocess.PIPE) as proc: 3234 self.assertEqual(proc.stdout.read(), b"stdout") 3235 self.assertStderrEqual(proc.stderr.read(), b"stderr") 3236 3237 self.assertTrue(proc.stdout.closed) 3238 self.assertTrue(proc.stderr.closed) 3239 3240 def test_returncode(self): 3241 with subprocess.Popen([sys.executable, "-c", 3242 "import sys; sys.exit(100)"]) as proc: 3243 pass 3244 # __exit__ calls wait(), so the returncode should be set 3245 self.assertEqual(proc.returncode, 100) 3246 3247 def test_communicate_stdin(self): 3248 with subprocess.Popen([sys.executable, "-c", 3249 "import sys;" 3250 "sys.exit(sys.stdin.read() == 'context')"], 3251 stdin=subprocess.PIPE) as proc: 3252 proc.communicate(b"context") 3253 self.assertEqual(proc.returncode, 1) 3254 3255 def test_invalid_args(self): 3256 with self.assertRaises(NONEXISTING_ERRORS): 3257 with subprocess.Popen(NONEXISTING_CMD, 3258 stdout=subprocess.PIPE, 3259 stderr=subprocess.PIPE) as proc: 3260 pass 3261 3262 def test_broken_pipe_cleanup(self): 3263 """Broken pipe error should not prevent wait() (Issue 21619)""" 3264 proc = subprocess.Popen([sys.executable, '-c', 'pass'], 3265 stdin=subprocess.PIPE, 3266 bufsize=support.PIPE_MAX_SIZE*2) 3267 proc = proc.__enter__() 3268 # Prepare to send enough data to overflow any OS pipe buffering and 3269 # guarantee a broken pipe error. Data is held in BufferedWriter 3270 # buffer until closed. 3271 proc.stdin.write(b'x' * support.PIPE_MAX_SIZE) 3272 self.assertIsNone(proc.returncode) 3273 # EPIPE expected under POSIX; EINVAL under Windows 3274 self.assertRaises(OSError, proc.__exit__, None, None, None) 3275 self.assertEqual(proc.returncode, 0) 3276 self.assertTrue(proc.stdin.closed) 3277 3278 3279 if __name__ == "__main__": 3280 unittest.main() 3281