diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py index 59eab3063c1..f713fc60659 100644 --- a/Lib/test/test_subprocess.py +++ b/Lib/test/test_subprocess.py @@ -751,28 +751,22 @@ class POSIXProcessTestCase(BaseTestCase): self.addCleanup(p.stdout.close) self.assertEqual(p.stdout.read(), "apple") - @unittest.skipIf(not os.path.exists("/dev/zero"), "/dev/zero required.") - def test_preexec_errpipe_does_not_double_close_pipes(self): - """Issue16140: Don't double close pipes on preexec error.""" - class SafeConstructorPopen(subprocess.Popen): - def __init__(self): - pass # Do nothing so we can modify the instance for testing. - def RealPopen(self, *args, **kwargs): - subprocess.Popen.__init__(self, *args, **kwargs) - def raise_it(): - raise RuntimeError("force the _execute_child() errpipe_data path.") + class _TestExecuteChildPopen(subprocess.Popen): + """Used to test behavior at the end of _execute_child.""" + def __init__(self, testcase, *args, **kwargs): + # Do nothing so we can modify the instance for testing. + self._testcase = testcase + subprocess.Popen.__init__(self, *args, **kwargs) - p = SafeConstructorPopen() - - def _test_fds_execute_child_wrapper( - args, executable, preexec_fn, close_fds, cwd, env, + def _execute_child( + self, args, executable, preexec_fn, close_fds, cwd, env, universal_newlines, startupinfo, creationflags, shell, p2cread, p2cwrite, c2pread, c2pwrite, errread, errwrite): try: subprocess.Popen._execute_child( - p, args, executable, preexec_fn, close_fds, + self, args, executable, preexec_fn, close_fds, cwd, env, universal_newlines, startupinfo, creationflags, shell, p2cread, p2cwrite, @@ -786,17 +780,23 @@ class POSIXProcessTestCase(BaseTestCase): for _ in range(8)] try: for fd in devzero_fds: - self.assertNotIn(fd, (p2cwrite, c2pread, errread)) - + self._testcase.assertNotIn( + fd, (p2cwrite, c2pread, errread)) finally: map(os.close, devzero_fds) - p._execute_child = _test_fds_execute_child_wrapper + @unittest.skipIf(not os.path.exists("/dev/zero"), "/dev/zero required.") + def test_preexec_errpipe_does_not_double_close_pipes(self): + """Issue16140: Don't double close pipes on preexec error.""" + + def raise_it(): + raise RuntimeError("force the _execute_child() errpipe_data path.") with self.assertRaises(RuntimeError): - p.RealPopen([sys.executable, "-c", "pass"], - stdin=subprocess.PIPE, stdout=subprocess.PIPE, - stderr=subprocess.PIPE, preexec_fn=raise_it) + self._TestExecuteChildPopen( + self, [sys.executable, "-c", "pass"], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, preexec_fn=raise_it) def test_args_string(self): # args is a string