1 import contextlib
2 import dataclasses
3 import json
4 import os
5 import subprocess
6 from typing import Any
7
8 from test import support
9
10 from .utils import (
11 StrPath, StrJSON, TestTuple, TestFilter, FilterTuple, FilterDict)
12
13
14 class ESC[4;38;5;81mJsonFileType:
15 UNIX_FD = "UNIX_FD"
16 WINDOWS_HANDLE = "WINDOWS_HANDLE"
17 STDOUT = "STDOUT"
18
19
20 @dataclasses.dataclass(slots=True, frozen=True)
21 class ESC[4;38;5;81mJsonFile:
22 # file type depends on file_type:
23 # - UNIX_FD: file descriptor (int)
24 # - WINDOWS_HANDLE: handle (int)
25 # - STDOUT: use process stdout (None)
26 file: int | None
27 file_type: str
28
29 def configure_subprocess(self, popen_kwargs: dict) -> None:
30 match self.file_type:
31 case JsonFileType.UNIX_FD:
32 # Unix file descriptor
33 popen_kwargs['pass_fds'] = [self.file]
34 case JsonFileType.WINDOWS_HANDLE:
35 # Windows handle
36 # We run mypy with `--platform=linux` so it complains about this:
37 startupinfo = subprocess.STARTUPINFO() # type: ignore[attr-defined]
38 startupinfo.lpAttributeList = {"handle_list": [self.file]}
39 popen_kwargs['startupinfo'] = startupinfo
40
41 @contextlib.contextmanager
42 def inherit_subprocess(self):
43 if self.file_type == JsonFileType.WINDOWS_HANDLE:
44 os.set_handle_inheritable(self.file, True)
45 try:
46 yield
47 finally:
48 os.set_handle_inheritable(self.file, False)
49 else:
50 yield
51
52 def open(self, mode='r', *, encoding):
53 if self.file_type == JsonFileType.STDOUT:
54 raise ValueError("for STDOUT file type, just use sys.stdout")
55
56 file = self.file
57 if self.file_type == JsonFileType.WINDOWS_HANDLE:
58 import msvcrt
59 # Create a file descriptor from the handle
60 file = msvcrt.open_osfhandle(file, os.O_WRONLY)
61 return open(file, mode, encoding=encoding)
62
63
64 @dataclasses.dataclass(slots=True, frozen=True)
65 class ESC[4;38;5;81mHuntRefleak:
66 warmups: int
67 runs: int
68 filename: StrPath
69
70
71 @dataclasses.dataclass(slots=True, frozen=True)
72 class ESC[4;38;5;81mRunTests:
73 tests: TestTuple
74 fail_fast: bool
75 fail_env_changed: bool
76 match_tests: TestFilter
77 match_tests_dict: FilterDict | None
78 rerun: bool
79 forever: bool
80 pgo: bool
81 pgo_extended: bool
82 output_on_failure: bool
83 timeout: float | None
84 verbose: int
85 quiet: bool
86 hunt_refleak: HuntRefleak | None
87 test_dir: StrPath | None
88 use_junit: bool
89 memory_limit: str | None
90 gc_threshold: int | None
91 use_resources: tuple[str, ...]
92 python_cmd: tuple[str, ...] | None
93 randomize: bool
94 random_seed: int | str
95
96 def copy(self, **override) -> 'RunTests':
97 state = dataclasses.asdict(self)
98 state.update(override)
99 return RunTests(**state)
100
101 def create_worker_runtests(self, **override):
102 state = dataclasses.asdict(self)
103 state.update(override)
104 return WorkerRunTests(**state)
105
106 def get_match_tests(self, test_name) -> FilterTuple | None:
107 if self.match_tests_dict is not None:
108 return self.match_tests_dict.get(test_name, None)
109 else:
110 return None
111
112 def get_jobs(self):
113 # Number of run_single_test() calls needed to run all tests.
114 # None means that there is not bound limit (--forever option).
115 if self.forever:
116 return None
117 return len(self.tests)
118
119 def iter_tests(self):
120 if self.forever:
121 while True:
122 yield from self.tests
123 else:
124 yield from self.tests
125
126 def json_file_use_stdout(self) -> bool:
127 # Use STDOUT in two cases:
128 #
129 # - If --python command line option is used;
130 # - On Emscripten and WASI.
131 #
132 # On other platforms, UNIX_FD or WINDOWS_HANDLE can be used.
133 return (
134 bool(self.python_cmd)
135 or support.is_emscripten
136 or support.is_wasi
137 )
138
139
140 @dataclasses.dataclass(slots=True, frozen=True)
141 class ESC[4;38;5;81mWorkerRunTests(ESC[4;38;5;149mRunTests):
142 json_file: JsonFile
143
144 def as_json(self) -> StrJSON:
145 return json.dumps(self, cls=_EncodeRunTests)
146
147 @staticmethod
148 def from_json(worker_json: StrJSON) -> 'WorkerRunTests':
149 return json.loads(worker_json, object_hook=_decode_runtests)
150
151
152 class ESC[4;38;5;81m_EncodeRunTests(ESC[4;38;5;149mjsonESC[4;38;5;149m.ESC[4;38;5;149mJSONEncoder):
153 def default(self, o: Any) -> dict[str, Any]:
154 if isinstance(o, WorkerRunTests):
155 result = dataclasses.asdict(o)
156 result["__runtests__"] = True
157 return result
158 else:
159 return super().default(o)
160
161
162 def _decode_runtests(data: dict[str, Any]) -> RunTests | dict[str, Any]:
163 if "__runtests__" in data:
164 data.pop('__runtests__')
165 if data['hunt_refleak']:
166 data['hunt_refleak'] = HuntRefleak(**data['hunt_refleak'])
167 if data['json_file']:
168 data['json_file'] = JsonFile(**data['json_file'])
169 return WorkerRunTests(**data)
170 else:
171 return data