From 92efaff49b100371d66e356b0ab62c35ec7c5fe8 Mon Sep 17 00:00:00 2001 From: Vyron Vasileiadis Date: Mon, 27 Jul 2026 21:24:10 +0300 Subject: [PATCH 1/6] gh-154751: Fix use-after-free in curses.initscr() after newterm() (GH-154752) initscr() called while a newterm() screen is current returned a second window object over that screen's standard window, with no reference to the screen. Either wrapper could then free the window used by the other. Return the screen's own standard window instead. --- Lib/test/test_curses.py | 18 ++++++++++++++++++ Modules/_cursesmodule.c | 18 +++++++++++++++++- 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/Lib/test/test_curses.py b/Lib/test/test_curses.py index ad5893e6754f68c..31b7371abd32300 100644 --- a/Lib/test/test_curses.py +++ b/Lib/test/test_curses.py @@ -3072,6 +3072,24 @@ def test_new_prescr(self): del screen gc_collect() + def test_initscr_after_newterm_keeps_screen_alive(self): + # initscr() called while a newterm() screen is current returns that + # screen's own standard window, so the window keeps the screen alive. + # It used to be a second wrapper created without a screen: using it + # after the screen was collected read freed memory, and both wrappers + # could delwin() the same window. + s1 = self.make_pty() + s2 = self.make_pty() + screen1 = curses.newterm('xterm', s1, s1) + screen2 = curses.newterm('xterm', s2, s2) + curses.set_term(screen1) + win = curses.initscr() + self.assertIs(win, screen1.stdscr) + curses.set_term(screen2) + del screen1 + gc_collect() + win.addstr(0, 0, 'x') + @cpython_only def test_disallow_instantiation(self): # The screen type cannot be instantiated directly (bpo-43916). diff --git a/Modules/_cursesmodule.c b/Modules/_cursesmodule.c index b2d745332317a36..2b580c3475e6d93 100644 --- a/Modules/_cursesmodule.c +++ b/Modules/_cursesmodule.c @@ -6573,7 +6573,23 @@ _curses_initscr_impl(PyObject *module) _curses_set_null_error(state, "wrefresh", "initscr"); return NULL; } - PyObject *winobj = PyCursesWindow_New(state, stdscr, NULL, NULL, NULL); + if (state->topscreen != NULL) { + /* The current screen is one made by newterm(); return its own + standard window instead of a second wrapper over the same + WINDOW, which would delwin() it on its own. */ + PyCursesScreenObject *so = (PyCursesScreenObject *)state->topscreen; + if (so->stdscr_win != NULL) { + if (curses_update_screen_encoding(so->stdscr_win) < 0) { + return NULL; + } + return Py_NewRef(so->stdscr_win); + } + } + /* Attach the current screen, like newwin(), newpad() and getwin() do, + so that the window keeps its screen alive. It is NULL for the + screen created by initscr(), which has no screen object. */ + PyObject *winobj = PyCursesWindow_New(state, stdscr, NULL, NULL, + state->topscreen); if (winobj == NULL) { return NULL; } From b099df5f177e43ebefde24a6714729277aa1bf8c Mon Sep 17 00:00:00 2001 From: Vyron Vasileiadis Date: Mon, 27 Jul 2026 21:59:46 +0300 Subject: [PATCH 2/6] gh-154749: Reject a terminal-less screen in curses.set_term() (GH-154750) set_term() accepted a screen returned by new_prescr(), which has no terminal, and the next refresh crashed inside curses. Raise curses.error instead. --- Doc/library/curses.rst | 2 ++ Lib/test/test_curses.py | 11 +++++++++++ Modules/_cursesmodule.c | 8 +++++++- 3 files changed, 20 insertions(+), 1 deletion(-) diff --git a/Doc/library/curses.rst b/Doc/library/curses.rst index 9d0bb239af06dfc..a833914a5d56d44 100644 --- a/Doc/library/curses.rst +++ b/Doc/library/curses.rst @@ -129,6 +129,8 @@ Initialization and termination and return the previously current screen. Returns ``None`` if the previous screen was the one created by :func:`initscr`. + Raises :exc:`error` if *screen* has no terminal, + as is the case for a screen returned by :func:`new_prescr`. .. versionadded:: next diff --git a/Lib/test/test_curses.py b/Lib/test/test_curses.py index 31b7371abd32300..441eafd6e0d4776 100644 --- a/Lib/test/test_curses.py +++ b/Lib/test/test_curses.py @@ -3072,6 +3072,17 @@ def test_new_prescr(self): del screen gc_collect() + @requires_curses_func('new_prescr') + def test_set_term_prescr_screen(self): + # A new_prescr() screen has no terminal, so it cannot become the + # current one. It used to be accepted, and the next refresh then + # crashed inside curses. + s = self.make_pty() + screen = curses.newterm('xterm', s, s) + self.assertRaises(curses.error, curses.set_term, curses.new_prescr()) + # The current screen is unchanged, so refreshing it still works. + screen.stdscr.refresh() + def test_initscr_after_newterm_keeps_screen_alive(self): # initscr() called while a newterm() screen is current returns that # screen's own standard window, so the window keeps the screen alive. diff --git a/Modules/_cursesmodule.c b/Modules/_cursesmodule.c index 2b580c3475e6d93..7314708cff7a814 100644 --- a/Modules/_cursesmodule.c +++ b/Modules/_cursesmodule.c @@ -6850,11 +6850,17 @@ _curses_set_term(PyObject *module, PyObject *screen) if (so == NULL) { return NULL; } + cursesmodule_state *state = get_cursesmodule_state(module); + if (so->stdscr_win == NULL) { + /* A screen from new_prescr() has no terminal, so it cannot become the + current one: a later refresh would dereference NULL in curses. */ + PyErr_SetString(state->error, "the screen has no terminal"); + return NULL; + } set_term(so->screen); if (!update_lines_cols(module)) { return NULL; } - cursesmodule_state *state = get_cursesmodule_state(module); PyObject *prev = state->topscreen; /* steal the owned reference */ state->topscreen = Py_NewRef(screen); return prev != NULL ? prev : Py_NewRef(Py_None); From bde526d7be55014ec49fd44b831c948664ce508b Mon Sep 17 00:00:00 2001 From: Serhiy Storchaka Date: Mon, 27 Jul 2026 22:02:59 +0300 Subject: [PATCH 3/6] gh-152548: Add a test.support.isolation.runInSubprocess() decorator (GH-152551) Run a test in a fresh interpreter subprocess, so that it does not share global or interpreter state with the rest of the test run. It can decorate a test method (only that method runs in a subprocess) or a TestCase subclass (the whole class runs in one subprocess, with its setUpClass()/setUp()/tearDown()/ tearDownClass() running there rather than in the parent). Failures, errors and skips, including those of individual subtests, are reported for the test; a failure or an error shows the original subprocess traceback. The subprocess inherits the parent's resource, memory, verbosity and failfast configuration, so that requires_resource(), bigmemtest() and similar behave the same in both processes. A decorated test is skipped where subprocesses are unavailable, since it must spawn one. The test.support.isolation.runningInSubprocess flag is true in the subprocess, so that fixtures can choose what to run there. Co-Authored-By: Claude Opus 4.8 --- Doc/library/test.rst | 53 ++++ Lib/test/_isolated_sample.py | 111 +++++++ Lib/test/support/isolation.py | 290 ++++++++++++++++++ Lib/test/support/subprocess_runner.py | 79 +++++ Lib/test/test_support.py | 121 ++++++++ ...-06-29-10-14-09.gh-issue-152548.Khw9J7.rst | 3 + 6 files changed, 657 insertions(+) create mode 100644 Lib/test/_isolated_sample.py create mode 100644 Lib/test/support/isolation.py create mode 100644 Lib/test/support/subprocess_runner.py create mode 100644 Misc/NEWS.d/next/Tests/2026-06-29-10-14-09.gh-issue-152548.Khw9J7.rst diff --git a/Doc/library/test.rst b/Doc/library/test.rst index 4e21e1ded82724c..660847ae3fe3c85 100644 --- a/Doc/library/test.rst +++ b/Doc/library/test.rst @@ -961,6 +961,59 @@ The :mod:`!test.support` module defines the following functions: :mod:`tracemalloc` is enabled. +.. currentmodule:: test.support.isolation + +.. decorator:: runInSubprocess() + + 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 + rest of the test run. It can decorate a test method or a whole + :class:`~unittest.TestCase` subclass. Decorated methods must take no extra + arguments. A failure, error or skip in the subprocess is reported for the + corresponding test, and individual :meth:`subtests + ` that fail or are skipped are reported + 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; 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`, + :meth:`~unittest.TestCase.tearDownClass`, :meth:`~unittest.TestCase.setUp` + and :meth:`~unittest.TestCase.tearDown` run once each in the subprocess and + are skipped in the parent process. A failure or skip of + :meth:`~unittest.TestCase.setUpClass` in the subprocess is reported for the + whole class. ``setUpModule()`` cannot be controlled by a class decorator, + so it still runs in the parent process too; test it with + :data:`runningInSubprocess` if needed. + + The subprocess inherits the enabled resources (``-u``), memory limit + (``-M``) and verbosity (``-v``) of the parent test run, so that + :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. + + +.. data:: runningInSubprocess + + ``True`` while the code runs in the isolated subprocess spawned by + :func:`runInSubprocess`, and ``False`` otherwise (including in the parent + process and in a normal, non-isolated test run). Fixtures such as + :meth:`~unittest.TestCase.setUp`, :meth:`~unittest.TestCase.tearDown`, + :meth:`~unittest.TestCase.setUpClass`, :meth:`~unittest.TestCase.tearDownClass`, + ``setUpModule()`` and ``tearDownModule()`` can test it to choose which code + 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. diff --git a/Lib/test/_isolated_sample.py b/Lib/test/_isolated_sample.py new file mode 100644 index 000000000000000..360a27a2b081173 --- /dev/null +++ b/Lib/test/_isolated_sample.py @@ -0,0 +1,111 @@ +"""Sample tests driven by test.test_support.TestIsolated. + +This module is imported, never run as a test file, so that +:func:`test.support.isolation.runInSubprocess` 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 + +# DurationSample sleeps this long in the subprocess; a parent-reported duration +# close to it proves the subprocess timing was forwarded, not the replay time. +DURATION_SLEEP = 0.2 + + +class MethodSample(unittest.TestCase): + + @isolation.runInSubprocess() + def test_pass(self): + self.assertTrue(isolation.runningInSubprocess) + + @isolation.runInSubprocess() + def test_fail(self): + self.assertEqual(1, 2) + + @isolation.runInSubprocess() + def test_error(self): + raise RuntimeError('boom') + + @isolation.runInSubprocess() + def test_skip(self): + self.skipTest('nope') + + @isolation.runInSubprocess() + @unittest.expectedFailure + def test_expected_failure(self): + self.assertEqual(1, 2) + + @isolation.runInSubprocess() + @unittest.expectedFailure + def test_unexpected_success(self): + pass + + +@isolation.runInSubprocess() +class ClassSample(unittest.TestCase): + + def test_pass(self): + self.assertTrue(isolation.runningInSubprocess) + + def test_fail(self): + self.assertEqual(1, 2) + + @unittest.expectedFailure + def test_expected_failure(self): + self.assertEqual(1, 2) + + +class SubtestSample(unittest.TestCase): + + @isolation.runInSubprocess() + def test_subtests(self): + for i in range(3): + with self.subTest(i=i): + self.assertNotEqual(i, 1) + + +@isolation.runInSubprocess() +class DurationSample(unittest.TestCase): + + def test_slow(self): + time.sleep(DURATION_SLEEP) + + +@isolation.runInSubprocess() +class SubclassingSample(unittest.TestCase): + # setUpClass must run bound to the runtime class, so a subclass sees its own + # name here rather than the base class's. + + @classmethod + def setUpClass(cls): + cls.setup_class_name = cls.__name__ + + def setUp(self): + self.set_up = True + + def test_runtime_class(self): + self.assertEqual(self.setup_class_name, type(self).__name__) + + +class SubclassSample(SubclassingSample): + # What a subclass adds or overrides must run in the subprocess too. + + def setUp(self): + super().setUp() + self.set_up_in_subclass = True + + def test_added_in_subclass(self): + self.assertTrue(isolation.runningInSubprocess) + self.assertTrue(self.set_up) + self.assertTrue(self.set_up_in_subclass) + + +class BrokenSubclassSample(SubclassingSample): + # An overriding setUpClass() that does not call super() bypasses the + # subprocess entirely. + + @classmethod + def setUpClass(cls): + pass diff --git a/Lib/test/support/isolation.py b/Lib/test/support/isolation.py new file mode 100644 index 000000000000000..f449bf44034da35 --- /dev/null +++ b/Lib/test/support/isolation.py @@ -0,0 +1,290 @@ +"""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 _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_outcomes(self, payload['outcomes']) + 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', ())) + + def tearDownClass(cls): + if runningInSubprocess: + orig_tearDownClass(cls) + else: + cls._isolated_outcomes = None + cls._isolated_durations = None + + 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 diff --git a/Lib/test/support/subprocess_runner.py b/Lib/test/support/subprocess_runner.py new file mode 100644 index 000000000000000..90d74cc757d878d --- /dev/null +++ b/Lib/test/support/subprocess_runner.py @@ -0,0 +1,79 @@ +"""Run a single test method in this (sub)process and report the result. + +Invoked as ``python -m test.support.subprocess_runner MODULE QUALNAME OUTFILE +CONFIG`` by :func:`test.support.isolation.runInSubprocess`. CONFIG is the +marshalled test.support configuration of the parent test run, as a hex string. +The outcome of the test (including that of each individual subtest) is +marshalled to OUTFILE. This module is not meant to be imported. + +Import as little as possible before running the test: every module imported +here is state that the test would not see in a normal test run. +""" + +import marshal +import sys +import unittest +from unittest.case import _SubTest + +if __name__ != '__main__': + raise ImportError('this module cannot be directly imported') + +if len(sys.argv) != 5: + print('usage: python -m test.support.subprocess_runner ' + 'MODULE QUALNAME OUTFILE CONFIG', file=sys.stderr) + sys.exit(2) + +module, qualname, outfile, config = sys.argv[1:] + +# Set up the child before importing the test. +from test.support.isolation import _apply_child_config +_apply_child_config(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)) + + +# Resolve the qualname in the imported module, rather than letting +# loadTestsFromName() guess where the module name ends: it guesses by trying +# imports that fail, and a failing import pulls in importlib.resources. +__import__(module) +suite = unittest.TestLoader().loadTestsFromName(qualname, sys.modules[module]) +result = _Result() +suite.run(result) + + +def _outcome(kind, test, detail): + subtest = isinstance(test, _SubTest) + real = test.test_case if subtest else test + return { + 'kind': kind, + 'subtest': subtest, + 'desc': test._subDescription() if subtest else '', + # id() groups outcomes by test method; a non-TestCase (e.g. an + # _ErrorHolder) marks a setUpClass()/setUpModule() fixture failure. + 'id': real.id(), + 'fixture': not isinstance(real, unittest.TestCase), + 'detail': 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, 'wb') as f: + marshal.dump(payload, f) + +sys.exit(0) diff --git a/Lib/test/test_support.py b/Lib/test/test_support.py index d556f96bc532ed1..4b9bc245d6f78a8 100644 --- a/Lib/test/test_support.py +++ b/Lib/test/test_support.py @@ -19,6 +19,7 @@ import warnings from test import support +from test.support import isolation from test.support import hashlib_helper from test.support import import_helper from test.support import os_helper @@ -1075,5 +1076,125 @@ def test_disable_hash_md5_in_fips_mode_allow_all(self): self.assertIsInstance(h, self._hashlib.HASH) +class TestIsolated(unittest.TestCase): + # Drive the sample tests in test._isolated_sample (which really spawn + # subprocesses through @isolation.runInSubprocess()) under a private + # TestResult, and check that each subprocess outcome is replayed in the parent. + + @staticmethod + def _run(name): + suite = unittest.TestLoader().loadTestsFromName( + 'test._isolated_sample.' + name) + result = unittest.TestResult() + suite.run(result) + return result + + @staticmethod + def _names(items): + # Map outcome entries (which are (test, detail) pairs, except + # unexpectedSuccesses which are bare tests) to their method names. + names = [] + for item in items: + test = item[0] if isinstance(item, tuple) else item + names.append(test.id().rpartition('.')[2]) + return sorted(names) + + @support.requires_subprocess() + def test_method_outcomes(self): + result = self._run('MethodSample') + self.assertEqual(result.testsRun, 6) + self.assertEqual(self._names(result.failures), ['test_fail']) + self.assertEqual(self._names(result.errors), ['test_error']) + self.assertEqual(self._names(result.skipped), ['test_skip']) + self.assertEqual(self._names(result.expectedFailures), + ['test_expected_failure']) + self.assertEqual(self._names(result.unexpectedSuccesses), + ['test_unexpected_success']) + + @support.requires_subprocess() + def test_class_outcomes(self): + result = self._run('ClassSample') + self.assertEqual(result.testsRun, 3) + self.assertEqual(self._names(result.failures), ['test_fail']) + self.assertEqual(self._names(result.expectedFailures), + ['test_expected_failure']) + self.assertEqual(result.errors, []) + self.assertEqual(result.unexpectedSuccesses, []) + + @support.requires_subprocess() + def test_subtests_reported_individually(self): + result = self._run('SubtestSample') + self.assertEqual(result.testsRun, 1) + self.assertEqual(len(result.failures), 1) + test, _ = result.failures[0] + self.assertIn('i=1', str(test)) + + @support.requires_subprocess() + def test_skip_reason_propagated(self): + result = self._run('MethodSample.test_skip') + self.assertEqual([reason for _, reason in result.skipped], ['nope']) + + @support.requires_subprocess() + def test_subprocess_traceback_is_cause(self): + result = self._run('MethodSample.test_fail') + self.assertEqual(len(result.failures), 1) + _, tb = result.failures[0] + # The real assertion that failed in the subprocess is shown ... + self.assertIn('self.assertEqual(1, 2)', tb) + # ... as the direct cause of the replayed failure ... + self.assertIn('direct cause', tb) + # ... without leaking the parent-side replay frames. + self.assertNotIn('isolation.py', tb) + + @support.requires_subprocess() + def test_durations_forwarded_for_class(self): + from test._isolated_sample import DURATION_SLEEP + result = unittest.TestResult() + suite = unittest.TestLoader().loadTestsFromName( + 'test._isolated_sample.DurationSample') + suite.run(result) + # The duration reported in the parent is the one measured in the + # subprocess (around the sleep), not the near-instant replay time. + self.assertEqual(len(result.collectedDurations), 1) + name, elapsed = result.collectedDurations[0] + self.assertEqual(name.split()[0], 'test_slow') + self.assertGreaterEqual(elapsed, DURATION_SLEEP / 2) + + @support.requires_subprocess() + def test_subclass_of_isolated_class(self): + # Both samples pass only if the fixtures are bound to the runtime class + # and what the subclass adds or overrides runs in the subprocess. + for name, count in (('SubclassingSample', 1), ('SubclassSample', 2)): + with self.subTest(sample=name): + result = self._run(name) + self.assertEqual(result.testsRun, count) + self.assertEqual(result.failures, []) + self.assertEqual(result.errors, []) + + @support.requires_subprocess() + def test_subclass_bypassing_setupclass_is_reported(self): + # A class that never ran in a subprocess must error out, not pass with + # no outcome to replay. + result = self._run('BrokenSubclassSample') + self.assertEqual(result.testsRun, 1) + self.assertEqual(len(result.errors), 1) + self.assertIn('did not run in a subprocess', 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. + calls = [] + orig = isolation._run_in_subprocess + with support.swap_attr(support, 'has_subprocess_support', False): + isolation._run_in_subprocess = lambda *a, **k: calls.append(a) + try: + result = self._run('MethodSample.test_pass') + finally: + isolation._run_in_subprocess = orig + self.assertEqual(result.testsRun, 1) + self.assertEqual(len(result.skipped), 1) + self.assertEqual(calls, []) + + if __name__ == '__main__': unittest.main() diff --git a/Misc/NEWS.d/next/Tests/2026-06-29-10-14-09.gh-issue-152548.Khw9J7.rst b/Misc/NEWS.d/next/Tests/2026-06-29-10-14-09.gh-issue-152548.Khw9J7.rst new file mode 100644 index 000000000000000..e8c81ed8a1f7ca9 --- /dev/null +++ b/Misc/NEWS.d/next/Tests/2026-06-29-10-14-09.gh-issue-152548.Khw9J7.rst @@ -0,0 +1,3 @@ +Add the :func:`test.support.isolation.runInSubprocess` decorator +to run a test method or ``TestCase`` subclass in a fresh interpreter subprocess, +isolated from the rest of the test run. From 85c8bcee6780bd7e6730fe3550e9b65bbf62ff56 Mon Sep 17 00:00:00 2001 From: k00shi <265808442+k00shi@users.noreply.github.com> Date: Tue, 28 Jul 2026 03:35:40 +0800 Subject: [PATCH 4/6] gh-152912: Fix audit hook exception check in sys.addaudithook() (GH-152913) --- Doc/c-api/sys.rst | 9 +++++++-- Lib/test/audit-tests.py | 10 ++++++++++ Lib/test/test_audit.py | 3 +++ .../2026-07-03-02-11-00.gh-issue-152912.Mv3KpR.rst | 1 + Python/sysmodule.c | 4 ++-- 5 files changed, 23 insertions(+), 4 deletions(-) create mode 100644 Misc/NEWS.d/next/Library/2026-07-03-02-11-00.gh-issue-152912.Mv3KpR.rst diff --git a/Doc/c-api/sys.rst b/Doc/c-api/sys.rst index ee73c1c8adaa7b3..0a446f86e22eaf6 100644 --- a/Doc/c-api/sys.rst +++ b/Doc/c-api/sys.rst @@ -433,7 +433,7 @@ accessible to C code. They all work with the current interpreter thread's This function is safe to call before :c:func:`Py_Initialize`. When called after runtime initialization, existing audit hooks are notified and may silently abort the operation by raising an error subclassed from - :class:`Exception` (other errors will not be silenced). + :class:`RuntimeError` (other errors will not be silenced). The hook function is always called with an :term:`attached thread state` by the Python interpreter that raised the event. @@ -447,7 +447,7 @@ accessible to C code. They all work with the current interpreter thread's If the interpreter is initialized, this function raises an auditing event ``sys.addaudithook`` with no arguments. If any existing hooks raise an - exception derived from :class:`Exception`, the new hook will not be + exception derived from :class:`RuntimeError`, the new hook will not be added and the exception is cleared. As a result, callers cannot assume that their hook has been added unless they control all existing hooks. @@ -462,6 +462,11 @@ accessible to C code. They all work with the current interpreter thread's .. versionadded:: 3.8 + .. versionchanged:: 3.8.1 + + Exceptions derived from :class:`Exception` but not :class:`RuntimeError` + are no longer suppressed. + .. _processcontrol: diff --git a/Lib/test/audit-tests.py b/Lib/test/audit-tests.py index 8be5bf8aa4f5469..53bd369fabe21f7 100644 --- a/Lib/test/audit-tests.py +++ b/Lib/test/audit-tests.py @@ -109,6 +109,16 @@ def test_block_add_hook_baseexception(): pass +def test_block_add_hook_valueerror(): + # Non-RuntimeError exceptions (like ValueError) should propagate out + with assertRaises(ValueError): + with TestHook( + raise_on_events="sys.addaudithook", exc_type=ValueError + ) as hook1: + with TestHook() as hook2: + pass + + def test_marshal(): import marshal o = ("a", "b", "c", 1, 2, 3) diff --git a/Lib/test/test_audit.py b/Lib/test/test_audit.py index db4e1eb9999c1fa..e9e546fdbfd6fbe 100644 --- a/Lib/test/test_audit.py +++ b/Lib/test/test_audit.py @@ -58,6 +58,9 @@ def test_block_add_hook(self): def test_block_add_hook_baseexception(self): self.do_test("test_block_add_hook_baseexception") + def test_block_add_hook_valueerror(self): + self.do_test("test_block_add_hook_valueerror") + def test_marshal(self): import_helper.import_module("marshal") diff --git a/Misc/NEWS.d/next/Library/2026-07-03-02-11-00.gh-issue-152912.Mv3KpR.rst b/Misc/NEWS.d/next/Library/2026-07-03-02-11-00.gh-issue-152912.Mv3KpR.rst new file mode 100644 index 000000000000000..a4d7c4a9b0a2948 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-07-03-02-11-00.gh-issue-152912.Mv3KpR.rst @@ -0,0 +1 @@ +``sys.addaudithook()`` now correctly suppresses only :exc:`RuntimeError` instead of all :exc:`Exception` subclasses when an existing audit hook raises during hook registration. Patch by Yeongu Kim. diff --git a/Python/sysmodule.c b/Python/sysmodule.c index 9442472b53abbe1..1e6e914b066bc5c 100644 --- a/Python/sysmodule.c +++ b/Python/sysmodule.c @@ -527,8 +527,8 @@ sys_addaudithook_impl(PyObject *module, PyObject *hook) /* Invoke existing audit hooks to allow them an opportunity to abort. */ if (_PySys_Audit(tstate, "sys.addaudithook", NULL) < 0) { - if (_PyErr_ExceptionMatches(tstate, PyExc_Exception)) { - /* We do not report errors derived from Exception */ + if (_PyErr_ExceptionMatches(tstate, PyExc_RuntimeError)) { + /* We do not report errors derived from RuntimeError */ _PyErr_Clear(tstate); Py_RETURN_NONE; } From 0284ce81c9615432473abac709d5b261a49399ad Mon Sep 17 00:00:00 2001 From: Brett Cannon Date: Mon, 27 Jul 2026 13:18:53 -0700 Subject: [PATCH 5/6] gh-154675: reorganize some imports in importlib (#154657) --- Lib/importlib/abc.py | 5 ++++- Lib/importlib/machinery.py | 6 ++++-- Lib/importlib/util.py | 16 ++++++++++------ 3 files changed, 18 insertions(+), 9 deletions(-) diff --git a/Lib/importlib/abc.py b/Lib/importlib/abc.py index 9ca127ad9c7d0fd..ec68379e76f6f31 100644 --- a/Lib/importlib/abc.py +++ b/Lib/importlib/abc.py @@ -1,6 +1,7 @@ """Abstract base classes related to import.""" from . import _bootstrap_external from . import machinery + try: import _frozen_importlib except ImportError as exc: @@ -11,10 +12,12 @@ import _frozen_importlib_external except ImportError: _frozen_importlib_external = _bootstrap_external -from ._abc import Loader import abc +# Public API +from ._abc import Loader + __all__ = [ 'Loader', 'MetaPathFinder', 'PathEntryFinder', 'ResourceLoader', 'InspectLoader', 'ExecutionLoader', diff --git a/Lib/importlib/machinery.py b/Lib/importlib/machinery.py index 023f77d750fd2bc..b22e538862299ff 100644 --- a/Lib/importlib/machinery.py +++ b/Lib/importlib/machinery.py @@ -1,5 +1,9 @@ """The machinery of importlib: finders, loaders, hooks, etc.""" +lazy import warnings + + +# Public API from ._bootstrap import ModuleSpec from ._bootstrap import BuiltinImporter from ._bootstrap import FrozenImporter @@ -33,8 +37,6 @@ def all_suffixes(): def __getattr__(name): - import warnings - if name == 'DEBUG_BYTECODE_SUFFIXES': warnings.warn('importlib.machinery.DEBUG_BYTECODE_SUFFIXES is ' 'deprecated; use importlib.machinery.BYTECODE_SUFFIXES ' diff --git a/Lib/importlib/util.py b/Lib/importlib/util.py index 2b564e9b52e0cb2..dbce696dc8d169a 100644 --- a/Lib/importlib/util.py +++ b/Lib/importlib/util.py @@ -1,19 +1,23 @@ """Utility code for constructing importers, etc.""" + +from ._bootstrap import _resolve_name +from ._bootstrap import _find_spec + +import _imp +import sys +import types + + +# Public API from ._abc import Loader from ._bootstrap import module_from_spec -from ._bootstrap import _resolve_name from ._bootstrap import spec_from_loader -from ._bootstrap import _find_spec from ._bootstrap_external import MAGIC_NUMBER from ._bootstrap_external import cache_from_source from ._bootstrap_external import decode_source from ._bootstrap_external import source_from_cache from ._bootstrap_external import spec_from_file_location -import _imp -import sys -import types - def source_hash(source_bytes): "Return the hash of *source_bytes* as used in hash-based pyc files." From 1ec56076642091c446ec95afcf166bb4f185f0b7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=81ukasz=20Langa?= Date: Mon, 27 Jul 2026 13:45:40 -0700 Subject: [PATCH 6/6] gh-154467: Fix empty (Pdb) prompt when attaching to a process with pdb -p (#154469) _PdbServer inherits _cmdloop, which wraps cmdloop() in _maybe_use_pyrepl_as_stdin(). That context manager blanks self.prompt to '' so that a local pyrepl draws the prompt itself. The remote server, however, never reads from a local pyrepl -- it transmits self.prompt to the client over the socket -- so the blanking made it send an empty prompt whenever the target process had a pyrepl-capable terminal (pyrepl_input is set). Override _maybe_use_pyrepl_as_stdin() in _PdbServer to a no-op, keeping the real prompt. Add an integration test that attaches to a target running under a pty, so PyREPL is genuinely enabled inside it, and asserts the transmitted prompt is "(Pdb) ". Co-authored-by: Claude Opus 4.8 (1M context) --- Lib/pdb.py | 9 +++ Lib/test/test_remote_pdb.py | 71 +++++++++++++++++++ ...-07-22-12-59-26.gh-issue-154467.AA9w1D.rst | 3 + 3 files changed, 83 insertions(+) create mode 100644 Misc/NEWS.d/next/Library/2026-07-22-12-59-26.gh-issue-154467.AA9w1D.rst diff --git a/Lib/pdb.py b/Lib/pdb.py index 01451f0229cacb2..458eb8352652366 100644 --- a/Lib/pdb.py +++ b/Lib/pdb.py @@ -3158,6 +3158,15 @@ def postloop(self): if self.quitting: self.detach() + @contextmanager + def _maybe_use_pyrepl_as_stdin(self): + # The server reads every command from the client over the socket, never + # from a local pyrepl. The base implementation swaps in pyrepl as stdin + # and blanks `self.prompt` to '' (pyrepl would draw the prompt itself), + # which here would transmit an empty prompt to the client whenever the + # target process happens to be pyrepl-capable. Keep the real prompt. + yield + def detach(self): # Detach the debugger and close the socket without raising BdbQuit self.quitting = False diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py index d26d63faa61ddb6..5b23a194098b01c 100644 --- a/Lib/test/test_remote_pdb.py +++ b/Lib/test/test_remote_pdb.py @@ -8,6 +8,7 @@ import subprocess import sys import textwrap +import threading import unittest import unittest.mock from contextlib import closing, contextmanager, redirect_stdout, redirect_stderr, ExitStack @@ -18,6 +19,11 @@ import pdb from pdb import _PdbServer, _PdbClient +try: + import pty +except ImportError: + pty = None + if not sys.is_remote_debug_enabled(): raise unittest.SkipTest('remote debugging is disabled') @@ -1090,6 +1096,45 @@ def _connect_and_get_client_file(self): return process, client_file + def _connect_and_get_client_file_via_pty(self): + """Like _connect_and_get_client_file, but run the target under a pty. + + With a real terminal on stdin/stdout, PyREPL is available *inside the + target process*, which is the condition that exercises pdb's PyREPL + input handling in the remote server. + """ + controller, worker = pty.openpty() + self.addCleanup(os.close, controller) + env = dict(os.environ, TERM="xterm-256color") + env.pop("PYTHON_BASIC_REPL", None) # don't opt out of PyREPL + process = subprocess.Popen( + [sys.executable, self.script_path], + stdin=worker, stdout=worker, stderr=worker, env=env, close_fds=True, + ) + os.close(worker) # only the child keeps the worker end open + + # Continuously drain the terminal so the child never blocks on a write. + drainer = threading.Thread( + target=self._drain_until_eof, args=(controller,), daemon=True + ) + drainer.start() + self.addCleanup(drainer.join, SHORT_TIMEOUT) + + client_sock, _ = self.server_sock.accept() + client_file = client_sock.makefile('rwb') + self.addCleanup(client_file.close) + self.addCleanup(client_sock.close) + + return process, client_file + + @staticmethod + def _drain_until_eof(fd): + try: + while os.read(fd, 1024): + pass + except OSError: + pass # controller closed, or the pty went away with the child + def _read_until_prompt(self, client_file): """Helper to read messages until a prompt is received.""" messages = [] @@ -1292,6 +1337,32 @@ def test_handle_eof(self): self.assertEqual(process.returncode, 0) self.assertEqual(stderr, "") + @unittest.skipUnless(pty, "requires pty") + def test_prompt_with_interactive_terminal(self): + """The server must send "(Pdb) " even when the target owns a terminal. + + The remote server transmits its prompt string to the client, which + displays it. When the target process has an interactive terminal, + PyREPL is enabled inside it; the base pdb machinery then blanks + ``self.prompt`` (a local PyREPL would draw the prompt itself). The + remote server has no local PyREPL -- it reads input from the socket -- + so it must keep the real prompt rather than transmit an empty one. + Regression test for gh-154467, where attaching with ``pdb -p`` to a + process running at an interactive prompt showed a blank prompt. + """ + self._create_script() + process, client_file = self._connect_and_get_client_file_via_pty() + + with kill_on_error(process): + messages = self._read_until_prompt(client_file) + # The message that ended the read is the prompt request. + self.assertEqual(messages[-1], {"prompt": "(Pdb) ", "state": "pdb"}) + + # Let the target run to completion so nothing is left attached. + self._send_command(client_file, "c") + process.wait(timeout=SHORT_TIMEOUT) + self.assertEqual(process.returncode, 0) + def test_protocol_version(self): """Test that incompatible protocol versions are properly detected.""" # Create a script using an incompatible protocol version diff --git a/Misc/NEWS.d/next/Library/2026-07-22-12-59-26.gh-issue-154467.AA9w1D.rst b/Misc/NEWS.d/next/Library/2026-07-22-12-59-26.gh-issue-154467.AA9w1D.rst new file mode 100644 index 000000000000000..9d7ac18a5e9db12 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-07-22-12-59-26.gh-issue-154467.AA9w1D.rst @@ -0,0 +1,3 @@ +Fixed :mod:`pdb` remote attaching (``python -m pdb -p PID``) sending an +empty prompt to the client instead of ``(Pdb)`` when the target process has +an interactive terminal.