grthtrhthjhtyjytjytkergtrhtrjytjerhrfh4:24 29/09/2026"""Run tests in isolated subprocesses (the test.support.isolation.runInSubprocess 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)
traceback is attached as the cause of the replayed exception, the same way
:mod:`concurrent.futures` surfaces tracebacks from worker processes.
"""

import functools
import os
import sys
import unittest

# Let unittest strip this module's frames from tracebacks, so only the original
# subprocess traceback (attached as the cause) is shown, not the replay frames.
__unittest = True

# test.support globals set by regrtest (libregrtest/setup.py) that affect how
# tests run and which are skipped at runtime in the subprocess.
_PROPAGATED_CONFIG = (
    'use_resources',                    # -u (is_resource_enabled/requires)
    'max_memuse', 'real_max_memuse',    # -M (bigmemtest)
    'verbose',                          # -v
    'failfast',                         # -f
)

def _child_config():
    import test.support as support
    return {name: getattr(support, name) for name in _PROPAGATED_CONFIG}

def _apply_child_config(config):
    """Set up the child to run the test like a regrtest worker would.

    Mark this process as the subprocess, mirror the parent's -u/-M/-v config,
    then suppress the Windows CRT assertion dialogs, which would block a debug
    build on a modal dialog and hang the parent.
    """
    global runningInSubprocess
    import marshal
    import test.support as support
    runningInSubprocess = True
    for name, value in marshal.loads(bytes.fromhex(config)).items():
        setattr(support, name, value)
    support.suppress_msvcrt_asserts(support.verbose >= 2)

# True inside the subprocess spawned by @runInSubprocess(), set by
# _apply_child_config() before the test is imported.  Fixtures can test it to
# decide what to run in the subprocess as opposed to the parent process.
runningInSubprocess = False


class _RemoteTraceback(Exception):
    """Carry a formatted traceback string from the subprocess for display.

    Attached as the ``__cause__`` of the replayed failure/error, so that the
    original traceback is shown by the traceback machinery.
    """
    def __init__(self, tb):
        self.tb = tb

    def __str__(self):
        return self.tb


class _SubprocessTestError(Exception):
    """Replay a subprocess error (as opposed to a failure) in the parent."""


def _decode(data):
    # Decode the child output, which is only ever shown as a diagnostic: an
    # undecodable byte must not hide the failure it is part of.
    if not data:
        return ''
    import locale
    encoding = 'utf-8' if sys.flags.utf8_mode else locale.getencoding()
    return data.decode(encoding, 'backslashreplace').replace('\r\n', '\n')


def _remote(detail):
    # Wrap the subprocess traceback the way concurrent.futures does, so it is
    # clearly delimited when shown as the cause.
    return _RemoteTraceback(f'\n"""\n{detail}"""')


def _check_subprocess_support():
    # runInSubprocess() always runs the test in a subprocess, so skip (in the
    # parent) on platforms that do not support spawning one.
    import test.support as support
    if not support.has_subprocess_support:
        raise unittest.SkipTest('requires subprocess support')


def _run_in_subprocess(module, qualname):
    """Run module.qualname (a test method or class) in a fresh subprocess.

    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 marshal
    import subprocess
    import tempfile
    fd, result_path = tempfile.mkstemp(suffix='.json')
    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',
               module, qualname, result_path,
               marshal.dumps(_child_config()).hex()]
        proc = subprocess.run(cmd, capture_output=True)
        try:
            with open(result_path, 'rb') as f:
                payload = marshal.load(f)
        except (OSError, EOFError, ValueError):
            payload = None
    finally:
        try:
            os.unlink(result_path)
        except OSError:
            pass
    return payload, _decode(proc.stdout) + _decode(proc.stderr), 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 in ('failure', 'expected_failure'):
        # Replay an expected failure like a failure: the wrapper keeps the
        # @expectedFailure marker (via functools.wraps), so the parent records
        # the raised exception as an expectedFailure.
        exc = test.failureException('test failed in the subprocess')
        raise exc from _remote(detail)
    else:  # 'error'
        exc = _SubprocessTestError('test failed in the subprocess')
        raise exc from _remote(detail)


def _replay_outcomes(test, outcomes):
    # Replay each subtest outcome in its own subTest() context so that they are
    # reported individually, then replay the whole-test outcome (if any).
    main = []
    for outcome in outcomes:
        if outcome['subtest']:
            with test.subTest(outcome['desc']):
                _replay_outcome(test, outcome)
        else:
            main.append(outcome)
    for outcome in main:
        _replay_outcome(test, outcome)


def _raise_fixture_outcome(outcome):
    # Reproduce a setUpClass()/setUpModule() failure or skip from the
    # subprocess in a parent-process fixture, so it applies to every test.
    if outcome['kind'] == 'skipped':
        raise unittest.SkipTest(outcome['detail'])
    exc = _SubprocessTestError('class failed in the subprocess')
    raise exc from _remote(outcome['detail'])


def _check_returncode(returncode, output, what):
    # The subprocess writes its result before exiting, so a non-zero exit code
    # means it died afterwards, during finalization, unnoticed by the result.
    if returncode:
        exc = _SubprocessTestError(
            f'the subprocess exited with code {returncode} '
            f'after running the {what}')
        raise exc from _remote(output)


def _isolate_method(func):
    @functools.wraps(func)
    def wrapper(self, /, *args, **kwargs):
        if runningInSubprocess:
            # Already running in the subprocess: run the real test.
            return func(self, *args, **kwargs)
        _check_subprocess_support()
        cls = type(self)
        qualname = f'{cls.__qualname__}.{func.__name__}'
        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 the outcomes first: a failure of the test itself is more useful.
        _replay_outcomes(self, payload['outcomes'])
        _check_returncode(returncode, output, 'test')
    return wrapper


def _isolate_class(cls):
    # 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.
    orig_setUpClass = cls.setUpClass.__func__
    orig_tearDownClass = cls.tearDownClass.__func__
    # Hook the _call*() indirections rather than setUp(), tearDown() and the
    # test methods themselves, to cover what a subclass adds or overrides too.
    orig_callSetUp = cls._callSetUp
    orig_callTearDown = cls._callTearDown
    orig_callTestMethod = cls._callTestMethod
    orig_addDuration = cls._addDuration

    def setUpClass(cls):
        if runningInSubprocess:
            orig_setUpClass(cls)
            return
        _check_subprocess_support()
        # 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__)
        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 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', ()))
        # Report the crash from tearDownClass(), after replaying the outcomes.
        cls._isolated_exit = (returncode, output)

    def tearDownClass(cls):
        if runningInSubprocess:
            orig_tearDownClass(cls)
            return
        cls._isolated_outcomes = None
        cls._isolated_durations = None
        # Missing if an overriding setUpClass() bypassed the subprocess.
        exited = getattr(cls, '_isolated_exit', None)
        cls._isolated_exit = None
        if exited is not None:
            _check_returncode(*exited, 'class')

    def _callSetUp(self):
        # In the parent the real test does not run, so neither should setUp().
        if runningInSubprocess:
            orig_callSetUp(self)

    def _callTearDown(self):
        if runningInSubprocess:
            orig_callTearDown(self)

    def _callTestMethod(self, method):
        if runningInSubprocess:
            orig_callTestMethod(self, method)
            return
        by_id = getattr(type(self), '_isolated_outcomes', None)
        if by_id is None:
            raise _SubprocessTestError(
                f'{type(self).__name__} did not run in a subprocess; '
                f'an overriding setUpClass() must call super().setUpClass()')
        _replay_outcomes(self, by_id.get(self.id(), []))

    def _addDuration(self, result, elapsed):
        # In the parent, report the subprocess timing rather than the (instant)
        # replay time; subprocess startup is paid once, in setUpClass.
        if not runningInSubprocess:
            durations = getattr(type(self), '_isolated_durations', None) or {}
            elapsed = durations.get(self.id(), elapsed)
        orig_addDuration(self, result, elapsed)

    cls.setUpClass = classmethod(setUpClass)
    cls.tearDownClass = classmethod(tearDownClass)
    cls._callSetUp = _callSetUp
    cls._callTearDown = _callTearDown
    cls._callTestMethod = _callTestMethod
    cls._addDuration = _addDuration
    return cls


def runInSubprocess():
    """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
    share global or interpreter state with the rest of the test run.  When a
    :class:`~unittest.TestCase` subclass is decorated, the whole class runs in a
    single subprocess and its ``setUpClass()``/``setUpModule()`` fixtures run
    once there; when a method is decorated, only that method runs in a
    subprocess.  Decorated methods must take no extra arguments.

    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
    shown as the cause of a reported failure or error.  Use
    :data:`runningInSubprocess` in fixtures to choose what to run in the subprocess.

    The test is skipped on platforms without subprocess support, since it must
    spawn one.
    """
    def decorator(obj):
        if isinstance(obj, type) and issubclass(obj, unittest.TestCase):
            return _isolate_class(obj)
        return _isolate_method(obj)
    return decorator