.. 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
: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.
import atexit
import os
+import sys
import time
import unittest
from test.support import isolation
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)
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}"""')
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
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)
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:
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})')
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.
# 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})')
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
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
"""
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
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.
projects / thirdparty / Python / cpython.git / commitdiff
