https://github.com/python/cpython/commit/296f016e93002e91fdbca5732f9eb4a80bc1af39
commit: 296f016e93002e91fdbca5732f9eb4a80bc1af39
branch: main
author: Aniket <[email protected]>
committer: serhiy-storchaka <[email protected]>
date: 2026-08-13T12:14:51+03:00
summary:
gh-109817: Add --single-process-per-case option to libregrtest (GH-151689)
files:
A Misc/NEWS.d/next/Tests/2026-08-12-20-52-16.gh-issue-109817.GZHUNg.rst
M Lib/test/libregrtest/cmdline.py
M Lib/test/libregrtest/findtests.py
M Lib/test/libregrtest/main.py
M Lib/test/libregrtest/run_workers.py
M Lib/test/libregrtest/runtests.py
M Lib/test/test_regrtest.py
diff --git a/Lib/test/libregrtest/cmdline.py b/Lib/test/libregrtest/cmdline.py
index 64c035307e66542..720b073e460127f 100644
--- a/Lib/test/libregrtest/cmdline.py
+++ b/Lib/test/libregrtest/cmdline.py
@@ -194,6 +194,7 @@ def __init__(self, **kwargs) -> None:
self._add_python_opts = True
self.xmlpath = None
self.single_process = False
+ self.single_process_per_case = False
super().__init__(**kwargs)
@@ -368,6 +369,11 @@ def _create_parser():
group.add_argument('--list-cases', action='store_true',
help='only write the name of test cases that will be
run, '
'don\'t execute them')
+ group.add_argument('--single-process-per-case', action='store_true',
+ help='run each test case in its own process. '
+ '(slow; for debugging order dependencies '
+ "and environment leaks). Test cases from "
+ 'the same module run sequentially.')
group.add_argument('-P', '--pgo', dest='pgo', action='store_true',
help='enable Profile Guided Optimization (PGO)
training')
group.add_argument('--pgo-extended', action='store_true',
@@ -492,6 +498,17 @@ def _parse_args(args, **kwargs):
if ns.single_process:
ns.use_mp = None
+ if ns.single_process_per_case:
+ if ns.rerun:
+ parser.error("--single-process-per-case and --rerun "
+ "options don't go together")
+ if ns.pgo:
+ parser.error("--single-process-per-case and --pgo "
+ "options don't go together")
+ if ns.single_process:
+ parser.error("--single-process-per-case and --single-process "
+ "options don't go together")
+
# When both --slow-ci and --fast-ci options are present,
# --slow-ci has the priority
if ns.slow_ci:
diff --git a/Lib/test/libregrtest/findtests.py
b/Lib/test/libregrtest/findtests.py
index 6c0e50846a466bb..e7692c5156812e4 100644
--- a/Lib/test/libregrtest/findtests.py
+++ b/Lib/test/libregrtest/findtests.py
@@ -93,20 +93,58 @@ def list_cases(tests: TestTuple, *,
match_tests: TestFilter | None = None,
test_dir: StrPath | None = None) -> None:
support.verbose = False
- set_match_tests(match_tests)
+ cases_by_module, skipped = collect_cases(tests, match_tests=match_tests,
+ test_dir=test_dir)
+ for cases in cases_by_module.values():
+ for case_id in cases:
+ print(case_id)
+ if skipped:
+ sys.stdout.flush()
+ stderr = sys.stderr
+ print(file=stderr)
+ print(count(len(skipped), "test"), "skipped:", file=stderr)
+ printlist(skipped, file=stderr)
+
+class _ModuleLoadFailed(Exception):
+ """The test module failed to load; its test cases are unknown."""
+
- skipped = []
+def collect_cases(tests: TestTuple, *,
+ match_tests: TestFilter | None = None,
+ test_dir: StrPath | None = None
+ ) -> tuple[dict[TestName, list[str]], list[TestName]]:
+ # Install the filter unconditionally: passing None clears any
+ # previously installed global filter, so collection is not
+ # affected by unrelated state in this process.
+ set_match_tests(match_tests)
+ result: dict[TestName, list[str]] = {}
+ skipped: list[TestName] = []
for test_name in tests:
module_name = abs_module_name(test_name, test_dir)
+ cases: list[str] = []
try:
suite = unittest.defaultTestLoader.loadTestsFromName(module_name)
- _list_cases(suite)
+ _collect_cases(suite, cases)
except unittest.SkipTest:
skipped.append(test_name)
+ continue
+ except _ModuleLoadFailed:
+ # The module failed to load. Run it as a whole, so that the
+ # error is reported as in the normal mode.
+ result[test_name] = [test_name]
+ continue
+ if cases:
+ result[test_name] = cases
+ return result, skipped
- if skipped:
- sys.stdout.flush()
- stderr = sys.stderr
- print(file=stderr)
- print(count(len(skipped), "test"), "skipped:", file=stderr)
- printlist(skipped, file=stderr)
+def _collect_cases(suite: unittest.TestSuite, out: list[str]) -> None:
+ for test in suite:
+ if isinstance(test, unittest.TestSuite):
+ _collect_cases(test, out)
+ elif isinstance(test, unittest.loader._FailedTest): # type:
ignore[attr-defined]
+ # The test module failed to load. Its test cases are
+ # unknown: let the caller run the whole module.
+ raise _ModuleLoadFailed
+ elif isinstance(test, unittest.TestCase):
+ if match_test(test):
+ out.append(test.id())
diff --git a/Lib/test/libregrtest/main.py b/Lib/test/libregrtest/main.py
index 8773e9df73263b7..7391056627cd068 100644
--- a/Lib/test/libregrtest/main.py
+++ b/Lib/test/libregrtest/main.py
@@ -12,7 +12,7 @@
from test.support import os_helper, MS_WINDOWS, flush_std_streams
from .cmdline import _parse_args, Namespace
-from .findtests import findtests, split_test_packages, list_cases
+from .findtests import findtests, split_test_packages, list_cases,
collect_cases
from .logger import Logger
from .pgo import setup_pgo_tests
from .result import TestResult
@@ -73,6 +73,7 @@ def __init__(self, ns: Namespace, _add_python_opts: bool =
False):
self.want_header: bool = ns.header
self.want_list_tests: bool = ns.list_tests
self.want_list_cases: bool = ns.list_cases
+ self.want_single_process_per_case: bool = ns.single_process_per_case
self.want_wait: bool = ns.wait
self.want_cleanup: bool = ns.cleanup
self.want_rerun: bool = ns.rerun
@@ -99,6 +100,10 @@ def __init__(self, ns: Namespace, _add_python_opts: bool =
False):
else:
num_workers = ns.use_mp # run in parallel
self.num_workers: int = num_workers
+ if ns.single_process_per_case and ns.use_mp is None:
+ # Each test case runs in its own worker subprocess;
+ # default to one worker when -j was not given.
+ self.num_workers = 1
self.worker_json: StrJSON | None = ns.worker_json
# Options to run tests
@@ -521,6 +526,8 @@ def create_run_tests(self, tests: TestTuple) -> RunTests:
randomize=self.randomize,
random_seed=self.random_seed,
parallel_threads=self.parallel_threads,
+ single_process_per_case=self.want_single_process_per_case,
+ case_groups=None,
)
def _run_tests(self, selected: TestTuple, tests: TestList | None) -> int:
@@ -546,6 +553,23 @@ def _run_tests(self, selected: TestTuple, tests: TestList
| None) -> int:
print("Using random seed:", self.random_seed)
runtests = self.create_run_tests(selected)
+ if self.want_single_process_per_case:
+ cases_by_module, _ = collect_cases(
+ selected,
+ match_tests=self.match_tests,
+ test_dir=self.test_dir)
+ groups = []
+ for module_name in selected:
+ cases = cases_by_module.get(module_name)
+ if cases:
+ groups.append((module_name, tuple(cases)))
+ else:
+ groups.append((module_name, (module_name,)))
+ case_groups = tuple(groups)
+ case_ids = tuple(
+ case_id for _, cases in case_groups for case_id in cases
+ )
+ runtests = runtests.copy(tests=case_ids, case_groups=case_groups)
self.first_runtests = runtests
self.logger.set_tests(runtests)
diff --git a/Lib/test/libregrtest/run_workers.py
b/Lib/test/libregrtest/run_workers.py
index 7e6c7fa4cc5507a..a146a439d9d3eca 100644
--- a/Lib/test/libregrtest/run_workers.py
+++ b/Lib/test/libregrtest/run_workers.py
@@ -70,7 +70,6 @@ def stop(self):
with self.lock:
self.tests_iter = None
-
@dataclasses.dataclass(slots=True, frozen=True)
class MultiprocessResult:
result: TestResult
@@ -269,16 +268,22 @@ def create_json_file(self, stack: contextlib.ExitStack)
-> tuple[JsonFile, TextI
json_file = JsonFile(json_fd, JsonFileType.UNIX_FD)
return (json_file, json_tmpfile)
- def create_worker_runtests(self, test_name: TestName, json_file: JsonFile)
-> WorkerRunTests:
- tests = (test_name,)
- if self.runtests.rerun:
- match_tests = self.runtests.get_match_tests(test_name)
+ def create_worker_runtests(self, test_name: TestName,
+ json_file: JsonFile,
+ module_name: TestName | None = None,
+ ) -> WorkerRunTests:
+ kwargs: dict[str, Any] = {}
+
+ if module_name is not None and test_name != module_name:
+ tests = (module_name,)
+ kwargs['match_tests'] = [(test_name, True)]
else:
- match_tests = None
+ tests = (test_name,)
+ if self.runtests.rerun:
+ match_tests = self.runtests.get_match_tests(test_name)
+ if match_tests:
+ kwargs['match_tests'] = [(test, True) for test in
match_tests]
- kwargs: dict[str, Any] = {}
- if match_tests:
- kwargs['match_tests'] = [(test, True) for test in match_tests]
if self.runtests.output_on_failure:
kwargs['verbose'] = True
kwargs['output_on_failure'] = False
@@ -356,11 +361,13 @@ def read_json(self, json_file: JsonFile, json_tmpfile:
TextIO | None,
return (result, stdout)
- def _runtest(self, test_name: TestName) -> MultiprocessResult:
+ def _runtest(self, test_name: TestName,
+ module_name: TestName | None = None) -> MultiprocessResult:
with contextlib.ExitStack() as stack:
stdout_file = self.create_stdout(stack)
json_file, json_tmpfile = self.create_json_file(stack)
- worker_runtests = self.create_worker_runtests(test_name, json_file)
+ worker_runtests = self.create_worker_runtests(
+ test_name, json_file, module_name=module_name)
retcode: str | int | None
retcode, tmp_files = self.run_tmp_files(worker_runtests,
@@ -393,26 +400,38 @@ def _runtest(self, test_name: TestName) ->
MultiprocessResult:
def run(self) -> None:
fail_fast = self.runtests.fail_fast
fail_env_changed = self.runtests.fail_env_changed
+ single_process_per_case = self.runtests.single_process_per_case
try:
- while not self._stopped:
+ stop = False
+ while not self._stopped and not stop:
try:
- test_name = next(self.pending)
+ module_name, case_ids = next(self.pending)
except StopIteration:
break
- self.start_time = time.monotonic()
- self.test_name = test_name
- try:
- mp_result = self._runtest(test_name)
- except WorkerError as exc:
- mp_result = exc.mp_result
- finally:
- self.test_name = _NOT_RUNNING
- mp_result.result.duration = time.monotonic() - self.start_time
- self.output.put((False, mp_result))
-
- if mp_result.result.must_stop(fail_fast, fail_env_changed):
- break
+ # All cases of a group run sequentially on this thread
+ for test_name in case_ids:
+ if self._stopped:
+ break
+ self.start_time = time.monotonic()
+ self.test_name = test_name
+ try:
+ mp_result = self._runtest(
+ test_name,
+ module_name if single_process_per_case else None)
+ except WorkerError as exc:
+ mp_result = exc.mp_result
+ finally:
+ self.test_name = _NOT_RUNNING
+ mp_result.result.duration = time.monotonic() -
self.start_time
+ if single_process_per_case:
+ # Report the test case, not the test module
+ mp_result.result.test_name = test_name
+ self.output.put((False, mp_result))
+
+ if mp_result.result.must_stop(fail_fast, fail_env_changed):
+ stop = True
+ break
except ExitThread:
pass
except BaseException:
@@ -489,8 +508,7 @@ def __init__(self, num_workers: int, runtests: RunTests,
self.live_worker_count = 0
self.output: queue.Queue[QueueContent] = queue.Queue()
- tests_iter = runtests.iter_tests()
- self.pending = MultiprocessIterator(tests_iter)
+ self.pending = MultiprocessIterator(runtests.iter_case_groups())
self.timeout = runtests.timeout
if self.timeout is not None:
# Rely on faulthandler to kill a worker process. This timouet is
diff --git a/Lib/test/libregrtest/runtests.py b/Lib/test/libregrtest/runtests.py
index 0a9edce1085be54..fbb04b5b705ecec 100644
--- a/Lib/test/libregrtest/runtests.py
+++ b/Lib/test/libregrtest/runtests.py
@@ -101,6 +101,8 @@ class RunTests:
randomize: bool
random_seed: int | str
parallel_threads: int | None
+ single_process_per_case: bool
+ case_groups: tuple[tuple[TestName, tuple[TestName, ...]], ...] | None
def copy(self, **override) -> 'RunTests':
state = dataclasses.asdict(self)
@@ -132,6 +134,20 @@ def iter_tests(self) -> Iterator[TestName]:
else:
yield from self.tests
+ def iter_case_groups(self) -> Iterator[tuple[TestName, tuple[TestName,
...]]]:
+ """
+ Yield (module_name, case_ids) pairs. All case_ids in a group
+ must run sequentially on the same worker thread.
+ """
+ if self.case_groups is None:
+ for name in self.iter_tests():
+ yield (name, (name,))
+ elif self.forever:
+ while True:
+ yield from self.case_groups
+ else:
+ yield from self.case_groups
+
def json_file_use_stdout(self) -> bool:
# Use STDOUT in two cases:
#
diff --git a/Lib/test/test_regrtest.py b/Lib/test/test_regrtest.py
index 6ba440053089161..f05008bf8bc2c01 100644
--- a/Lib/test/test_regrtest.py
+++ b/Lib/test/test_regrtest.py
@@ -20,11 +20,15 @@
import sys
import sysconfig
import tempfile
+import threading
import textwrap
import unittest
import unittest.mock
from xml.etree import ElementTree
+from test.libregrtest.findtests import collect_cases
+from test.libregrtest.filter import set_match_tests
+from test.libregrtest.run_workers import MultiprocessIterator
from test import support
from test.support import import_helper
from test.support import os_helper
@@ -32,7 +36,7 @@
from test.libregrtest import main
from test.libregrtest import setup
from test.libregrtest import utils
-from test.libregrtest.filter import get_match_tests, set_match_tests,
match_test
+from test.libregrtest.filter import get_match_tests, match_test
from test.libregrtest.result import TestStats
from test.libregrtest.utils import normalize_test_name
@@ -574,6 +578,30 @@ def test_single_process(self):
self.assertEqual(regrtest.num_workers, 0)
self.assertTrue(regrtest.single_process)
+ def test_single_process_per_case(self):
+ ns = self.parse_args(['--single-process-per-case'])
+ self.assertTrue(ns.single_process_per_case)
+
+ # No -j given: default to one worker
+ regrtest = self.create_regrtest(['--single-process-per-case'])
+ self.assertEqual(regrtest.num_workers, 1)
+
+ # Explicit -j2 is respected
+ regrtest = self.create_regrtest(['-j2', '--single-process-per-case'])
+ self.assertEqual(regrtest.num_workers, 2)
+
+ def test_single_process_per_case_conflicts(self):
+ # --single-process-per-case doesn't compose with options that
+ # re-run or restrict test selection after collection (gh-109817)
+ self.checkError(['--single-process-per-case', '--rerun'],
+ "don't go together")
+ self.checkError(['--single-process-per-case', '--single-process'],
+ "don't go together")
+ self.checkError(['--single-process-per-case', '--pgo'],
+ "don't go together")
+ self.checkError(['--single-process-per-case', '--fast-ci'],
+ "don't go together")
+
def test_pythoninfo(self):
ns = self.parse_args([])
self.assertFalse(ns.pythoninfo)
@@ -2332,6 +2360,94 @@ def test_crash(self):
self.assertIn(f"Exit code {exitcode} (SIGSEGV)", output)
self.check_line(output, "just before crash!", full=True, regex=False)
+ def test_single_process_per_case(self):
+ code = textwrap.dedent("""
+ import unittest
+
+ class Tests(unittest.TestCase):
+ def test_one(self):
+ pass
+ def test_two(self):
+ pass
+ """)
+ testname = self.create_test(code=code)
+
+ output = self.run_tests('--single-process-per-case', '-v', testname)
+ # Each case is reported individually (progress lines have a
+ # timestamp prefix, so match mid-line with assertIn)
+ self.assertIn(f'{testname}.Tests.test_one passed', output)
+ self.assertIn(f'{testname}.Tests.test_two passed', output)
+ self.check_line(output, 'All 2 tests OK.', regex=False)
+
+ def test_single_process_per_case_order_independence(self):
+ # A module where test_b only passes if it runs in isolation.
+ # This simulates the real order-dependency bugs this feature
+ # is meant to catch (gh-109817): running each case in its own
+ # process means test_b can never observe state left behind
+ # by test_a.
+ code = textwrap.dedent("""
+ import unittest
+
+ counter = {'n': 0}
+
+ class Tests(unittest.TestCase):
+ def test_a(self):
+ counter['n'] += 1
+ def test_b(self):
+ self.assertEqual(counter['n'], 0)
+ """)
+ testname = self.create_test(code=code)
+
+ output = self.run_tests('--single-process-per-case', '-v', testname)
+ self.assertIn(f'{testname}.Tests.test_a passed', output)
+ self.assertIn(f'{testname}.Tests.test_b passed', output)
+ self.check_line(output, 'All 2 tests OK.', regex=False)
+
+ def test_single_process_per_case_failure_isolated(self):
+ # One failing case must not abort or contaminate sibling
+ # cases in the same module.
+ code = textwrap.dedent("""
+ import unittest
+
+ class Tests(unittest.TestCase):
+ def test_pass(self):
+ pass
+ def test_fail(self):
+ self.fail("expected failure")
+ """)
+ testname = self.create_test(code=code)
+
+ output = self.run_tests('--single-process-per-case', testname,
+ exitcode=EXITCODE_BAD_TEST)
+ # The failing case is reported by its case ID...
+ self.assertIn(f'{testname}.Tests.test_fail failed', output)
+ self.check_line(output, '1 test failed:', regex=False)
+ self.assertIn(f' {testname}.Tests.test_fail', output)
+ # ...and the sibling case in the same module still ran and passed
+ self.assertIn(f'{testname}.Tests.test_pass passed', output)
+ self.check_line(output, '1 test OK.', regex=False)
+
+ def test_single_process_per_case_import_error(self):
+ testname = self.create_test(code='raise ImportError("boom")')
+
+ output = self.run_tests('--single-process-per-case', testname,
+ exitcode=EXITCODE_BAD_TEST)
+ self.check_executed_tests(output, [testname], failed=[testname],
+ stats=0)
+ self.assertNotIn('_FailedTest', output)
+
+ def test_single_process_per_case_skipped_module(self):
+ code = textwrap.dedent("""
+ import unittest
+ raise unittest.SkipTest("nope")
+ """)
+ testname = self.create_test(code=code)
+
+ output = self.run_tests('--single-process-per-case', testname)
+ self.check_executed_tests(output, [testname],
+ skipped=[testname], stats=0)
+
+
def test_verbose3(self):
code = textwrap.dedent(r"""
import unittest
@@ -2442,6 +2558,113 @@ def test_pythoninfo(self):
self.assertIn("Python build information", output)
+class FindTestsTestCase(BaseTestCase):
+ def test_collect_cases_groups_by_module(self):
+ code = textwrap.dedent("""
+ import unittest
+
+ class Tests(unittest.TestCase):
+ def test_one(self):
+ pass
+ def test_two(self):
+ pass
+ """)
+ testname = self.create_test(code=code)
+ sys.path.insert(0, self.tmptestdir)
+ self.addCleanup(sys.path.remove, self.tmptestdir)
+ self.addCleanup(sys.modules.pop, testname, None)
+ self.addCleanup(set_match_tests, None)
+
+ cases_by_module, skipped = collect_cases((testname,),
+ test_dir=self.tmptestdir)
+ self.assertIn(testname, cases_by_module)
+ self.assertEqual(len(cases_by_module[testname]), 2)
+ self.assertEqual(skipped, [])
+
+ def test_collect_cases_match_tests_filters(self):
+ code = textwrap.dedent("""
+ import unittest
+
+ class Tests(unittest.TestCase):
+ def test_keep(self):
+ pass
+ def test_drop(self):
+ pass
+ """)
+ testname = self.create_test(code=code)
+ sys.path.insert(0, self.tmptestdir)
+ self.addCleanup(sys.path.remove, self.tmptestdir)
+ self.addCleanup(sys.modules.pop, testname, None)
+ self.addCleanup(set_match_tests, None)
+
+ cases_by_module, _ = collect_cases(
+ (testname,),
+ match_tests=[('*test_keep*', True)],
+ test_dir=self.tmptestdir)
+ case_ids = cases_by_module.get(testname, [])
+ self.assertTrue(any('test_keep' in c for c in case_ids))
+ self.assertFalse(any('test_drop' in c for c in case_ids))
+
+ def test_collect_cases_skiptest(self):
+ code = textwrap.dedent("""
+ import unittest
+ raise unittest.SkipTest("module-level skip")
+ """)
+ testname = self.create_test(code=code)
+ sys.path.insert(0, self.tmptestdir)
+ self.addCleanup(sys.path.remove, self.tmptestdir)
+ self.addCleanup(sys.modules.pop, testname, None)
+ self.addCleanup(set_match_tests, None)
+
+ cases_by_module, skipped = collect_cases((testname,),
+ test_dir=self.tmptestdir)
+ self.assertEqual(cases_by_module, {})
+ self.assertIn(testname, skipped)
+
+
+class MultiprocessIteratorTestCase(unittest.TestCase):
+ def test_yields_all_groups_once(self):
+ groups = [("mod_a", ("mod_a.A.t1", "mod_a.A.t2")),
+ ("mod_b", ("mod_b.B.t1",))]
+ it = MultiprocessIterator(iter(groups))
+ seen = []
+ while (g := next(it, None)) is not None:
+ seen.append(g)
+ self.assertEqual(seen, groups)
+
+ def test_exhausted_returns_none(self):
+ it = MultiprocessIterator(iter([]))
+ self.assertIsNone(next(it, None))
+
+ def test_stop_halts_iteration(self):
+ groups = [("mod_a", ("mod_a.A.t1",)), ("mod_b", ("mod_b.B.t1",))]
+ it = MultiprocessIterator(iter(groups))
+ next(it, None)
+ it.stop()
+ self.assertIsNone(next(it, None))
+
+ def test_thread_safety_no_duplicate_or_lost_groups(self):
+ n = 200
+ groups = [(f"mod_{i}", (f"mod_{i}.T.t",)) for i in range(n)]
+ it = MultiprocessIterator(iter(groups))
+ results = []
+ results_lock = threading.Lock()
+
+ def worker():
+ while (g := next(it, None)) is not None:
+ with results_lock:
+ results.append(g)
+
+ threads = [threading.Thread(target=worker) for _ in range(8)]
+ for t in threads:
+ t.start()
+ for t in threads:
+ t.join()
+
+ self.assertEqual(sorted(results), sorted(groups))
+ self.assertEqual(len(results), n)
+
+
class TestUtils(unittest.TestCase):
def test_format_duration(self):
self.assertEqual(utils.format_duration(0),
diff --git
a/Misc/NEWS.d/next/Tests/2026-08-12-20-52-16.gh-issue-109817.GZHUNg.rst
b/Misc/NEWS.d/next/Tests/2026-08-12-20-52-16.gh-issue-109817.GZHUNg.rst
new file mode 100644
index 000000000000000..6064b331992d715
--- /dev/null
+++ b/Misc/NEWS.d/next/Tests/2026-08-12-20-52-16.gh-issue-109817.GZHUNg.rst
@@ -0,0 +1,4 @@
+Add the ``--single-process-per-case`` option to libregrtest to run
+every test case in a separate process. Test cases from the same
+test module are run sequentially. This helps to detect order
+dependencies and environment leaks between test cases.
_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3//lists/python-checkins.python.org
Member address: [email protected]
