gh-109817: Add --single-process-per-case option to libregrtest (GH-151689)

Author: Aniketsy

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:

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())

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)
 

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

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:
         #