gh-152548: Address review of test.support.isolated()

Move the helper into a public test.support.isolation submodule (used as
"from test.support import isolation"), drop the test.support re-export,
and document running_isolated and isolated() under that module.

Replay expected failures and forward subprocess durations to the parent,
so an @expectedfailure isolated test is no longer misreported as an
unexpected success and reported timings reflect the subprocess run.

Add test.test_support.TestIsolated covering the outcomes, subtest
reporting, traceback-as-cause, duration forwarding and the no-subprocess
skip.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: serhiy-storchaka

Doc/library/test.rst

@@ -961,6 +961,8 @@ The :mod:`!test.support` module defines the following functions:
    :mod:`tracemalloc` is enabled.
 
 
+.. currentmodule:: test.support.isolation
+
 .. decorator:: isolated()
 
    Decorator that runs the decorated test in isolation, in a fresh interpreter
@@ -973,10 +975,11 @@ The :mod:`!test.support` module defines the following functions:
    individually.  A reported failure or error shows the original subprocess
    traceback as the cause of the exception.
 
-   When a **method** is decorated, only that method runs in a subprocess;
-   :meth:`~unittest.TestCase.setUp` and :meth:`~unittest.TestCase.tearDown`
-   run both in the parent process (as usual) and in the subprocess around the
-   method.
+   When a **method** is decorated, only that method runs in a subprocess; all
+   fixtures (:meth:`~unittest.TestCase.setUp` / :meth:`~unittest.TestCase.tearDown`,
+   :meth:`~unittest.TestCase.setUpClass` / :meth:`~unittest.TestCase.tearDownClass`
+   and ``setUpModule()`` / ``tearDownModule()``) run both in the parent process
+   (as usual) and in the subprocess around the method.
 
    When a **class** is decorated, the whole class runs in a single subprocess,
    and :meth:`~unittest.TestCase.setUpClass`,
@@ -988,13 +991,11 @@ The :mod:`!test.support` module defines the following functions:
    so it still runs in the parent process too; test it with
    :data:`running_isolated` if needed.
 
-   Fixtures can test :data:`running_isolated` to decide what to run in each
-   process.
-
    The subprocess inherits the enabled resources (``-u``), memory limit
    (``-M``) and verbosity (``-v``) of the parent test run, so that
-   :func:`requires_resource`, :func:`requires`, :func:`bigmemtest` and the like
-   behave consistently in both processes.
+   :func:`~test.support.requires_resource`, :func:`~test.support.requires`,
+   :func:`~test.support.bigmemtest` and the like behave consistently in both
+   processes.
 
    The test is skipped on platforms without subprocess support.
 
@@ -1010,6 +1011,9 @@ The :mod:`!test.support` module defines the following functions:
    to run in the subprocess.
 
 
+.. currentmodule:: test.support
+
+
 .. function:: check_free_after_iterating(test, iter, cls, args=())
 
    Assert instances of *cls* are deallocated after iterating.

Lib/test/_isolated_sample.py

@@ -0,0 +1,74 @@
+"""Sample tests driven by test.test_support.TestIsolated.
+
+This module is imported, never run as a test file, so that
+:func:`test.support.isolation.isolated` has a real, importable target to run in
+a subprocess.  Several of these tests fail, error or are skipped on purpose.
+"""
+
+import time
+import unittest
+from test.support import isolation
+
+# A test in DurationSample sleeps this long in the subprocess; the parent
+# replays it instantly, so a parent-reported duration close to this proves the
+# subprocess timing was forwarded rather than the replay time measured.
+DURATION_SLEEP = 0.2
+
+
+class MethodSample(unittest.TestCase):
+
+    @isolation.isolated()
+    def test_pass(self):
+        self.assertTrue(isolation.running_isolated)
+
+    @isolation.isolated()
+    def test_fail(self):
+        self.assertEqual(1, 2)
+
+    @isolation.isolated()
+    def test_error(self):
+        raise RuntimeError('boom')
+
+    @isolation.isolated()
+    def test_skip(self):
+        self.skipTest('nope')
+
+    @isolation.isolated()
+    @unittest.expectedFailure
+    def test_expected_failure(self):
+        self.assertEqual(1, 2)
+
+    @isolation.isolated()
+    @unittest.expectedFailure
+    def test_unexpected_success(self):
+        pass
+
+
+@isolation.isolated()
+class ClassSample(unittest.TestCase):
+
+    def test_pass(self):
+        self.assertTrue(isolation.running_isolated)
+
+    def test_fail(self):
+        self.assertEqual(1, 2)
+
+    @unittest.expectedFailure
+    def test_expected_failure(self):
+        self.assertEqual(1, 2)
+
+
+class SubtestSample(unittest.TestCase):
+
+    @isolation.isolated()
+    def test_subtests(self):
+        for i in range(3):
+            with self.subTest(i=i):
+                self.assertNotEqual(i, 1)
+
+
+@isolation.isolated()
+class DurationSample(unittest.TestCase):
+
+    def test_slow(self):
+        time.sleep(DURATION_SLEEP)

Lib/test/support/__init__.py

@@ -24,7 +24,6 @@
 __all__ = [
     # globals
     "PIPE_MAX_SIZE", "verbose", "max_memuse", "use_resources", "failfast",
-    "running_isolated",
     # exceptions
     "Error", "TestFailed", "TestDidNotRun", "ResourceDenied",
     # io
@@ -37,7 +36,6 @@
     "check_syntax_error",
     "requires_gzip", "requires_bz2", "requires_lzma", "requires_zstd",
     "bigmemtest", "nomemtest", "bigaddrspacetest", "cpython_only", "get_attribute",
-    "isolated",
     "requires_IEEE_754", "requires_zlib",
     "has_fork_support", "requires_fork",
     "has_subprocess_support", "requires_subprocess",
@@ -1091,12 +1089,6 @@ def wrapper(self, /, *args, **kwargs):
         return wrapper
     return decorator
 
-# Run a test method or class in an isolated subprocess.  Implemented in a
-# dedicated module so that its frames carry the __unittest marker and are
-# stripped from reported tracebacks.
-from test.support._isolation import isolated, running_isolated
-
-
 #=======================================================================
 # Decorator/context manager for running a code in a different locale,
 # correctly resetting it afterwards.

Lib/test/support/_isolation.py Lib/test/support/isolation.pyLib/test/support/_isolation.py renamed to Lib/test/support/isolation.py

@@ -1,4 +1,4 @@
-"""Run tests in isolated subprocesses (the test.support.isolated decorator).
+"""Run tests in isolated subprocesses (the test.support.isolation.isolated decorator).
 
 A failure, error or skip that happens in the subprocess is replayed in the
 parent process so that the test runner records it.  The original (subprocess)
@@ -78,7 +78,7 @@ class _SubprocessTestError(Exception):
 def _remote(detail):
     # Wrap the subprocess traceback the way concurrent.futures does, so it is
     # clearly delimited when shown as the cause.
-    return _RemoteTraceback('\n"""\n%s"""' % detail)
+    return _RemoteTraceback(f'\n"""\n{detail}"""')
 
 
 def _check_subprocess_support():
@@ -92,9 +92,9 @@ def _check_subprocess_support():
 def _run_in_subprocess(module, qualname):
     """Run module.qualname (a test method or class) in a fresh subprocess.
 
-    Return ``(outcomes, output, returncode)``, where *outcomes* is the list of
-    test outcomes decoded from the subprocess, or ``None`` if it did not run to
-    completion (crash, import error, ...).
+    Return ``(payload, output, returncode)``, where *payload* is the decoded
+    ``{'outcomes': ..., 'durations': ...}`` mapping from the subprocess, or
+    ``None`` if it did not run to completion (crash, import error, ...).
     """
     import json
     import subprocess
@@ -110,28 +110,31 @@ def _run_in_subprocess(module, qualname):
         proc = subprocess.run(cmd, capture_output=True, text=True, env=env)
         try:
             with open(result_path, encoding='utf-8') as f:
-                outcomes = json.load(f)
+                payload = json.load(f)
         except (OSError, ValueError):
-            outcomes = None
+            payload = None
     finally:
         try:
             os.unlink(result_path)
         except OSError:
             pass
-    return outcomes, (proc.stdout or '') + (proc.stderr or ''), proc.returncode
+    return payload, (proc.stdout or '') + (proc.stderr or ''), proc.returncode
 
 
 def _replay_outcome(test, outcome):
     kind = outcome['kind']
     detail = outcome['detail']
     if kind == 'skipped':
         test.skipTest(detail)  # the detail is the skip reason, not a traceback
-    elif kind == 'failure':
-        raise test.failureException('test failed in the subprocess') \
-            from _remote(detail)
+    elif kind in ('failure', 'expected_failure'):
+        # An expected failure is replayed like a failure: the wrapper carries
+        # the @expectedFailure marker (copied by functools.wraps), so the parent
+        # records the raised exception as an expectedFailure, not a failure.
+        exc = test.failureException('test failed in the subprocess')
+        raise exc from _remote(detail)
     else:  # 'error'
-        raise _SubprocessTestError('test failed in the subprocess') \
-            from _remote(detail)
+        exc = _SubprocessTestError('test failed in the subprocess')
+        raise exc from _remote(detail)
 
 
 def _replay_outcomes(test, outcomes):
@@ -153,8 +156,8 @@ def _raise_fixture_outcome(outcome):
     # subprocess in a parent-process fixture, so it applies to every test.
     if outcome['kind'] == 'skipped':
         raise unittest.SkipTest(outcome['detail'])
-    raise _SubprocessTestError('class failed in the subprocess') \
-        from _remote(outcome['detail'])
+    exc = _SubprocessTestError('class failed in the subprocess')
+    raise exc from _remote(outcome['detail'])
 
 
 def _isolate_method(func):
@@ -166,21 +169,27 @@ def wrapper(self, /, *args, **kwargs):
         _check_subprocess_support()
         cls = type(self)
         qualname = f'{cls.__qualname__}.{func.__name__}'
-        outcomes, output, returncode = _run_in_subprocess(cls.__module__,
-                                                          qualname)
-        if outcomes is None:
-            raise _SubprocessTestError(
-                f'test did not complete in a subprocess (exit code '
-                f'{returncode})') from _remote(output)
-        _replay_outcomes(self, outcomes)
+        payload, output, returncode = _run_in_subprocess(cls.__module__,
+                                                         qualname)
+        if payload is None:
+            exc = _SubprocessTestError(
+                f'test did not complete in a subprocess (exit code {returncode})')
+            raise exc from _remote(output)
+        # The parent measures this method's own duration (the real cost of the
+        # isolated run, subprocess startup included), so nothing to forward here.
+        _replay_outcomes(self, payload['outcomes'])
     return wrapper
 
 
 def _isolate_class(cls):
+    # Unwrap to the plain functions: the replacements below call them with the
+    # runtime cls, so a subclass of an isolated class runs the fixtures bound to
+    # itself (a bound classmethod would freeze the decoration-time class).
     orig_setUpClass = cls.setUpClass.__func__
     orig_tearDownClass = cls.tearDownClass.__func__
     orig_setUp = cls.setUp
     orig_tearDown = cls.tearDown
+    orig_addDuration = getattr(cls, '_addDuration', None)
 
     def setUpClass(cls):
         if running_isolated:
@@ -189,26 +198,28 @@ def setUpClass(cls):
         _check_subprocess_support()
         # Run the whole class in a single subprocess and stash the outcomes
         # for the wrapped test methods to replay.
-        outcomes, output, returncode = _run_in_subprocess(cls.__module__,
-                                                          cls.__qualname__)
-        if outcomes is None:
-            raise _SubprocessTestError(
-                f'class did not complete in a subprocess (exit code '
-                f'{returncode})') from _remote(output)
+        payload, output, returncode = _run_in_subprocess(cls.__module__,
+                                                         cls.__qualname__)
+        if payload is None:
+            exc = _SubprocessTestError(
+                f'class did not complete in a subprocess (exit code {returncode})')
+            raise exc from _remote(output)
         by_id = {}
-        for outcome in outcomes:
+        for outcome in payload['outcomes']:
             if outcome['fixture']:
                 # A setUpClass()/setUpModule() failure or skip: apply it to the
                 # whole class by raising it here, in the parent's setUpClass().
                 _raise_fixture_outcome(outcome)
             by_id.setdefault(outcome['id'], []).append(outcome)
         cls._isolated_outcomes = by_id
+        cls._isolated_durations = dict(payload.get('durations', ()))
 
     def tearDownClass(cls):
         if running_isolated:
             orig_tearDownClass(cls)
         else:
             cls._isolated_outcomes = None
+            cls._isolated_durations = None
 
     def setUp(self):
         # In the parent the real test does not run, so neither should setUp().
@@ -219,6 +230,15 @@ def tearDown(self):
         if running_isolated:
             orig_tearDown(self)
 
+    def _addDuration(self, result, elapsed):
+        # In the parent, report the per-test duration measured in the subprocess
+        # rather than the replay time (subprocess startup is paid once, in
+        # setUpClass).
+        if not running_isolated:
+            durations = getattr(type(self), '_isolated_durations', None) or {}
+            elapsed = durations.get(self.id(), elapsed)
+        orig_addDuration(self, result, elapsed)
+
     def replay(self):
         by_id = getattr(type(self), '_isolated_outcomes', None) or {}
         _replay_outcomes(self, by_id.get(self.id(), []))
@@ -227,6 +247,8 @@ def replay(self):
     cls.tearDownClass = classmethod(tearDownClass)
     cls.setUp = setUp
     cls.tearDown = tearDown
+    if orig_addDuration is not None:
+        cls._addDuration = _addDuration
     for name in unittest.TestLoader().getTestCaseNames(cls):
         method = getattr(cls, name)
         @functools.wraps(method)
@@ -254,10 +276,11 @@ def isolated():
     shown as the cause of a reported failure or error.  Use
     :data:`running_isolated` in fixtures to choose what to run in the subprocess.
 
-    The test is skipped on platforms without subprocess support.
+    The test is skipped on platforms without subprocess support, since it must
+    spawn one.
     """
     def decorator(obj):
-        if isinstance(obj, type):
+        if isinstance(obj, type) and issubclass(obj, unittest.TestCase):
             return _isolate_class(obj)
         return _isolate_method(obj)
     return decorator

Lib/test/support/subprocess_runner.py

@@ -1,7 +1,7 @@
 """Run a single test method in this (sub)process and report the result.
 
 Invoked as ``python -m test.support.subprocess_runner MODULE QUALNAME OUTFILE``
-by :func:`test.support.isolated`.  The outcome of the test (including
+by :func:`test.support.isolation.isolated`.  The outcome of the test (including
 that of each individual subtest) is written as JSON to OUTFILE.  This module is
 not meant to be imported.
 """
@@ -25,11 +25,24 @@
 
 # Mirror the parent's regrtest configuration (-u, -M, -v, ...) before importing
 # the test, so resource gating and bigmem sizing match the parent process.
-from test.support._isolation import _apply_child_config
+from test.support.isolation import _apply_child_config
 _apply_child_config()
 
+
+class _Result(unittest.TestResult):
+    # Capture per-test durations keyed by test id, so the parent can report the
+    # subprocess timings instead of its own replay time.
+    def __init__(self):
+        super().__init__()
+        self.id_durations = []
+
+    def addDuration(self, test, elapsed):
+        super().addDuration(test, elapsed)
+        self.id_durations.append((test.id(), elapsed))
+
+
 suite = unittest.TestLoader().loadTestsFromName(f'{module}.{qualname}')
-result = unittest.TestResult()
+result = _Result()
 suite.run(result)
 
 
@@ -50,9 +63,12 @@ def _outcome(kind, test, detail):
 
 outcomes = [_outcome('failure', t, tb) for t, tb in result.failures]
 outcomes += [_outcome('error', t, tb) for t, tb in result.errors]
+outcomes += [_outcome('expected_failure', t, tb)
+             for t, tb in result.expectedFailures]
 outcomes += [_outcome('skipped', t, reason) for t, reason in result.skipped]
 
+payload = {'outcomes': outcomes, 'durations': result.id_durations}
 with open(outfile, 'w', encoding='utf-8') as f:
-    json.dump(outcomes, f)
+    json.dump(payload, f)
 
 sys.exit(0)