[3.12] gh-109413: regrtest: add WorkerRunTests class (GH-112588) (#112593)

gh-109413: regrtest: add WorkerRunTests class (GH-112588)
(cherry picked from commit f8ff80f635)

Co-authored-by: Victor Stinner <vstinner@python.org>

Lib/test/libregrtest/main.py

@@ -418,7 +418,6 @@ class Regrtest:
             python_cmd=self.python_cmd,
             randomize=self.randomize,
             random_seed=self.random_seed,
-            json_file=None,
         )
 
     def _run_tests(self, selected: TestTuple, tests: TestList | None) -> int:

Lib/test/libregrtest/run_workers.py

@@ -18,7 +18,7 @@ from test.support import os_helper, MS_WINDOWS
 from .logger import Logger
 from .result import TestResult, State
 from .results import TestResults
-from .runtests import RunTests, JsonFile, JsonFileType
+from .runtests import RunTests, WorkerRunTests, JsonFile, JsonFileType
 from .single import PROGRESS_MIN_TIME
 from .utils import (
     StrPath, TestName,
@@ -162,7 +162,7 @@ class WorkerThread(threading.Thread):
         self._stopped = True
         self._kill()
 
-    def _run_process(self, runtests: RunTests, output_fd: int,
+    def _run_process(self, runtests: WorkerRunTests, output_fd: int,
                      tmp_dir: StrPath | None = None) -> int | None:
         popen = create_worker_process(runtests, output_fd, tmp_dir)
         self._popen = popen
@@ -250,9 +250,7 @@ class WorkerThread(threading.Thread):
                 json_file = JsonFile(json_fd, JsonFileType.UNIX_FD)
         return (json_file, json_tmpfile)
 
-    def create_worker_runtests(self, test_name: TestName, json_file: JsonFile) -> RunTests:
-        """Create the worker RunTests."""
-
+    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)
@@ -265,12 +263,12 @@ class WorkerThread(threading.Thread):
         if self.runtests.output_on_failure:
             kwargs['verbose'] = True
             kwargs['output_on_failure'] = False
-        return self.runtests.copy(
+        return self.runtests.create_worker_runtests(
             tests=tests,
             json_file=json_file,
             **kwargs)
 
-    def run_tmp_files(self, worker_runtests: RunTests,
+    def run_tmp_files(self, worker_runtests: WorkerRunTests,
                       stdout_fd: int) -> tuple[int | None, list[StrPath]]:
         # gh-93353: Check for leaked temporary files in the parent process,
         # since the deletion of temporary files can happen late during

Lib/test/libregrtest/runtests.py

@@ -91,13 +91,17 @@ class RunTests:
     python_cmd: tuple[str, ...] | None
     randomize: bool
     random_seed: int | str
-    json_file: JsonFile | None
 
-    def copy(self, **override):
+    def copy(self, **override) -> 'RunTests':
         state = dataclasses.asdict(self)
         state.update(override)
         return RunTests(**state)
 
+    def create_worker_runtests(self, **override):
+        state = dataclasses.asdict(self)
+        state.update(override)
+        return WorkerRunTests(**state)
+
     def get_match_tests(self, test_name) -> FilterTuple | None:
         if self.match_tests_dict is not None:
             return self.match_tests_dict.get(test_name, None)
@@ -118,13 +122,6 @@ class RunTests:
         else:
             yield from self.tests
 
-    def as_json(self) -> StrJSON:
-        return json.dumps(self, cls=_EncodeRunTests)
-
-    @staticmethod
-    def from_json(worker_json: StrJSON) -> 'RunTests':
-        return json.loads(worker_json, object_hook=_decode_runtests)
-
     def json_file_use_stdout(self) -> bool:
         # Use STDOUT in two cases:
         #
@@ -139,9 +136,21 @@ class RunTests:
         )
 
 
+@dataclasses.dataclass(slots=True, frozen=True)
+class WorkerRunTests(RunTests):
+    json_file: JsonFile
+
+    def as_json(self) -> StrJSON:
+        return json.dumps(self, cls=_EncodeRunTests)
+
+    @staticmethod
+    def from_json(worker_json: StrJSON) -> 'WorkerRunTests':
+        return json.loads(worker_json, object_hook=_decode_runtests)
+
+
 class _EncodeRunTests(json.JSONEncoder):
     def default(self, o: Any) -> dict[str, Any]:
-        if isinstance(o, RunTests):
+        if isinstance(o, WorkerRunTests):
             result = dataclasses.asdict(o)
             result["__runtests__"] = True
             return result
@@ -156,6 +165,6 @@ def _decode_runtests(data: dict[str, Any]) -> RunTests | dict[str, Any]:
             data['hunt_refleak'] = HuntRefleak(**data['hunt_refleak'])
         if data['json_file']:
             data['json_file'] = JsonFile(**data['json_file'])
-        return RunTests(**data)
+        return WorkerRunTests(**data)
     else:
         return data

Lib/test/libregrtest/worker.py

@@ -7,7 +7,7 @@ from test import support
 from test.support import os_helper
 
 from .setup import setup_process, setup_test_dir
-from .runtests import RunTests, JsonFile, JsonFileType
+from .runtests import WorkerRunTests, JsonFile, JsonFileType
 from .single import run_single_test
 from .utils import (
     StrPath, StrJSON, TestFilter,
@@ -17,7 +17,7 @@ from .utils import (
 USE_PROCESS_GROUP = (hasattr(os, "setsid") and hasattr(os, "killpg"))
 
 
-def create_worker_process(runtests: RunTests, output_fd: int,
+def create_worker_process(runtests: WorkerRunTests, output_fd: int,
                           tmp_dir: StrPath | None = None) -> subprocess.Popen:
     python_cmd = runtests.python_cmd
     worker_json = runtests.as_json()
@@ -71,7 +71,7 @@ def create_worker_process(runtests: RunTests, output_fd: int,
 
 
 def worker_process(worker_json: StrJSON) -> NoReturn:
-    runtests = RunTests.from_json(worker_json)
+    runtests = WorkerRunTests.from_json(worker_json)
     test_name = runtests.tests[0]
     match_tests: TestFilter = runtests.match_tests
     json_file: JsonFile = runtests.json_file