gh-152548: Add options, env and timeout parameters to runInSubprocess()

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.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: serhiy-storchaka

Doc/library/test.rst

@@ -963,7 +963,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
@@ -997,6 +997,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.
 
 

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()
+
+
+@isolation.runInSubprocess(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)

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

Lib/test/test_support.py

@@ -1205,6 +1205,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.