Add JsonFile and JsonFileType classes

Author: vstinner

Lib/test/libregrtest/run_workers.py

@@ -17,7 +17,7 @@
 from .logger import Logger
 from .result import TestResult, State
 from .results import TestResults
-from .runtests import RunTests
+from .runtests import RunTests, JsonFile, JsonFileType
 from .single import PROGRESS_MIN_TIME
 from .utils import (
     StrPath, StrJSON, TestName, MS_WINDOWS, TMP_PREFIX,
@@ -240,11 +240,15 @@ def _runtest(self, test_name: TestName) -> MultiprocessResult:
         with (stdout_file, json_tmpfile):
             stdout_fd = stdout_file.fileno()
             if json_file_use_filename:
-                json_file = json_tmpfile.name
+                json_file = JsonFile(json_tmpfile.name, JsonFileType.FILENAME)
             else:
-                json_file = json_tmpfile.fileno()
+                json_fd = json_tmpfile.fileno()
                 if MS_WINDOWS:
-                    json_file = msvcrt.get_osfhandle(json_file)
+                    json_handle = msvcrt.get_osfhandle(json_fd)
+                    json_file = JsonFile(json_handle, JsonFileType.WINDOWS_HANDLE)
+                else:
+                    json_file = JsonFile(json_fd, JsonFileType.UNIX_FD)
+
 
             kwargs = {}
             if match_tests:

Lib/test/libregrtest/runtests.py

@@ -1,3 +1,4 @@
+import contextlib
 import dataclasses
 import json
 from typing import Any
@@ -8,8 +9,50 @@
     StrPath, StrJSON, TestTuple, FilterTuple, FilterDict)
 
 
-# See RunTests.json_file_use_filename()
-JsonFileType = int | StrPath
+class JsonFileType:
+    UNIX_FD = "UNIX_FD"
+    WINDOWS_HANDLE = "WINDOWS_HANDLE"
+    FILENAME = "FILENAME"
+
+
+@dataclasses.dataclass(slots=True, frozen=True)
+class JsonFile:
+    # See RunTests.json_file_use_filename()
+    file: int | StrPath
+    file_type: str
+
+    def configure_subprocess(self, popen_kwargs: dict) -> None:
+        match self.file_type:
+            case JsonFileType.UNIX_FD:
+                # Unix file descriptor
+                popen_kwargs['pass_fds'] = [self.file]
+            case JsonFileType.WINDOWS_HANDLE:
+                # Windows handle
+                startupinfo = subprocess.STARTUPINFO()
+                startupinfo.lpAttributeList = {"handle_list": [self.file]}
+                popen_kwargs['startupinfo'] = startupinfo
+            case JsonFileType.FILENAME:
+                # Filename: nothing to do to
+                pass
+
+    @contextlib.contextmanager
+    def inherit_subprocess(self):
+        if self.file_type == JsonFileType.WINDOWS_HANDLE:
+            os.set_handle_inheritable(self.file, True)
+            try:
+                yield
+            finally:
+                os.set_handle_inheritable(self.file, False)
+        else:
+            yield
+
+    def open(self, mode, *, encoding):
+        file = self.file
+        if self.file_type == JsonFileType.WINDOWS_HANDLE:
+            import msvcrt
+            # Create a file descriptor from the handle
+            file = msvcrt.open_osfhandle(file, os.O_WRONLY)
+        return open(file, mode, encoding=encoding)
 
 
 @dataclasses.dataclass(slots=True, frozen=True)
@@ -44,7 +87,7 @@ class RunTests:
     python_cmd: tuple[str] | None
     randomize: bool
     random_seed: int | None
-    json_file: JsonFileType | None
+    json_file: JsonFile | None
 
     def copy(self, **override):
         state = dataclasses.asdict(self)
@@ -105,6 +148,8 @@ def _decode_runtests(data: dict[str, Any]) -> RunTests | dict[str, Any]:
         data.pop('__runtests__')
         if data['hunt_refleak']:
             data['hunt_refleak'] = HuntRefleak(**data['hunt_refleak'])
+        if data['json_file']:
+            data['json_file'] = JsonFile(**data['json_file'])
         return RunTests(**data)
     else:
         return data

Lib/test/libregrtest/worker.py

@@ -7,7 +7,7 @@
 from test.support import os_helper
 
 from .setup import setup_process, setup_test_dir
-from .runtests import RunTests, JsonFileType
+from .runtests import RunTests, JsonFile
 from .single import run_single_test
 from .utils import (
     StrPath, StrJSON, FilterTuple, MS_WINDOWS,
@@ -57,28 +57,10 @@ def create_worker_process(runtests: RunTests, output_fd: int,
 
     # Pass json_file to the worker process
     json_file = runtests.json_file
-    if isinstance(json_file, str):
-        # Filename: nothing to do to
-        pass
-    elif MS_WINDOWS:
-        # Windows handle
-        startupinfo = subprocess.STARTUPINFO()
-        startupinfo.lpAttributeList = {"handle_list": [json_file]}
-        kwargs['startupinfo'] = startupinfo
-    else:
-        # Unix file descriptor
-        kwargs['pass_fds'] = [json_file]
-
-    if USE_PROCESS_GROUP:
-        kwargs['start_new_session'] = True
+    json_file.configure_subprocess(kwargs)
 
-    if MS_WINDOWS:
-        os.set_handle_inheritable(json_file, True)
-    try:
+    with json_file.inherit_subprocess():
         return subprocess.Popen(cmd, **kwargs)
-    finally:
-        if MS_WINDOWS:
-            os.set_handle_inheritable(json_file, False)
 
 
 def worker_process(worker_json: StrJSON) -> NoReturn:
@@ -89,13 +71,7 @@ def worker_process(worker_json: StrJSON) -> NoReturn:
     # - Unix: file descriptor (int)
     # - Windows: handle (int)
     # - Emscripten/WASI or if --python is used: filename (str)
-    json_file: JsonFileType = runtests.json_file
-
-    if MS_WINDOWS:
-        import msvcrt
-        # Create a file descriptor from the handle
-        json_file = msvcrt.open_osfhandle(json_file, os.O_WRONLY)
-
+    json_file: JsonFile = runtests.json_file
 
     setup_test_dir(runtests.test_dir)
     setup_process()
@@ -109,7 +85,7 @@ def worker_process(worker_json: StrJSON) -> NoReturn:
 
     result = run_single_test(test_name, runtests)
 
-    with open(json_file, 'w', encoding='utf-8') as json_fp:
+    with json_file.open('w', encoding='utf-8') as json_fp:
         result.write_json_into(json_fp)
 
     sys.exit(0)