add support for os.fork and related functions by itsankitkp · Pull Request #4877 · RustPython/RustPython

Expand Up @@ -69,6 +69,10 @@ except ImportError: INT_MAX = PY_SSIZE_T_MAX = sys.maxsize
try: import _testcapi except ImportError: _testcapi = None
from test.support.script_helper import assert_python_ok from test.support import unix_shell Expand Down Expand Up @@ -3067,11 +3071,13 @@ def check_waitpid(self, code, exitcode, callback=None): self.assertEqual(pid2, pid)
# TODO: RUSTPYTHON (AttributeError: module 'os' has no attribute 'spawnv') @unittest.skipUnless(hasattr(threading.Lock(), '_at_fork_reinit'), 'TODO: RUSTPYTHON, test needs lock._at_fork_reinit') @unittest.expectedFailure def test_waitpid(self): self.check_waitpid(code='pass', exitcode=0)
# TODO: RUSTPYTHON (AttributeError: module 'os' has no attribute 'spawnv') @unittest.skipUnless(hasattr(threading.Lock(), '_at_fork_reinit'), 'TODO: RUSTPYTHON, test needs lock._at_fork_reinit') @unittest.expectedFailure def test_waitstatus_to_exitcode(self): exitcode = 23 Expand Down Expand Up @@ -3103,7 +3109,8 @@ def test_waitstatus_to_exitcode_windows(self): os.waitstatus_to_exitcode((max_exitcode + 1) << 8) with self.assertRaises(OverflowError): os.waitstatus_to_exitcode(-1)

@unittest.skipUnless(hasattr(threading.Lock(), '_at_fork_reinit'), 'TODO: RUSTPYTHON, test needs lock._at_fork_reinit') # TODO: RUSTPYTHON (AttributeError: module 'os' has no attribute 'spawnv') @unittest.expectedFailure # Skip the test on Windows Expand Down Expand Up @@ -3146,31 +3153,36 @@ def create_args(self, *, with_env=False, use_bytes=False): for k, v in self.env.items()}
return args

@unittest.skipUnless(hasattr(threading.Lock(), '_at_fork_reinit'), 'TODO: RUSTPYTHON, test needs lock._at_fork_reinit') @requires_os_func('spawnl') def test_spawnl(self): args = self.create_args() exitcode = os.spawnl(os.P_WAIT, args[0], *args) self.assertEqual(exitcode, self.exitcode)
@unittest.skipUnless(hasattr(threading.Lock(), '_at_fork_reinit'), 'TODO: RUSTPYTHON, test needs lock._at_fork_reinit') @requires_os_func('spawnle') def test_spawnle(self): args = self.create_args(with_env=True) exitcode = os.spawnle(os.P_WAIT, args[0], *args, self.env) self.assertEqual(exitcode, self.exitcode)
@unittest.skipUnless(hasattr(threading.Lock(), '_at_fork_reinit'), 'TODO: RUSTPYTHON, test needs lock._at_fork_reinit') @requires_os_func('spawnlp') def test_spawnlp(self): args = self.create_args() exitcode = os.spawnlp(os.P_WAIT, args[0], *args) self.assertEqual(exitcode, self.exitcode)
@unittest.skipUnless(hasattr(threading.Lock(), '_at_fork_reinit'), 'TODO: RUSTPYTHON, test needs lock._at_fork_reinit') @requires_os_func('spawnlpe') def test_spawnlpe(self): args = self.create_args(with_env=True) exitcode = os.spawnlpe(os.P_WAIT, args[0], *args, self.env) self.assertEqual(exitcode, self.exitcode)
@unittest.skipUnless(hasattr(threading.Lock(), '_at_fork_reinit'), 'TODO: RUSTPYTHON, test needs lock._at_fork_reinit') @requires_os_func('spawnv') def test_spawnv(self): args = self.create_args() Expand All @@ -3181,30 +3193,35 @@ def test_spawnv(self): exitcode = os.spawnv(os.P_WAIT, FakePath(args[0]), args) self.assertEqual(exitcode, self.exitcode)
@unittest.skipUnless(hasattr(threading.Lock(), '_at_fork_reinit'), 'TODO: RUSTPYTHON, test needs lock._at_fork_reinit') @requires_os_func('spawnve') def test_spawnve(self): args = self.create_args(with_env=True) exitcode = os.spawnve(os.P_WAIT, args[0], args, self.env) self.assertEqual(exitcode, self.exitcode)
@unittest.skipUnless(hasattr(threading.Lock(), '_at_fork_reinit'), 'TODO: RUSTPYTHON, test needs lock._at_fork_reinit') @requires_os_func('spawnvp') def test_spawnvp(self): args = self.create_args() exitcode = os.spawnvp(os.P_WAIT, args[0], args) self.assertEqual(exitcode, self.exitcode)
@unittest.skipUnless(hasattr(threading.Lock(), '_at_fork_reinit'), 'TODO: RUSTPYTHON, test needs lock._at_fork_reinit') @requires_os_func('spawnvpe') def test_spawnvpe(self): args = self.create_args(with_env=True) exitcode = os.spawnvpe(os.P_WAIT, args[0], args, self.env) self.assertEqual(exitcode, self.exitcode)
@unittest.skipUnless(hasattr(threading.Lock(), '_at_fork_reinit'), 'TODO: RUSTPYTHON, test needs lock._at_fork_reinit') @requires_os_func('spawnv') def test_nowait(self): args = self.create_args() pid = os.spawnv(os.P_NOWAIT, args[0], args) support.wait_process(pid, exitcode=self.exitcode)
@unittest.skipUnless(hasattr(threading.Lock(), '_at_fork_reinit'), 'TODO: RUSTPYTHON, test needs lock._at_fork_reinit') @requires_os_func('spawnve') def test_spawnve_bytes(self): # Test bytes handling in parse_arglist and parse_envlist (#28114) Expand Down Expand Up @@ -3286,10 +3303,12 @@ def _test_invalid_env(self, spawn): exitcode = spawn(os.P_WAIT, args[0], args, newenv) self.assertEqual(exitcode, 0)
@unittest.skipUnless(hasattr(threading.Lock(), '_at_fork_reinit'), 'TODO: RUSTPYTHON, test needs lock._at_fork_reinit') @requires_os_func('spawnve') def test_spawnve_invalid_env(self): self._test_invalid_env(os.spawnve)
@unittest.skipUnless(hasattr(threading.Lock(), '_at_fork_reinit'), 'TODO: RUSTPYTHON, test needs lock._at_fork_reinit') @requires_os_func('spawnvpe') def test_spawnvpe_invalid_env(self): self._test_invalid_env(os.spawnvpe) Expand Down Expand Up @@ -4660,6 +4679,35 @@ def test_fork(self): assert_python_ok("-c", code) assert_python_ok("-c", code, PYTHONMALLOC="malloc_debug")
@unittest.skipIf(_testcapi is None, 'TODO: RUSTPYTHON; needs _testcapi') @unittest.skipUnless(sys.platform in ("linux", "darwin"), "Only Linux and macOS detect this today.") def test_fork_warns_when_non_python_thread_exists(self): code = """if 1: import os, threading, warnings from _testcapi import _spawn_pthread_waiter, _end_spawned_pthread _spawn_pthread_waiter() try: with warnings.catch_warnings(record=True) as ws: warnings.filterwarnings( "always", category=DeprecationWarning) if os.fork() == 0: assert not ws, f"unexpected warnings in child: {ws}" os._exit(0) # child else: assert ws[0].category == DeprecationWarning, ws[0] assert 'fork' in str(ws[0].message), ws[0] # Waiting allows an error in the child to hit stderr. exitcode = os.wait()[1] assert exitcode == 0, f"child exited {exitcode}" assert threading.active_count() == 1, threading.enumerate() finally: _end_spawned_pthread() """ _, out, err = assert_python_ok("-c", code, PYTHONOPTIMIZE='0') self.assertEqual(err.decode("utf-8"), "") self.assertEqual(out.decode("utf-8"), "")

# Only test if the C version is provided, otherwise TestPEP519 already tested # the pure Python implementation. Expand Down