"requires_gzip", "requires_bz2", "requires_lzma",
"bigmemtest", "bigaddrspacetest", "cpython_only", "get_attribute",
"requires_IEEE_754", "requires_zlib",
+ "has_fork_support", "requires_fork",
"anticipate_failure", "load_package_tests", "detect_api_mismatch",
"check__all__", "skip_if_buggy_ucrt_strfptime",
"check_disallow_instantiation",
# sys
- "is_jython", "is_android", "check_impl_detail", "unix_shell",
- "setswitchinterval",
+ "is_jython", "is_android", "is_emscripten",
+ "check_impl_detail", "unix_shell", "setswitchinterval",
# network
"open_urlresource",
# processes
else:
unix_shell = None
+# wasm32-emscripten is POSIX-like but does not provide a
+# working fork() or subprocess API.
+is_emscripten = sys.platform == "emscripten"
+
+has_fork_support = hasattr(os, "fork") and not is_emscripten
+
+def requires_fork():
+ return unittest.skipUnless(has_fork_support, "requires working os.fork()")
+
# Define the URL of a dedicated HTTP server for the network tests.
# The URL must use clear-text HTTP: no redirection to encrypted HTTPS.
TEST_HTTP_URL = "http://www.pythontest.net"
# Skip test if fork does not exist.
-support.get_attribute(os, 'fork')
+if not support.has_fork_support:
+ raise unittest.SkipTest("test module requires working os.fork")
+
class ForkTest(ForkWait):
def test_threaded_import_lock_fork(self):
# tests validity but not completeness of the __all__ list
self.assertTrue(set(random.__all__) <= set(dir(random)))
- @unittest.skipUnless(hasattr(os, "fork"), "fork() required")
+ @test.support.requires_fork()
def test_after_fork(self):
# Test the global Random instance gets reseeded in child
r, w = os.pipe()
f'temporary directory {path!r}: '),
warn)
- @unittest.skipUnless(hasattr(os, "fork"), "test requires os.fork")
+ @support.requires_fork()
def test_temp_dir__forked_child(self):
"""Test that a forked child process does not remove the directory."""
# See bpo-30028 for details.
@unittest.skipUnless(hasattr(os, 'waitpid') and hasattr(os, 'WNOHANG'),
'need os.waitpid() and os.WNOHANG')
+ @support.requires_fork()
def test_reap_children(self):
# Make sure that there is no other pending child process
support.reap_children()
'EXT_SUFFIX required for this test')
def test_EXT_SUFFIX_in_vars(self):
import _imp
+ if not _imp.extension_suffixes():
+ self.skipTest("stub loader has no suffixes")
vars = sysconfig.get_config_vars()
self.assertIsNotNone(vars['SO'])
self.assertEqual(vars['SO'], vars['EXT_SUFFIX'])
if i == 20:
break
- @unittest.skipUnless(hasattr(os, 'fork'),
- "os.fork is required for this test")
+ @support.requires_fork()
def test_process_awareness(self):
# ensure that the random source differs between
# child and parent.
def setUp(self):
self.read_fd, self.write_fd = os.pipe()
- @unittest.skipUnless(hasattr(os, 'fork'), 'need os.fork')
+ @support.requires_fork()
@threading_helper.reap_threads
def test_forkinthread(self):
pid = None
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.
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.
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.
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:
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:
"""
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
"""
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
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
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()
equal(u, self.uuid.UUID(v))
equal(str(u), v)
- @unittest.skipUnless(hasattr(os, 'fork'), 'need os.fork')
+ @support.requires_fork()
def testIssue8621(self):
# On at least some versions of OSX self.uuid.uuid4 generates
# the same sequence of UUIDs in the parent and any
--- /dev/null
+Add :func:`test.support.requires_fork` decorators to mark tests that require
+a working :func:`os.fork`.
projects / thirdparty / Python / cpython.git / commitdiff
