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

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.