https://github.com/python/cpython/commit/8ece4da730f21e7fb1bf1f14fb0c66ba7f7cec9d commit: 8ece4da730f21e7fb1bf1f14fb0c66ba7f7cec9d branch: 3.14 author: Miss Islington (bot) <[email protected]> committer: serhiy-storchaka <[email protected]> date: 2026-08-05T11:31:40Z summary:
[3.14] gh-152548: Add options, env and timeout parameters to runInSubprocess() (GH-155163) (GH-155230) They run the test subprocess with specific interpreter command line options and environment variables, and limit how long it may take. All are keyword-only. env is layered over the inherited environment; a None value unsets a variable. There is no timeout by default. (cherry picked from commit 1d90627f1660253b30e5efc2ca092a6231737993) Co-authored-by: Serhiy Storchaka <[email protected]> Co-authored-by: Claude Opus 5 (1M context) <[email protected]> files: M Doc/library/test.rst M Lib/test/_isolated_sample.py M Lib/test/support/isolation.py M Lib/test/test_support.py diff --git a/Doc/library/test.rst b/Doc/library/test.rst index c1e086059376ac9..7404007fa287c83 100644 --- a/Doc/library/test.rst +++ b/Doc/library/test.rst @@ -965,7 +965,7 @@ The :mod:`!test.support` module defines the following functions: .. currentmodule:: test.support.isolation -.. decorator:: runInSubprocess() +.. decorator:: runInSubprocess(*, options=(), env=None, timeout=None) Decorator that runs the decorated test in a fresh interpreter subprocess, in isolation, so that it does not share global or interpreter state with the @@ -999,6 +999,19 @@ The :mod:`!test.support` module defines the following functions: :func:`~test.support.bigmemtest` and the like behave consistently in both processes. + *options* is a sequence of interpreter command line options + to run the subprocess with, + and *env* is a mapping of environment variables to set in it, + on top of the inherited environment. + A value of ``None`` in *env* unsets the variable. + Note that :option:`-E` and :option:`-I` make the subprocess ignore + the ``PYTHON*`` environment variables, including :envvar:`PYTHONPATH`. + + *timeout* is the number of seconds to wait for the subprocess; + the test is reported as an error if it does not complete in time. + By default there is no timeout, + and a hung test is left to the timeout of the test runner. + The test is skipped on platforms without subprocess support. diff --git a/Lib/test/_isolated_sample.py b/Lib/test/_isolated_sample.py index c89f7145e7328d3..5853b654fc28cb6 100644 --- a/Lib/test/_isolated_sample.py +++ b/Lib/test/_isolated_sample.py @@ -7,6 +7,7 @@ import atexit import os +import sys import time import unittest from test.support import isolation @@ -141,3 +142,39 @@ def test_pass(self): def test_dies(self): _die_at_exit() + + [email protected](options=['-X', 'dev', '-W', 'error::BytesWarning']) +class OptionsSample(unittest.TestCase): + + def test_options_applied(self): + self.assertTrue(sys.flags.dev_mode) + self.assertIn('error::BytesWarning', sys.warnoptions) + + +class EnvSample(unittest.TestCase): + + @isolation.runInSubprocess(env={'_PYTHON_ISOLATION_PROBE': 'set-by-test'}) + def test_env_set(self): + self.assertEqual(os.environ.get('_PYTHON_ISOLATION_PROBE'), 'set-by-test') + + @isolation.runInSubprocess(env={'_PYTHON_ISOLATION_PROBE': None}) + def test_env_unset(self): + self.assertNotIn('_PYTHON_ISOLATION_PROBE', os.environ) + + @isolation.runInSubprocess() + def test_env_inherited(self): + # Without env= the subprocess inherits the parent environment as it is. + self.assertEqual(os.environ.get('_PYTHON_ISOLATION_PROBE'), 'set-by-parent') + + +# TimeoutSample hangs this long, so that the timeout always fires first. +TIMEOUT_HANG = 60.0 +TIMEOUT = 0.5 + + +class TimeoutSample(unittest.TestCase): + + @isolation.runInSubprocess(timeout=TIMEOUT) + def test_hang(self): + time.sleep(TIMEOUT_HANG) diff --git a/Lib/test/support/isolation.py b/Lib/test/support/isolation.py index bc2189329c03997..bb4fa6b003cc20c 100644 --- a/Lib/test/support/isolation.py +++ b/Lib/test/support/isolation.py @@ -78,7 +78,11 @@ def _decode(data): def _remote(detail): # Wrap the subprocess traceback the way concurrent.futures does, so it is - # clearly delimited when shown as the cause. + # clearly delimited when shown as the cause. Return None if the subprocess + # said nothing (a hung one usually does not), so that "raise ... from None" + # suppresses an empty cause. + if not detail: + return None return _RemoteTraceback(f'\n"""\n{detail}"""') @@ -90,7 +94,21 @@ def _check_subprocess_support(): raise unittest.SkipTest('requires subprocess support') -def _run_in_subprocess(module, qualname): +def _child_environ(env): + # Start from the inherited environment, so that *env* only has to name what + # the test changes. + if not env: + return None + environ = dict(os.environ) + for name, value in env.items(): + if value is None: + environ.pop(name, None) + else: + environ[name] = value + return environ + + +def _run_in_subprocess(module, qualname, options, env, timeout): """Run module.qualname (a test method or class) in a fresh subprocess. Return ``(payload, output, returncode)``, where *payload* is the decoded @@ -104,13 +122,22 @@ def _run_in_subprocess(module, qualname): os.close(fd) try: # Pass the config on the command line, not in the environment, so that - # the test cannot pass it on to the processes it spawns itself. Use - # marshal, not json: it is built in, so the child imports nothing that - # the test would not see in a normal test run. - cmd = [sys.executable, '-m', 'test.support.subprocess_runner', + # the test cannot pass it on to the processes it spawns itself, and so + # that it survives the -E and -I options. Use marshal, not json: it is + # built in, so the child imports nothing that the test would not see in + # a normal test run. + cmd = [sys.executable, *options, '-m', 'test.support.subprocess_runner', module, qualname, result_path, marshal.dumps(_child_config()).hex()] - proc = subprocess.run(cmd, capture_output=True) + try: + proc = subprocess.run(cmd, capture_output=True, + env=_child_environ(env), timeout=timeout) + except subprocess.TimeoutExpired as exc: + # Report the hang rather than leaving the test runner stuck. + output = _decode(exc.stdout) + _decode(exc.stderr) + raise _SubprocessTestError( + f'test did not complete in a subprocess ' + f'within {timeout} seconds') from _remote(output) try: with open(result_path, 'rb') as f: payload = marshal.load(f) @@ -173,7 +200,7 @@ def _check_returncode(returncode, output, what): raise exc from _remote(output) -def _isolate_method(func): +def _isolate_method(func, options, env, timeout): @functools.wraps(func) def wrapper(self, /, *args, **kwargs): if runningInSubprocess: @@ -183,7 +210,8 @@ def wrapper(self, /, *args, **kwargs): cls = type(self) qualname = f'{cls.__qualname__}.{func.__name__}' payload, output, returncode = _run_in_subprocess(cls.__module__, - qualname) + qualname, options, + env, timeout) if payload is None: exc = _SubprocessTestError( f'test did not complete in a subprocess (exit code {returncode})') @@ -196,7 +224,7 @@ def wrapper(self, /, *args, **kwargs): return wrapper -def _isolate_class(cls): +def _isolate_class(cls, options, env, timeout): # Unwrap to the plain functions so the replacements can call them with the # runtime cls; a bound classmethod would freeze the decoration-time class # and a subclass would run the fixtures bound to the base class. @@ -217,7 +245,8 @@ def setUpClass(cls): # Run the whole class in a single subprocess and stash the outcomes # for the test methods to replay. payload, output, returncode = _run_in_subprocess(cls.__module__, - cls.__qualname__) + cls.__qualname__, + options, env, timeout) if payload is None: exc = _SubprocessTestError( f'class did not complete in a subprocess (exit code {returncode})') @@ -283,7 +312,7 @@ def _addDuration(self, result, elapsed): return cls -def runInSubprocess(): +def runInSubprocess(*, options=(), env=None, timeout=None): """Decorator to run a test method or class in a fresh subprocess. The decorated test runs in a separate, fresh Python process, so it does not @@ -293,6 +322,16 @@ def runInSubprocess(): once there; when a method is decorated, only that method runs in a subprocess. Decorated methods must take no extra arguments. + *options* is a sequence of interpreter command line options for the + subprocess, and *env* is a mapping of environment variables to set in it, + on top of the inherited environment; a value of ``None`` unsets a variable. + Note that ``-E`` and ``-I`` make the subprocess ignore the ``PYTHON*`` + variables, including ``PYTHONPATH``. + + *timeout* is the number of seconds to wait for the subprocess; the test is + reported as an error if it does not complete in time. By default there is + no timeout, and a hung test is left to the timeout of the test runner. + A failure, error or skip of the whole test is reported for the test, and individual subtests (:meth:`~unittest.TestCase.subTest`) that fail or are skipped are reported individually. The original subprocess traceback is @@ -304,6 +343,6 @@ def runInSubprocess(): """ def decorator(obj): if isinstance(obj, type) and issubclass(obj, unittest.TestCase): - return _isolate_class(obj) - return _isolate_method(obj) + return _isolate_class(obj, options, env, timeout) + return _isolate_method(obj, options, env, timeout) return decorator diff --git a/Lib/test/test_support.py b/Lib/test/test_support.py index 5c702cde04f3896..1d27c7fff7d14af 100644 --- a/Lib/test/test_support.py +++ b/Lib/test/test_support.py @@ -962,6 +962,32 @@ def test_class_subprocess_dying_after_the_tests_is_reported(self): self.assertIn('tearDownClass', str(result.errors[0][0])) self.assertIn(f'exited with code {EXIT_CODE}', result.errors[0][1]) + @support.requires_subprocess() + def test_options_passed_to_subprocess(self): + result = self._run('OptionsSample') + self.assertEqual(result.testsRun, 1) + self.assertEqual(result.failures, []) + self.assertEqual(result.errors, []) + + @support.requires_subprocess() + def test_env_passed_to_subprocess(self): + # The samples check the variable, so set it here to let them tell + # env= from the inherited environment. + with os_helper.EnvironmentVarGuard() as env: + env['_PYTHON_ISOLATION_PROBE'] = 'set-by-parent' + result = self._run('EnvSample') + self.assertEqual(result.testsRun, 3) + self.assertEqual(result.failures, []) + self.assertEqual(result.errors, []) + + @support.requires_subprocess() + def test_timeout_reported_as_error(self): + from test._isolated_sample import TIMEOUT + result = self._run('TimeoutSample') + self.assertEqual(result.testsRun, 1) + self.assertEqual(len(result.errors), 1) + self.assertIn(f'within {TIMEOUT} seconds', result.errors[0][1]) + def test_skipped_without_subprocess_support(self): # On a platform without subprocess support the test is skipped in the # parent, before any subprocess is spawned. _______________________________________________ Python-checkins mailing list -- [email protected] To unsubscribe send an email to [email protected] https://mail.python.org/mailman3//lists/python-checkins.python.org Member address: [email protected]
