Refactor test_preexec_errpipe to not create an uncollectable reference cycle.

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.")
+    class _TestExecuteChildPopen(subprocess.Popen):
-    def test_preexec_errpipe_does_not_double_close_pipes(self):
+        """Used to test behavior at the end of _execute_child."""
-        """Issue16140: Don't double close pipes on preexec error."""
+        def __init__(self, testcase, *args, **kwargs):
-        class SafeConstructorPopen(subprocess.Popen):
+            # Do nothing so we can modify the instance for testing.
-            def __init__(self):
+            self._testcase = testcase
-                pass  # Do nothing so we can modify the instance for testing.
+            subprocess.Popen.__init__(self, *args, **kwargs)
-            def RealPopen(self, *args, **kwargs):
-                subprocess.Popen.__init__(self, *args, **kwargs)
-        def raise_it():
-            raise RuntimeError("force the _execute_child() errpipe_data path.")
 
-        p = SafeConstructorPopen()
+        def _execute_child(
-
+                self, args, executable, preexec_fn, close_fds, cwd, env,
-        def _test_fds_execute_child_wrapper(
-                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"],
+            self._TestExecuteChildPopen(
-                        stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+                    self, [sys.executable, "-c", "pass"],
-                        stderr=subprocess.PIPE, preexec_fn=raise_it)
+                    stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE, preexec_fn=raise_it)
 
     def test_args_string(self):
         # args is a string