wip

Author: bityob

Lib/test/test_concurrent_futures.py

@@ -1,6 +1,7 @@
 from test import support
 from test.support import import_helper
 from test.support import threading_helper
+from test.test_unittest.support import BufferedWriter
 
 # Skip tests if _multiprocessing wasn't built.
 import_helper.import_module('_multiprocessing')
@@ -285,11 +286,33 @@ def _assert_logged(self, msg):
         self.assertTrue(any(msg in line for line in output),
                         output)
 
+    # def tearDown(self):
+    #     super().tearDown()
+    #     breakpoint()
+    #     text = self.stream.getvalue()
+    #     self.assertNotIn("leaked semaphore objects", text)
+
 
 create_executor_tests(InitializerMixin)
 create_executor_tests(FailingInitializerMixin)
 
 
+class FailingInitializerTextTest(unittest.TestCase):
+    def test_spawn(self):
+        stream = BufferedWriter()
+        runner = unittest.TextTestRunner(stream=stream)
+        result = runner.run(ProcessPoolSpawnFailingInitializerTest('test_initializer'))
+        text = stream.getvalue()
+        self.assertNotIn("leaked semaphore objects", text)
+
+    def test_forkserver(self):
+        stream = BufferedWriter()
+        runner = unittest.TextTestRunner(stream=stream)
+        result = runner.run(ProcessPoolForkserverFailingInitializerTest('test_initializer'))
+        text = stream.getvalue()
+        self.assertNotIn("leaked semaphore objects", text)
+
+
 class ExecutorShutdownTest:
     def test_run_after_shutdown(self):
         self.executor.shutdown()

Lib/unittest/case.py

@@ -400,6 +400,7 @@ def __init__(self, methodName='runTest'):
         self._outcome = None
         self._testMethodDoc = 'No test'
         try:
+            # breakpoint()
             testMethod = getattr(self, methodName)
         except AttributeError:
             if methodName != 'runTest':