gh-104090: Fix unittest collectedDurations resources leak

Author: bityob

Lib/multiprocessing/resource_tracker.py

@@ -57,6 +57,7 @@ def __init__(self):
         self._lock = threading.Lock()
         self._fd = None
         self._pid = None
+        self._exitcode = None
 
     def _stop(self):
         with self._lock:
@@ -68,7 +69,7 @@ def _stop(self):
             os.close(self._fd)
             self._fd = None
 
-            os.waitpid(self._pid, 0)
+            _, self._exitcode = os.waitpid(self._pid, 0)
             self._pid = None
 
     def getfd(self):
@@ -191,6 +192,8 @@ def main(fd):
             pass
 
     cache = {rtype: set() for rtype in _CLEANUP_FUNCS.keys()}
+    exit_code = 0
+
     try:
         # keep track of registered/unregistered resources
         with open(fd, 'rb') as f:
@@ -221,6 +224,7 @@ def main(fd):
         for rtype, rtype_cache in cache.items():
             if rtype_cache:
                 try:
+                    exit_code = 1
                     warnings.warn('resource_tracker: There appear to be %d '
                                   'leaked %s objects to clean up at shutdown' %
                                   (len(rtype_cache), rtype))
@@ -237,3 +241,5 @@ def main(fd):
                         warnings.warn('resource_tracker: %r: %s' % (name, e))
                 finally:
                     pass
+
+        sys.exit(exit_code)

Lib/test/test_concurrent_futures.py

@@ -290,6 +290,34 @@ def _assert_logged(self, msg):
 create_executor_tests(FailingInitializerMixin)
 
 
+class FailingInitializerResourcesTest(unittest.TestCase):
+    """
+    Source: https://github.com/python/cpython/issues/104090
+    """
+
+    def _test(self, test_class):
+        runner = unittest.TextTestRunner()
+        result = runner.run(test_class('test_initializer'))
+
+        self.assertEqual(result.testsRun, 1)
+        self.assertEqual(result.failures, [])
+        self.assertEqual(result.errors, [])
+
+        # GH-104090:
+        # Stop resource tracker manually now, so we can verify there are not leaked resources by checking
+        # the process exit code
+        from multiprocessing.resource_tracker import _resource_tracker
+        _resource_tracker._stop()
+
+        self.assertEqual(_resource_tracker._exitcode, 0)
+
+    def test_spawn(self):
+        self._test(ProcessPoolSpawnFailingInitializerTest)
+
+    def test_forkserver(self):
+        self._test(ProcessPoolForkserverFailingInitializerTest)
+
+
 class ExecutorShutdownTest:
     def test_run_after_shutdown(self):
         self.executor.shutdown()

Lib/unittest/result.py

@@ -166,7 +166,8 @@ def addDuration(self, test, elapsed):
         """
         # support for a TextTestRunner using an old TestResult class
         if hasattr(self, "collectedDurations"):
-            self.collectedDurations.append((test, elapsed))
+            # Pass test repr and not the test object itself to avoid resources leak
+            self.collectedDurations.append((str(test), elapsed))
 
     def wasSuccessful(self):
         """Tells whether or not this result was a success."""