From cb811728fbcd582314bb6673fec5016013d122c0 Mon Sep 17 00:00:00 2001 From: Aniket Singh Yadav Date: Fri, 19 Jun 2026 05:25:46 +0000 Subject: [PATCH 1/2] Add --single-process-per-case option to libregrtest --- Lib/test/libregrtest/cmdline.py | 6 ++ Lib/test/libregrtest/findtests.py | 40 +++++++++--- Lib/test/libregrtest/main.py | 25 +++++++- Lib/test/libregrtest/run_workers.py | 99 ++++++++++++++++++++++++++--- Lib/test/libregrtest/runtests.py | 16 +++++ 5 files changed, 168 insertions(+), 18 deletions(-) diff --git a/Lib/test/libregrtest/cmdline.py b/Lib/test/libregrtest/cmdline.py index 64c035307e66542..4ee4c207307fd52 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', diff --git a/Lib/test/libregrtest/findtests.py b/Lib/test/libregrtest/findtests.py index 6c0e50846a466bb..9794cf7309e6fb5 100644 --- a/Lib/test/libregrtest/findtests.py +++ b/Lib/test/libregrtest/findtests.py @@ -94,19 +94,43 @@ def list_cases(tests: TestTuple, *, 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) - skipped = [] +def collect_cases(tests: TestTuple, *, + match_tests: TestFilter | None = None, + test_dir: StrPath | None = None + ) -> tuple[dict[TestName, list[str]], list[TestName]]: + 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 + 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.loader._FailedTest): + continue + if isinstance(test, unittest.TestSuite): + _collect_cases(test, out) + 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..0e31859cd9dfaf8 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 @@ -521,6 +522,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: @@ -529,6 +532,10 @@ def _run_tests(self, selected: TestTuple, tests: TestList | None) -> int: "less than 3 warmup repetitions can give false positives!") print(msg, file=sys.stdout, flush=True) + if self.want_single_process_per_case and self.num_workers == 0: + # Each test case must run in its own subprocess + self.num_workers = 1 + if self.num_workers < 0: # Use all CPUs + 2 extra worker processes for tests # that like to sleep @@ -546,6 +553,22 @@ 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) + case_groups = tuple( + (module_name, tuple(cases)) + for module_name, cases in cases_by_module.items() + ) + if not case_groups: + self.log("No test cases found") + return 0 + 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 befdac7ee77f107..4c845842cd1fc08 100644 --- a/Lib/test/libregrtest/run_workers.py +++ b/Lib/test/libregrtest/run_workers.py @@ -70,6 +70,26 @@ def stop(self): with self.lock: self.tests_iter = None +class GroupedMultiprocessIterator: + """Provide test groups safely across multiple worker threads.""" + + def __init__(self, groups_iter): + self.lock = threading.Lock() + self.groups_iter = groups_iter + + def next_group(self): + with self.lock: + if self.groups_iter is None: + return None + try: + return next(self.groups_iter) + except StopIteration: + return None + + def stop(self): + with self.lock: + self.groups_iter = None + @dataclasses.dataclass(slots=True, frozen=True) class MultiprocessResult: @@ -119,6 +139,7 @@ def __init__(self, worker_id: int, runner: "RunWorkers") -> None: self._popen: subprocess.Popen[str] | None = None self._killed = False self._stopped = False + self.current_module: TestName = "" def __repr__(self) -> str: info = [f'WorkerThread #{self.worker_id}'] @@ -267,15 +288,18 @@ def create_json_file(self, stack: contextlib.ExitStack) -> tuple[JsonFile, TextI 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) + kwargs: dict[str, Any] = {} + + if self.runtests.single_process_per_case: + tests = (self.current_module,) + 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 @@ -388,6 +412,13 @@ def _runtest(self, test_name: TestName) -> MultiprocessResult: return MultiprocessResult(result, stdout) def run(self) -> None: + if self.runtests.single_process_per_case: + self._run_grouped() + else: + self._run_flat() + + def _run_flat(self) -> None: + """Original behavior: one test name (module) per iteration.""" fail_fast = self.runtests.fail_fast fail_env_changed = self.runtests.fail_env_changed try: @@ -417,6 +448,52 @@ def run(self) -> None: finally: self.output.put(WorkerThreadExited()) + def _run_grouped(self) -> None: + """Execute all tests in a group on the same thread before moving on.""" + fail_fast = self.runtests.fail_fast + fail_env_changed = self.runtests.fail_env_changed + try: + while not self._stopped: + group = self.pending.next_group() + if group is None: + break + + module_name, case_ids = group + must_stop = False + for test_name in case_ids: + if self._stopped: + break + self.current_module = module_name + 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 = dataclasses.replace( + mp_result, + result=dataclasses.replace( + mp_result.result, + test_name=test_name, + duration=time.monotonic() - self.start_time)) + + self.output.put((False, mp_result)) + if mp_result.result.must_stop(fail_fast, fail_env_changed): + must_stop = True + break + + if must_stop: + break + except ExitThread: + pass + except BaseException: + self.output.put((True, traceback.format_exc())) + finally: + self.output.put(WorkerThreadExited()) + def _wait_completed(self) -> None: popen = self._popen # only needed for mypy: @@ -486,8 +563,12 @@ 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) + if runtests.single_process_per_case: + groups_iter = runtests.iter_case_groups() + self.pending = GroupedMultiprocessIterator(groups_iter) + else: + tests_iter = runtests.iter_tests() + self.pending = MultiprocessIterator(tests_iter) 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: # From dd50eebf605e316a2e1b43aa7f648edde35c0bf7 Mon Sep 17 00:00:00 2001 From: Aniket Singh Yadav Date: Sat, 11 Jul 2026 06:14:19 +0000 Subject: [PATCH 2/2] added tests and improved --single-process-per-case to run each test case in its own worker process --- Lib/test/libregrtest/cmdline.py | 11 ++ Lib/test/libregrtest/findtests.py | 6 +- Lib/test/libregrtest/main.py | 8 +- Lib/test/libregrtest/run_workers.py | 24 ++-- Lib/test/test_regrtest.py | 205 +++++++++++++++++++++++++++- 5 files changed, 240 insertions(+), 14 deletions(-) diff --git a/Lib/test/libregrtest/cmdline.py b/Lib/test/libregrtest/cmdline.py index 4ee4c207307fd52..720b073e460127f 100644 --- a/Lib/test/libregrtest/cmdline.py +++ b/Lib/test/libregrtest/cmdline.py @@ -498,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 9794cf7309e6fb5..ee584ed52d32d1d 100644 --- a/Lib/test/libregrtest/findtests.py +++ b/Lib/test/libregrtest/findtests.py @@ -110,6 +110,10 @@ 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: @@ -127,7 +131,7 @@ def collect_cases(tests: TestTuple, *, def _collect_cases(suite: unittest.TestSuite, out: list[str]) -> None: for test in suite: - if isinstance(test, unittest.loader._FailedTest): + if isinstance(test, unittest.loader._FailedTest): # type: ignore[attr-defined] continue if isinstance(test, unittest.TestSuite): _collect_cases(test, out) diff --git a/Lib/test/libregrtest/main.py b/Lib/test/libregrtest/main.py index 0e31859cd9dfaf8..6bd202c42ceb0e0 100644 --- a/Lib/test/libregrtest/main.py +++ b/Lib/test/libregrtest/main.py @@ -100,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 @@ -532,10 +536,6 @@ def _run_tests(self, selected: TestTuple, tests: TestList | None) -> int: "less than 3 warmup repetitions can give false positives!") print(msg, file=sys.stdout, flush=True) - if self.want_single_process_per_case and self.num_workers == 0: - # Each test case must run in its own subprocess - self.num_workers = 1 - if self.num_workers < 0: # Use all CPUs + 2 extra worker processes for tests # that like to sleep diff --git a/Lib/test/libregrtest/run_workers.py b/Lib/test/libregrtest/run_workers.py index 4c845842cd1fc08..d4ed0cc65260c3c 100644 --- a/Lib/test/libregrtest/run_workers.py +++ b/Lib/test/libregrtest/run_workers.py @@ -139,7 +139,6 @@ def __init__(self, worker_id: int, runner: "RunWorkers") -> None: self._popen: subprocess.Popen[str] | None = None self._killed = False self._stopped = False - self.current_module: TestName = "" def __repr__(self) -> str: info = [f'WorkerThread #{self.worker_id}'] @@ -287,11 +286,16 @@ 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: + def create_worker_runtests(self, test_name: TestName, + json_file: JsonFile, + module_name: TestName | None = None, + ) -> WorkerRunTests: kwargs: dict[str, Any] = {} - if self.runtests.single_process_per_case: - tests = (self.current_module,) + if module_name is not None: + # single_process_per_case mode: run a single test case + # (test_name is a case ID) inside its module. + tests = (module_name,) kwargs['match_tests'] = [(test_name, True)] else: tests = (test_name,) @@ -377,11 +381,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, @@ -419,6 +425,7 @@ def run(self) -> None: def _run_flat(self) -> None: """Original behavior: one test name (module) per iteration.""" + assert isinstance(self.pending, MultiprocessIterator) fail_fast = self.runtests.fail_fast fail_env_changed = self.runtests.fail_env_changed try: @@ -450,6 +457,7 @@ def _run_flat(self) -> None: def _run_grouped(self) -> None: """Execute all tests in a group on the same thread before moving on.""" + assert isinstance(self.pending, GroupedMultiprocessIterator) fail_fast = self.runtests.fail_fast fail_env_changed = self.runtests.fail_env_changed try: @@ -463,11 +471,10 @@ def _run_grouped(self) -> None: for test_name in case_ids: if self._stopped: break - self.current_module = module_name self.start_time = time.monotonic() self.test_name = test_name try: - mp_result = self._runtest(test_name) + mp_result = self._runtest(test_name, module_name) except WorkerError as exc: mp_result = exc.mp_result finally: @@ -563,6 +570,7 @@ def __init__(self, num_workers: int, runtests: RunTests, self.live_worker_count = 0 self.output: queue.Queue[QueueContent] = queue.Queue() + self.pending: MultiprocessIterator | GroupedMultiprocessIterator if runtests.single_process_per_case: groups_iter = runtests.iter_case_groups() self.pending = GroupedMultiprocessIterator(groups_iter) diff --git a/Lib/test/test_regrtest.py b/Lib/test/test_regrtest.py index 6d30d267cd5ad29..5a5edfd25076c34 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 GroupedMultiprocessIterator 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 @@ -575,6 +579,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) @@ -2333,6 +2361,74 @@ 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_verbose3(self): code = textwrap.dedent(r""" import unittest @@ -2443,6 +2539,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 GroupedMultiprocessIteratorTestCase(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 = GroupedMultiprocessIterator(iter(groups)) + seen = [] + while (g := it.next_group()) is not None: + seen.append(g) + self.assertEqual(seen, groups) + + def test_exhausted_returns_none(self): + it = GroupedMultiprocessIterator(iter([])) + self.assertIsNone(it.next_group()) + + def test_stop_halts_iteration(self): + groups = [("mod_a", ("mod_a.A.t1",)), ("mod_b", ("mod_b.B.t1",))] + it = GroupedMultiprocessIterator(iter(groups)) + it.next_group() + it.stop() + self.assertIsNone(it.next_group()) + + 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 = GroupedMultiprocessIterator(iter(groups)) + results = [] + results_lock = threading.Lock() + + def worker(): + while (g := it.next_group()) 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),