1 import dataclasses
2 import json
3 from typing import Any
4
5 from .utils import (
6 StrJSON, TestName, FilterTuple,
7 format_duration, normalize_test_name, print_warning)
8
9
10 @dataclasses.dataclass(slots=True)
11 class ESC[4;38;5;81mTestStats:
12 tests_run: int = 0
13 failures: int = 0
14 skipped: int = 0
15
16 @staticmethod
17 def from_unittest(result):
18 return TestStats(result.testsRun,
19 len(result.failures),
20 len(result.skipped))
21
22 @staticmethod
23 def from_doctest(results):
24 return TestStats(results.attempted,
25 results.failed,
26 results.skipped)
27
28 def accumulate(self, stats):
29 self.tests_run += stats.tests_run
30 self.failures += stats.failures
31 self.skipped += stats.skipped
32
33
34 # Avoid enum.Enum to reduce the number of imports when tests are run
35 class ESC[4;38;5;81mState:
36 PASSED = "PASSED"
37 FAILED = "FAILED"
38 SKIPPED = "SKIPPED"
39 UNCAUGHT_EXC = "UNCAUGHT_EXC"
40 REFLEAK = "REFLEAK"
41 ENV_CHANGED = "ENV_CHANGED"
42 RESOURCE_DENIED = "RESOURCE_DENIED"
43 INTERRUPTED = "INTERRUPTED"
44 WORKER_FAILED = "WORKER_FAILED" # non-zero worker process exit code
45 WORKER_BUG = "WORKER_BUG" # exception when running a worker
46 DID_NOT_RUN = "DID_NOT_RUN"
47 TIMEOUT = "TIMEOUT"
48
49 @staticmethod
50 def is_failed(state):
51 return state in {
52 State.FAILED,
53 State.UNCAUGHT_EXC,
54 State.REFLEAK,
55 State.WORKER_FAILED,
56 State.WORKER_BUG,
57 State.TIMEOUT}
58
59 @staticmethod
60 def has_meaningful_duration(state):
61 # Consider that the duration is meaningless for these cases.
62 # For example, if a whole test file is skipped, its duration
63 # is unlikely to be the duration of executing its tests,
64 # but just the duration to execute code which skips the test.
65 return state not in {
66 State.SKIPPED,
67 State.RESOURCE_DENIED,
68 State.INTERRUPTED,
69 State.WORKER_FAILED,
70 State.WORKER_BUG,
71 State.DID_NOT_RUN}
72
73 @staticmethod
74 def must_stop(state):
75 return state in {
76 State.INTERRUPTED,
77 State.WORKER_BUG,
78 }
79
80
81 @dataclasses.dataclass(slots=True)
82 class ESC[4;38;5;81mTestResult:
83 test_name: TestName
84 state: str | None = None
85 # Test duration in seconds
86 duration: float | None = None
87 xml_data: list[str] | None = None
88 stats: TestStats | None = None
89
90 # errors and failures copied from support.TestFailedWithDetails
91 errors: list[tuple[str, str]] | None = None
92 failures: list[tuple[str, str]] | None = None
93
94 def is_failed(self, fail_env_changed: bool) -> bool:
95 if self.state == State.ENV_CHANGED:
96 return fail_env_changed
97 return State.is_failed(self.state)
98
99 def _format_failed(self):
100 if self.errors and self.failures:
101 le = len(self.errors)
102 lf = len(self.failures)
103 error_s = "error" + ("s" if le > 1 else "")
104 failure_s = "failure" + ("s" if lf > 1 else "")
105 return f"{self.test_name} failed ({le} {error_s}, {lf} {failure_s})"
106
107 if self.errors:
108 le = len(self.errors)
109 error_s = "error" + ("s" if le > 1 else "")
110 return f"{self.test_name} failed ({le} {error_s})"
111
112 if self.failures:
113 lf = len(self.failures)
114 failure_s = "failure" + ("s" if lf > 1 else "")
115 return f"{self.test_name} failed ({lf} {failure_s})"
116
117 return f"{self.test_name} failed"
118
119 def __str__(self) -> str:
120 match self.state:
121 case State.PASSED:
122 return f"{self.test_name} passed"
123 case State.FAILED:
124 return self._format_failed()
125 case State.SKIPPED:
126 return f"{self.test_name} skipped"
127 case State.UNCAUGHT_EXC:
128 return f"{self.test_name} failed (uncaught exception)"
129 case State.REFLEAK:
130 return f"{self.test_name} failed (reference leak)"
131 case State.ENV_CHANGED:
132 return f"{self.test_name} failed (env changed)"
133 case State.RESOURCE_DENIED:
134 return f"{self.test_name} skipped (resource denied)"
135 case State.INTERRUPTED:
136 return f"{self.test_name} interrupted"
137 case State.WORKER_FAILED:
138 return f"{self.test_name} worker non-zero exit code"
139 case State.WORKER_BUG:
140 return f"{self.test_name} worker bug"
141 case State.DID_NOT_RUN:
142 return f"{self.test_name} ran no tests"
143 case State.TIMEOUT:
144 return f"{self.test_name} timed out ({format_duration(self.duration)})"
145 case _:
146 raise ValueError("unknown result state: {state!r}")
147
148 def has_meaningful_duration(self):
149 return State.has_meaningful_duration(self.state)
150
151 def set_env_changed(self):
152 if self.state is None or self.state == State.PASSED:
153 self.state = State.ENV_CHANGED
154
155 def must_stop(self, fail_fast: bool, fail_env_changed: bool) -> bool:
156 if State.must_stop(self.state):
157 return True
158 if fail_fast and self.is_failed(fail_env_changed):
159 return True
160 return False
161
162 def get_rerun_match_tests(self) -> FilterTuple | None:
163 match_tests = []
164
165 errors = self.errors or []
166 failures = self.failures or []
167 for error_list, is_error in (
168 (errors, True),
169 (failures, False),
170 ):
171 for full_name, *_ in error_list:
172 match_name = normalize_test_name(full_name, is_error=is_error)
173 if match_name is None:
174 # 'setUpModule (test.test_sys)': don't filter tests
175 return None
176 if not match_name:
177 error_type = "ERROR" if is_error else "FAIL"
178 print_warning(f"rerun failed to parse {error_type} test name: "
179 f"{full_name!r}: don't filter tests")
180 return None
181 match_tests.append(match_name)
182
183 if not match_tests:
184 return None
185 return tuple(match_tests)
186
187 def write_json_into(self, file) -> None:
188 json.dump(self, file, cls=_EncodeTestResult)
189
190 @staticmethod
191 def from_json(worker_json: StrJSON) -> 'TestResult':
192 return json.loads(worker_json, object_hook=_decode_test_result)
193
194
195 class ESC[4;38;5;81m_EncodeTestResult(ESC[4;38;5;149mjsonESC[4;38;5;149m.ESC[4;38;5;149mJSONEncoder):
196 def default(self, o: Any) -> dict[str, Any]:
197 if isinstance(o, TestResult):
198 result = dataclasses.asdict(o)
199 result["__test_result__"] = o.__class__.__name__
200 return result
201 else:
202 return super().default(o)
203
204
205 def _decode_test_result(data: dict[str, Any]) -> TestResult | dict[str, Any]:
206 if "__test_result__" in data:
207 data.pop('__test_result__')
208 if data['stats'] is not None:
209 data['stats'] = TestStats(**data['stats'])
210 return TestResult(**data)
211 else:
212 return data