bpo-40280: Add requires_fork test helper (GH-30622) by tiran · Pull Request #30622 · python/cpython
Expand Up
@@ -505,7 +505,7 @@ def test_daemon_param(self):
t = threading.Thread(daemon=True)
self.assertTrue(t.daemon)
@unittest.skipUnless(hasattr(os, 'fork'), 'needs os.fork()') @support.requires_fork() def test_fork_at_exit(self): # bpo-42350: Calling os.fork() after threading._shutdown() must # not log an error. Expand Down Expand Up @@ -533,7 +533,7 @@ def exit_handler(): self.assertEqual(out, b'') self.assertEqual(err.rstrip(), b'child process ok')
@unittest.skipUnless(hasattr(os, 'fork'), 'test needs fork()') @support.requires_fork() def test_dummy_thread_after_fork(self): # Issue #14308: a dummy thread in the active list doesn't mess up # the after-fork mechanism. Expand All @@ -560,7 +560,7 @@ def background_thread(evt): self.assertEqual(out, b'') self.assertEqual(err, b'')
@unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()") @support.requires_fork() def test_is_alive_after_fork(self): # Try hard to trigger #18418: is_alive() could sometimes be True on # threads that vanished after a fork. Expand Down Expand Up @@ -594,7 +594,7 @@ def f(): th.start() th.join()
@unittest.skipUnless(hasattr(os, 'fork'), "test needs os.fork()") @support.requires_fork() @unittest.skipUnless(hasattr(os, 'waitpid'), "test needs os.waitpid()") def test_main_thread_after_fork(self): code = """if 1: Expand All @@ -616,7 +616,7 @@ def test_main_thread_after_fork(self): self.assertEqual(data, "MainThread\nTrue\nTrue\n")
@unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug") @unittest.skipUnless(hasattr(os, 'fork'), "test needs os.fork()") @support.requires_fork() @unittest.skipUnless(hasattr(os, 'waitpid'), "test needs os.waitpid()") def test_main_thread_after_fork_from_nonmain_thread(self): code = """if 1: Expand Down Expand Up @@ -993,7 +993,7 @@ def test_1_join_on_shutdown(self): """ self._run_and_join(script)
@unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()") @support.requires_fork() @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug") def test_2_join_in_forked_process(self): # Like the test above, but from a forked interpreter Expand All @@ -1014,7 +1014,7 @@ def test_2_join_in_forked_process(self): """ self._run_and_join(script)
@unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()") @support.requires_fork() @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug") def test_3_join_in_forked_from_thread(self): # Like the test above, but fork() was called from a worker thread Expand Down Expand Up @@ -1085,7 +1085,7 @@ def main(): rc, out, err = assert_python_ok('-c', script) self.assertFalse(err)
@unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()") @support.requires_fork() @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug") def test_reinit_tls_after_fork(self): # Issue #13817: fork() would deadlock in a multithreaded program with Expand All @@ -1109,7 +1109,7 @@ def do_fork_and_wait(): for t in threads: t.join()
@unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()") @support.requires_fork() def test_clear_threads_states_after_fork(self): # Issue #17094: check that threads states are cleared after fork()
Expand Down
@unittest.skipUnless(hasattr(os, 'fork'), 'needs os.fork()') @support.requires_fork() def test_fork_at_exit(self): # bpo-42350: Calling os.fork() after threading._shutdown() must # not log an error. Expand Down Expand Up @@ -533,7 +533,7 @@ def exit_handler(): self.assertEqual(out, b'') self.assertEqual(err.rstrip(), b'child process ok')
@unittest.skipUnless(hasattr(os, 'fork'), 'test needs fork()') @support.requires_fork() def test_dummy_thread_after_fork(self): # Issue #14308: a dummy thread in the active list doesn't mess up # the after-fork mechanism. Expand All @@ -560,7 +560,7 @@ def background_thread(evt): self.assertEqual(out, b'') self.assertEqual(err, b'')
@unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()") @support.requires_fork() def test_is_alive_after_fork(self): # Try hard to trigger #18418: is_alive() could sometimes be True on # threads that vanished after a fork. Expand Down Expand Up @@ -594,7 +594,7 @@ def f(): th.start() th.join()
@unittest.skipUnless(hasattr(os, 'fork'), "test needs os.fork()") @support.requires_fork() @unittest.skipUnless(hasattr(os, 'waitpid'), "test needs os.waitpid()") def test_main_thread_after_fork(self): code = """if 1: Expand All @@ -616,7 +616,7 @@ def test_main_thread_after_fork(self): self.assertEqual(data, "MainThread\nTrue\nTrue\n")
@unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug") @unittest.skipUnless(hasattr(os, 'fork'), "test needs os.fork()") @support.requires_fork() @unittest.skipUnless(hasattr(os, 'waitpid'), "test needs os.waitpid()") def test_main_thread_after_fork_from_nonmain_thread(self): code = """if 1: Expand Down Expand Up @@ -993,7 +993,7 @@ def test_1_join_on_shutdown(self): """ self._run_and_join(script)
@unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()") @support.requires_fork() @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug") def test_2_join_in_forked_process(self): # Like the test above, but from a forked interpreter Expand All @@ -1014,7 +1014,7 @@ def test_2_join_in_forked_process(self): """ self._run_and_join(script)
@unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()") @support.requires_fork() @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug") def test_3_join_in_forked_from_thread(self): # Like the test above, but fork() was called from a worker thread Expand Down Expand Up @@ -1085,7 +1085,7 @@ def main(): rc, out, err = assert_python_ok('-c', script) self.assertFalse(err)
@unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()") @support.requires_fork() @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug") def test_reinit_tls_after_fork(self): # Issue #13817: fork() would deadlock in a multithreaded program with Expand All @@ -1109,7 +1109,7 @@ def do_fork_and_wait(): for t in threads: t.join()
@unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()") @support.requires_fork() def test_clear_threads_states_after_fork(self): # Issue #17094: check that threads states are cleared after fork()
Expand Down