gh-109162: Refactor Regrtest.main()

Lib/test/libregrtest/cmdline.py

@@ -448,4 +448,13 @@ def _parse_args(args, **kwargs):
         # --forever implies --failfast
         ns.failfast = True
 
+    if ns.huntrleaks:
+        warmup, repetitions, _ = ns.huntrleaks
+        if warmup < 1 or repetitions < 1:
+            msg = ("Invalid values for the --huntrleaks/-R parameters. The "
+                   "number of warmups and repetitions must be at least 1 "
+                   "each (1:1).")
+            print(msg, file=sys.stderr, flush=True)
+            sys.exit(2)
+
     return ns

Lib/test/libregrtest/main.py

@@ -9,10 +9,10 @@
 import tempfile
 import time
 import unittest
-from test.libregrtest.cmdline import _parse_args
+from test.libregrtest.cmdline import _parse_args, Namespace
 from test.libregrtest.runtest import (
     findtests, split_test_packages, runtest, abs_module_name,
-    PROGRESS_MIN_TIME, State, MatchTestsDict, RunTests)
+    PROGRESS_MIN_TIME, State, FilterDict, RunTests, TestResult, TestList)
 from test.libregrtest.setup import setup_tests
 from test.libregrtest.pgo import setup_pgo_tests
 from test.libregrtest.utils import (strip_py_suffix, count, format_duration,
@@ -58,24 +58,24 @@ class Regrtest:
     directly to set the values that would normally be set by flags
     on the command line.
     """
-    def __init__(self):
+    def __init__(self, ns: Namespace):
         # Namespace of command line options
-        self.ns = None
+        self.ns: Namespace = ns
 
         # tests
         self.tests = []
         self.selected = []
         self.all_runtests: list[RunTests] = []
 
         # test results
-        self.good: list[str] = []
-        self.bad: list[str] = []
-        self.rerun_bad: list[str] = []
-        self.skipped: list[str] = []
-        self.resource_denied: list[str] = []
-        self.environment_changed: list[str] = []
-        self.run_no_tests: list[str] = []
-        self.rerun: list[str] = []
+        self.good: TestList = []
+        self.bad: TestList = []
+        self.rerun_bad: TestList = []
+        self.skipped: TestList = []
+        self.resource_denied: TestList = []
+        self.environment_changed: TestList = []
+        self.run_no_tests: TestList = []
+        self.rerun: TestList = []
 
         self.need_rerun: list[TestResult] = []
         self.first_state: str | None = None
@@ -184,29 +184,7 @@ def display_progress(self, test_index, text):
             line = f"{line}/{fails}"
         self.log(f"[{line}] {text}")
 
-    def parse_args(self, kwargs):
-        ns = _parse_args(sys.argv[1:], **kwargs)
-
-        if ns.xmlpath:
-            support.junit_xml_list = self.testsuite_xml = []
-
-        strip_py_suffix(ns.args)
-
-        if ns.huntrleaks:
-            warmup, repetitions, _ = ns.huntrleaks
-            if warmup < 1 or repetitions < 1:
-                msg = ("Invalid values for the --huntrleaks/-R parameters. The "
-                       "number of warmups and repetitions must be at least 1 "
-                       "each (1:1).")
-                print(msg, file=sys.stderr, flush=True)
-                sys.exit(2)
-
-        if ns.tempdir:
-            ns.tempdir = os.path.expanduser(ns.tempdir)
-
-        self.ns = ns
-
-    def find_tests(self, tests):
+    def find_tests(self):
         ns = self.ns
         single = ns.single
         fromfile = ns.fromfile
@@ -216,8 +194,6 @@ def find_tests(self, tests):
         starting_test = ns.start
         randomize = ns.randomize
 
-        self.tests = tests
-
         if single:
             self.next_single_filename = os.path.join(self.tmp_dir, 'pynexttest')
             try:
@@ -321,7 +297,7 @@ def list_cases(self):
             print(count(len(skipped), "test"), "skipped:", file=stderr)
             printlist(skipped, file=stderr)
 
-    def get_rerun_match(self, rerun_list) -> MatchTestsDict:
+    def get_rerun_match(self, rerun_list) -> FilterDict:
         rerun_match_tests = {}
         for result in rerun_list:
             match_tests = result.get_rerun_match_tests()
@@ -352,7 +328,7 @@ def _rerun_failed_tests(self, need_rerun):
 
         # Re-run failed tests
         self.log(f"Re-running {len(tests)} failed tests in verbose mode in subprocesses")
-        runtests = RunTests(tests, match_tests=match_tests, rerun=True)
+        runtests = RunTests(tuple(tests), match_tests=match_tests, rerun=True)
         self.all_runtests.append(runtests)
         self._run_tests_mp(runtests)
 
@@ -624,7 +600,7 @@ def run_tests(self):
 
         tests = self.selected
         self.set_tests(tests)
-        runtests = RunTests(tests, forever=self.ns.forever)
+        runtests = RunTests(tuple(tests), forever=self.ns.forever)
         self.all_runtests.append(runtests)
         if self.ns.use_mp:
             self._run_tests_mp(runtests)
@@ -737,8 +713,12 @@ def fix_umask(self):
                 os.umask(old_mask)
 
     def set_temp_dir(self):
-        if self.ns.tempdir:
-            self.tmp_dir = self.ns.tempdir
+        ns = self.ns
+        if ns.tempdir:
+            ns.tempdir = os.path.expanduser(ns.tempdir)
+
+        if ns.tempdir:
+            self.tmp_dir = ns.tempdir
 
         if not self.tmp_dir:
             # When tests are run from the Python build directory, it is best practice
@@ -795,14 +775,20 @@ def cleanup(self):
                 print("Remove file: %s" % name)
                 os_helper.unlink(name)
 
-    def main(self, tests=None, **kwargs):
-        self.parse_args(kwargs)
+    def main(self, tests: TestList | None = None):
+        ns = self.ns
+        self.tests = tests
+
+        if ns.xmlpath:
+            support.junit_xml_list = self.testsuite_xml = []
+
+        strip_py_suffix(ns.args)
 
         self.set_temp_dir()
 
         self.fix_umask()
 
-        if self.ns.cleanup:
+        if ns.cleanup:
             self.cleanup()
             sys.exit(0)
 
@@ -817,9 +803,9 @@ def main(self, tests=None, **kwargs):
                 # When using multiprocessing, worker processes will use test_cwd
                 # as their parent temporary directory. So when the main process
                 # exit, it removes also subdirectories of worker processes.
-                self.ns.tempdir = test_cwd
+                ns.tempdir = test_cwd
 
-                self._main(tests, kwargs)
+                self._main()
         except SystemExit as exc:
             # bpo-38203: Python can hang at exit in Py_Finalize(), especially
             # on threading._shutdown() call: put a timeout
@@ -862,7 +848,7 @@ def action_run_tests(self):
         self.display_summary()
         self.finalize()
 
-    def _main(self, tests, kwargs):
+    def _main(self):
         if self.is_worker():
             from test.libregrtest.runtest_mp import run_tests_worker
             run_tests_worker(self.ns.worker_args)
@@ -872,7 +858,7 @@ def _main(self, tests, kwargs):
             input("Press any key to continue...")
 
         setup_tests(self.ns)
-        self.find_tests(tests)
+        self.find_tests()
 
         exitcode = 0
         if self.ns.list_tests:
@@ -888,4 +874,5 @@ def _main(self, tests, kwargs):
 
 def main(tests=None, **kwargs):
     """Run the Python suite."""
-    Regrtest().main(tests=tests, **kwargs)
+    ns = _parse_args(sys.argv[1:], **kwargs)
+    Regrtest(ns).main(tests=tests)

Lib/test/libregrtest/runtest.py

@@ -19,8 +19,13 @@
 from test.libregrtest.utils import clear_caches, format_duration, print_warning
 
 
-MatchTests = list[str]
-MatchTestsDict = dict[str, MatchTests]
+TestTuple = list[str]
+TestList = list[str]
+
+# --match and --ignore options: list of patterns
+# ('*' joker character can be used)
+FilterTuple = tuple[str, ...]
+FilterDict = dict[str, FilterTuple]
 
 
 # Avoid enum.Enum to reduce the number of imports when tests are run
@@ -174,7 +179,7 @@ def must_stop(self, fail_fast: bool, fail_env_changed: bool) -> bool:
             return True
         return False
 
-    def get_rerun_match_tests(self):
+    def get_rerun_match_tests(self) -> FilterTuple | None:
         match_tests = []
 
         errors = self.errors or []
@@ -195,29 +200,30 @@ def get_rerun_match_tests(self):
                     return None
                 match_tests.append(match_name)
 
-        return match_tests
+        if not match_tests:
+            return None
+        return tuple(match_tests)
 
 
 @dataclasses.dataclass(slots=True, frozen=True)
 class RunTests:
-    tests: list[str]
-    match_tests: MatchTestsDict | None = None
+    tests: TestTuple
+    match_tests: FilterDict | None = None
     rerun: bool = False
     forever: bool = False
 
-    def get_match_tests(self, test_name) -> MatchTests | None:
+    def get_match_tests(self, test_name) -> FilterTuple | None:
         if self.match_tests is not None:
             return self.match_tests.get(test_name, None)
         else:
             return None
 
     def iter_tests(self):
-        tests = tuple(self.tests)
         if self.forever:
             while True:
-                yield from tests
+                yield from self.tests
         else:
-            yield from tests
+            yield from self.tests
 
 
 # Minimum duration of a test to display its duration or to mention that

Lib/test/libregrtest/runtest_mp.py

@@ -20,7 +20,7 @@
 from test.libregrtest.main import Regrtest
 from test.libregrtest.runtest import (
     runtest, TestResult, State, PROGRESS_MIN_TIME,
-    MatchTests, RunTests)
+    FilterTuple, RunTests)
 from test.libregrtest.setup import setup_tests
 from test.libregrtest.utils import format_duration, print_warning
 
@@ -49,7 +49,7 @@ class WorkerJob:
     test_name: str
     namespace: Namespace
     rerun: bool = False
-    match_tests: MatchTests | None = None
+    match_tests: FilterTuple | None = None
 
 
 class _EncodeWorkerJob(json.JSONEncoder):