1"""
2Test execution: drives lifecycles around each collected test.
3"""
4
5import asyncio
6import dataclasses
7import inspect
8import sys
9import time
10import warnings
11from collections.abc import Callable
12from contextlib import ExitStack
13from dataclasses import dataclass, field
14
15from ..lifecycle import TestLifecycle
16from ..skipping import TestSkipped
17from .collection import RunnableTest
18from .failure import (
19 Failure,
20 describe_failure,
21 failure_that_could_not_be_described,
22 format_traceback,
23 shown_path,
24)
25from .output_capture import NO_OUTPUT, Output, OutputCapture
26from .phases import Part
27
28__all__ = []
29
30
31@dataclass
32class TestResult:
33 test: RunnableTest
34 outcome: str # "passed" | "failed" | "skipped"
35 duration: float = 0.0
36 # What a failed test says about its failure, already printed. The
37 # exception itself isn't kept: it would hold every frame it passed
38 # through, and everything those frames had in hand, until the run ends.
39 failure: Failure | None = None
40 # Why a skipped test was skipped — from `@skip` or from `skip_test()`.
41 skip_reason: str | None = None
42
43
44@dataclass
45class InterruptedTest:
46 """The test that was running when the run was stopped from outside."""
47
48 test: RunnableTest
49 # What it had written by then.
50 output: Output
51
52
53@dataclass
54class TeardownError:
55 """A lifecycle that raised while being taken down, after the last test."""
56
57 # Formatted, with the runner's own frames taken off the top.
58 traceback: str
59 # What had been written since the last test.
60 output: Output
61
62
63@dataclass
64class SetupFailure:
65 """
66 A lifecycle that didn't get through `setup_worker()`: it raised, or it
67 wrote its reason and called `sys.exit()`. No test was run.
68 """
69
70 # The lifecycle's class: "PostgresTestLifecycle".
71 lifecycle: str
72 # "ValueError", or "SystemExit" for one that exited.
73 error_type: str
74 error_message: str
75 # Formatted, with the runner's own frames taken off the top.
76 traceback: str
77
78
79@dataclass(frozen=True, kw_only=True)
80class RaisedWarning:
81 """A warning tests raised, however many times and in however many tests."""
82
83 # "DeprecationWarning"
84 category: str
85 message: str
86 # How many times it was raised, in all the tests that raised it.
87 count: int
88 # The first test that raised it, and the line of code it was about
89 # then. The file is relative to where the run started, when it is under
90 # there.
91 first_test: str
92 file: str
93 line: int
94
95
96@dataclass
97class TestRun:
98 results: list[TestResult]
99 duration: float
100 # Set when Ctrl-C stopped the run. The tests after it were not run.
101 interrupted: InterruptedTest | None = None
102 teardown_errors: list[TeardownError] = field(default_factory=list)
103 # Set when the lifecycles couldn't be set up. No test was run.
104 setup_failure: SetupFailure | None = None
105 # What the lifecycles wrote being set up, before the first test, and
106 # being taken down, after the last. It is no test's.
107 setup_output: Output = NO_OUTPUT
108 teardown_output: Output = NO_OUTPUT
109 # Each distinct warning once, in the order first raised.
110 warnings: list[RaisedWarning] = field(default_factory=list)
111 # How long each lifecycle took to set up and to take down, by its class,
112 # and how long the tests took between, first to last.
113 lifecycle_setup: tuple[Part, ...] = ()
114 lifecycle_teardown: tuple[Part, ...] = ()
115 tests_seconds: float = 0.0
116
117 @property
118 def passed(self) -> list[TestResult]:
119 return [r for r in self.results if r.outcome == "passed"]
120
121 @property
122 def failed(self) -> list[TestResult]:
123 return [r for r in self.results if r.outcome == "failed"]
124
125 @property
126 def skipped(self) -> list[TestResult]:
127 return [r for r in self.results if r.outcome == "skipped"]
128
129 @property
130 def ok(self) -> bool:
131 return not self.failed and self.interrupted is None
132
133
134def run_tests(
135 tests: list[RunnableTest],
136 *,
137 lifecycles: list[TestLifecycle],
138 fail_fast: bool = False,
139 full_values: bool = False,
140 on_result: Callable[[TestResult], None] | None = None,
141 capture: OutputCapture | None = None,
142) -> TestRun:
143 """
144 Run the tests, each inside every lifecycle.
145
146 `capture` is the run's hold on what is written to stdout and stderr. A
147 failed test's failure is given what the test wrote, and what a passing
148 test wrote is thrown away. Without one, output goes where it always
149 went.
150 """
151 if capture is None:
152 # Entered by nobody, it holds nothing and has nothing to give.
153 capture = OutputCapture(show_output=True)
154
155 run_start = time.monotonic()
156 results: list[TestResult] = []
157 interrupted = None
158 teardown_errors = []
159 setup_failure = None
160 setup_output = NO_OUTPUT
161 raised_warnings = _RaisedWarnings()
162 lifecycle_setup: list[Part] = []
163 lifecycle_teardown: list[Part] = []
164 tests_seconds = 0.0
165
166 # Track which lifecycles actually set up, so a failure partway through
167 # setup still tears down the ones that completed (e.g. drops the test
168 # database instead of leaking it).
169 started: list[TestLifecycle] = []
170 try:
171 for lifecycle in lifecycles:
172 setup_start = time.monotonic()
173 try:
174 lifecycle.setup_worker()
175 except KeyboardInterrupt:
176 raise
177 except BaseException as error:
178 # SystemExit too: code that can't set up writes why and
179 # calls `sys.exit()`, as creating the test database does
180 # when the server can't be reached.
181 setup_failure = SetupFailure(
182 lifecycle=type(lifecycle).__qualname__,
183 error_type=type(error).__qualname__,
184 error_message=str(error),
185 traceback=format_traceback(error),
186 )
187 break
188 finally:
189 lifecycle_setup.append(
190 Part(
191 name=type(lifecycle).__qualname__,
192 seconds=time.monotonic() - setup_start,
193 parts=_what_setup_spent_its_time_on(lifecycle)
194 if setup_failure is None
195 else (),
196 )
197 )
198 started.append(lifecycle)
199
200 # What setting up wrote is the run's, not the first test's.
201 setup_output = capture.take()
202
203 tests_start = time.monotonic()
204 try:
205 for test in tests if setup_failure is None else ():
206 try:
207 result = _run_one(
208 test,
209 lifecycles=lifecycles,
210 full_values=full_values,
211 capture=capture,
212 raised_warnings=raised_warnings,
213 )
214 except KeyboardInterrupt:
215 interrupted = InterruptedTest(test=test, output=capture.take())
216 break
217 results.append(result)
218 if on_result is not None:
219 on_result(result)
220 if fail_fast and result.outcome == "failed":
221 break
222 finally:
223 tests_seconds = time.monotonic() - tests_start
224 finally:
225 for lifecycle in reversed(started):
226 teardown_start = time.monotonic()
227 # One lifecycle's teardown failure shouldn't skip the others.
228 try:
229 lifecycle.teardown_worker()
230 except KeyboardInterrupt:
231 raise
232 except BaseException as error:
233 teardown_errors.append(
234 TeardownError(
235 traceback=format_traceback(error), output=capture.take()
236 )
237 )
238 finally:
239 lifecycle_teardown.append(
240 Part(
241 name=type(lifecycle).__qualname__,
242 seconds=time.monotonic() - teardown_start,
243 )
244 )
245
246 return TestRun(
247 results=results,
248 duration=time.monotonic() - run_start,
249 interrupted=interrupted,
250 teardown_errors=teardown_errors,
251 setup_failure=setup_failure,
252 setup_output=setup_output,
253 # What a teardown that raised wrote is with its error.
254 teardown_output=capture.take(),
255 warnings=raised_warnings.each_once(),
256 lifecycle_setup=tuple(lifecycle_setup),
257 lifecycle_teardown=tuple(lifecycle_teardown),
258 tests_seconds=tests_seconds,
259 )
260
261
262def _what_setup_spent_its_time_on(lifecycle: TestLifecycle) -> tuple[Part, ...]:
263 return tuple(
264 Part(name=name, seconds=seconds) for name, seconds in lifecycle.describe_setup()
265 )
266
267
268class _RaisedWarnings:
269 """The warnings a run's tests raised, each distinct one kept once."""
270
271 def __init__(self) -> None:
272 # The same warning is the same kind saying the same thing, wherever
273 # it is raised from: a deprecated function called from two hundred
274 # places is one thing to fix. A dict keeps the order they were first
275 # raised in.
276 self._raised: dict[tuple[str, str], RaisedWarning] = {}
277
278 def add(self, raised: list[warnings.WarningMessage], *, test: RunnableTest) -> None:
279 for warning in raised:
280 category = warning.category.__qualname__
281 message = str(warning.message)
282 key = (category, message)
283 before = self._raised.get(key)
284 if before is None:
285 self._raised[key] = RaisedWarning(
286 category=category,
287 message=message,
288 count=1,
289 first_test=test.id,
290 file=shown_path(warning.filename),
291 line=warning.lineno,
292 )
293 else:
294 self._raised[key] = dataclasses.replace(before, count=before.count + 1)
295
296 def each_once(self) -> list[RaisedWarning]:
297 return list(self._raised.values())
298
299
300def _run_one(
301 test: RunnableTest,
302 *,
303 lifecycles: list[TestLifecycle],
304 full_values: bool,
305 capture: OutputCapture,
306 raised_warnings: _RaisedWarnings,
307) -> TestResult:
308 """
309 Warnings the test raises are kept for the run to count, whatever comes
310 of the test. Python would have written each to stderr, where a passing
311 test's is thrown away with the rest of what it wrote.
312 """
313 with warnings.catch_warnings(record=True) as raised:
314 # Python ignores a DeprecationWarning unless it is raised by the
315 # script being run, which a test never is. A test run is where a
316 # deprecation is wanted: it says what to change before it breaks.
317 # With `-W` or PYTHONWARNINGS the filters are as they were given.
318 if not sys.warnoptions:
319 warnings.simplefilter("always", DeprecationWarning)
320 warnings.simplefilter("always", PendingDeprecationWarning)
321 result = _run_one_with_its_lifecycles(
322 test, lifecycles=lifecycles, full_values=full_values, capture=capture
323 )
324 raised_warnings.add(raised, test=test)
325 return result
326
327
328def _run_one_with_its_lifecycles(
329 test: RunnableTest,
330 *,
331 lifecycles: list[TestLifecycle],
332 full_values: bool,
333 capture: OutputCapture,
334) -> TestResult:
335 if test.skip_reason is not None:
336 return TestResult(test=test, outcome="skipped", skip_reason=test.skip_reason)
337
338 def described(error: BaseException) -> Failure:
339 try:
340 return describe_failure(
341 error,
342 test=test,
343 describers=[lifecycle.describe_value for lifecycle in lifecycles],
344 full_values=full_values,
345 )
346 except Exception as while_describing:
347 return failure_that_could_not_be_described(
348 error, test=test, while_describing=while_describing
349 )
350
351 start = time.monotonic()
352 described_error = None
353 failure = None
354 try:
355 with ExitStack() as stack:
356 for lifecycle in lifecycles:
357 stack.enter_context(lifecycle.around_test(test))
358 try:
359 outcome = test.func()
360 if inspect.iscoroutine(outcome):
361 asyncio.run(outcome)
362 except KeyboardInterrupt, TestSkipped:
363 raise
364 except BaseException as error:
365 # Described here, before the lifecycles exit. What the test
366 # had in hand is still as the test left it: its transaction
367 # hasn't been rolled back, its settings haven't been put
368 # back.
369 described_error = error
370 failure = described(error)
371 raise
372 except KeyboardInterrupt:
373 raise
374 except TestSkipped as skipped:
375 # Raised by `skip_test()` in the test body. It left through the
376 # lifecycles' `with` blocks like any exception, so they have exited.
377 capture.discard()
378 return TestResult(
379 test=test,
380 outcome="skipped",
381 duration=time.monotonic() - start,
382 skip_reason=skipped.reason,
383 )
384 except BaseException as error:
385 if failure is None or error is not described_error:
386 # A lifecycle raised, entering or exiting. Its error is the
387 # one that came out, with the test's own above it in the
388 # traceback if the test had failed too.
389 failure = described(error)
390 # Taken now that the lifecycles have exited: what they wrote on the
391 # way out (a rollback that failed) is the test's too.
392 output = capture.take()
393 failure = dataclasses.replace(
394 failure, stdout=output.stdout, stderr=output.stderr
395 )
396 return TestResult(
397 test=test,
398 outcome="failed",
399 duration=time.monotonic() - start,
400 failure=failure,
401 )
402
403 capture.discard()
404 return TestResult(test=test, outcome="passed", duration=time.monotonic() - start)