1"""
2Test output as text: answers "what do I do next", not just "what happened".
3
4Every failure block ends with the exact re-run command for that test,
5quoted so that pasting it into a shell runs that test and no other.
6
7What is printed here is read from the same `RunReport` that `--json` prints
8as a document. Nothing here works anything out about the run.
9"""
10
11import textwrap
12from typing import TextIO
13
14import click
15
16from ..definition import TestDefinitionError
17from .execution import RaisedWarning, TeardownError, TestResult
18from .failure import (
19 AssertedPart,
20 CollectionFailure,
21 Failure,
22 LocalValue,
23 format_collection_traceback,
24)
25from .output_capture import Output, StreamOutput
26from .phases import Phase, seconds_in_all, seconds_outside_the_tests
27from .printing import FULL_VALUES_FLAG, PrintedValue
28from .report import ParameterNothingPasses, RunReport
29
30__all__ = []
31
32# A run that spent this long on everything but the tests says where it went,
33# without being asked. Under it, a passing run is still two lines.
34SECONDS_OUTSIDE_THE_TESTS_WORTH_A_WORD = 1.0
35
36_STATUS_COLORS = {
37 "passed": "green",
38 "failed": "red",
39 "skipped": "yellow",
40}
41
42# Takes the cursor back to the start of the line and clears the line: what
43# the progress line is written over, and erased, with.
44_START_OF_A_CLEARED_LINE = "\r\x1b[K"
45
46
47def collection_error_text(cause: BaseException) -> str:
48 """
49 What to print for a file that couldn't be collected.
50
51 A TestDefinitionError is printed as its message: it already says what is
52 wrong and what to write instead. Anything else is an error in the test
53 file like any other (a NameError, an ImportError), and is printed with
54 the traceback that says where, starting at the test file.
55 """
56 if isinstance(cause, TestDefinitionError):
57 return str(cause)
58 return format_collection_traceback(cause)
59
60
61def failure_text(failure: Failure) -> str:
62 """
63 What is printed for one failed test, under its `failed` line: where it
64 failed, the assert with the values inside it, what differs, what else
65 the test had in hand, and what it wrote.
66 """
67 sections = [failure.traceback.rstrip()]
68
69 failed_assert = failure.failed_assert
70 if failed_assert is not None:
71 # The message the test gave is the traceback's last line, just
72 # above.
73 lines = [f"assert {failed_assert.expression}"]
74 for part in failed_assert.parts:
75 lines.extend(_part_lines(part))
76 sections.append("\n".join(lines))
77
78 diff = failed_assert.diff
79 if diff is not None:
80 lines = ["diff:", *(f" {line}" for line in diff.lines)]
81 if diff.cut_lines:
82 lines.append(
83 f" ... {diff.cut_lines:,} more lines"
84 f" ({FULL_VALUES_FLAG} prints them)"
85 )
86 sections.append("\n".join(lines))
87
88 if failure.locals:
89 lines = ["locals:"]
90 for local in failure.locals:
91 lines.extend(_local_lines(local))
92 sections.append("\n".join(lines))
93
94 sections.extend(
95 output_sections(Output(stdout=failure.stdout, stderr=failure.stderr))
96 )
97
98 return "\n\n".join(sections)
99
100
101def output_sections(output: Output) -> list[str]:
102 """What was written to stdout and to stderr, each under its name."""
103 sections = []
104 for name, stream in (("stdout", output.stdout), ("stderr", output.stderr)):
105 if stream.text:
106 sections.append("\n".join(_stream_lines(name, stream)))
107 return sections
108
109
110def _stream_lines(name: str, stream: StreamOutput) -> list[str]:
111 lines = [f"{name}:"]
112 if stream.cut_characters:
113 lines.append(
114 f" ... {stream.cut_characters:,} characters before this"
115 f" ({FULL_VALUES_FLAG} prints them)"
116 )
117 lines.extend(f" {line}" for line in stream.text.splitlines())
118 return lines
119
120
121def _part_lines(part: AssertedPart) -> list[str]:
122 indent = " " * (part.depth + 1)
123 if part.value is None:
124 return [f"{indent}{part.source} (not evaluated)"]
125 return _named_value_lines(part.source, part.value, indent=indent)
126
127
128def _local_lines(local: LocalValue) -> list[str]:
129 return _named_value_lines(local.name, local.value, indent=" ")
130
131
132def _named_value_lines(name: str, value: PrintedValue, *, indent: str) -> list[str]:
133 """
134 `name = value` on one line, or the name and then a value of several
135 lines under it. After a value the cap cut short, how much was cut.
136 """
137 if "\n" in value.text:
138 lines = [f"{indent}{name} ="]
139 lines.extend(f"{indent} {line}" for line in value.text.splitlines())
140 else:
141 lines = [f"{indent}{name} = {value.text}"]
142 if value.cut_characters:
143 lines.append(
144 f"{indent} ... {value.cut_characters:,} more characters"
145 f" ({FULL_VALUES_FLAG} prints them)"
146 )
147 return lines
148
149
150class TextReporter:
151 """
152 Prints a run as it goes, and what came of it when it is over.
153
154 `out` and `err` are where the runner's own output goes. While the run
155 holds what tests write, they are not `sys.stdout` and `sys.stderr`.
156
157 A test that passes prints nothing. With `verbose` each test prints a
158 line as it finishes. With `progress`, which is for a terminal someone is
159 watching, one line says how far the run has got: it is written over
160 itself as tests finish and erased before the report, so nothing of it
161 is left in what the run printed.
162 """
163
164 def __init__(
165 self,
166 *,
167 out: TextIO,
168 err: TextIO,
169 verbose: bool = False,
170 progress: bool = False,
171 ) -> None:
172 self.out = out
173 self.err = err
174 self.verbose = verbose
175 self.progress = progress and not verbose
176 self._collected = 0
177 self._finished = 0
178 self._failed = 0
179 self._progress_is_shown = False
180
181 def _print(
182 self,
183 text: str = "",
184 *,
185 newline: bool = True,
186 fg: str | None = None,
187 bold: bool = False,
188 dim: bool = False,
189 ) -> None:
190 click.secho(text, file=self.out, nl=newline, fg=fg, bold=bold, dim=dim)
191
192 def _print_error(self, text: str = "", *, fg: str | None = None) -> None:
193 click.secho(text, file=self.err, fg=fg)
194
195 def collected(self, count: int) -> None:
196 plural = "" if count == 1 else "s"
197 self._print(f"Collected {count} test{plural}", dim=True)
198 self._collected = count
199
200 def result(self, result: TestResult) -> None:
201 self._finished += 1
202 if result.outcome == "failed":
203 self._failed += 1
204
205 if self.verbose:
206 # The outcome as `--json` spells it.
207 line = f"{result.outcome:<7} {result.test.id}"
208 if result.outcome == "skipped":
209 line += f" ({result.skip_reason})"
210 else:
211 line += f" ({result.duration:.3f}s)"
212 self._print(
213 line,
214 fg=_STATUS_COLORS[result.outcome],
215 bold=result.outcome == "failed",
216 )
217 elif self.progress:
218 self._show_progress()
219
220 def _show_progress(self) -> None:
221 line = f"{self._finished} of {self._collected}"
222 if self._failed:
223 line += f", {self._failed} failed"
224 self.out.write(f"{_START_OF_A_CLEARED_LINE}{line}")
225 self.out.flush()
226 self._progress_is_shown = True
227
228 def _erase_progress(self) -> None:
229 if self._progress_is_shown:
230 self.out.write(_START_OF_A_CLEARED_LINE)
231 self.out.flush()
232 self._progress_is_shown = False
233
234 def _stopped(self, report: RunReport) -> None:
235 """The run ended before any test was run."""
236 stopped = report.stopped
237 assert stopped is not None
238 self._erase_progress()
239 if stopped.reason == "no_tests_found":
240 self._print(stopped.message, fg="yellow")
241 return
242
243 self._print_error(stopped.message, fg="red")
244 if stopped.traceback is not None:
245 self._print_error()
246 self._print_error(textwrap.indent(stopped.traceback.rstrip(), " "))
247 for section in output_sections(stopped.output):
248 self._print_error()
249 self._print_error(textwrap.indent(section, " "))
250
251 # The lifecycles that had been set up were taken down again.
252 if report.run is not None:
253 for error in report.run.teardown_errors:
254 self._print_error()
255 self._print_error("teardown error", fg="red")
256 self._print_error()
257 self._print_error(textwrap.indent(_teardown_error_text(error), " "))
258
259 if stopped.reason == "setup_error":
260 self._print_error()
261 self._print_error("No test was run.", fg="red")
262
263 def finished(self, report: RunReport) -> None:
264 if report.stopped is not None:
265 self._stopped(report)
266 return
267
268 run = report.run
269 assert run is not None
270
271 self._erase_progress()
272
273 for result in run.failed:
274 assert result.failure is not None
275 self._print()
276 self._print(f"failed {result.test.id}", fg="red", bold=True)
277 self._print()
278 self._print(textwrap.indent(failure_text(result.failure), " "))
279 self._print()
280 self._print(f"Re-run: {result.failure.rerun_command}", dim=True)
281
282 # Verbose output already gave each skipped test its own line.
283 if run.skipped and not self.verbose:
284 self._print()
285 for result in run.skipped:
286 self._print(
287 f"skipped {result.test.id} ({result.skip_reason})", fg="yellow"
288 )
289
290 # A definition error that is one paragraph is printed under its
291 # heading with no lines left blank. Eighty files with the same
292 # thing wrong are eighty of these, and every line is read.
293 after_a_short_one = False
294 for failure in report.collection_failures:
295 text = _collection_failure_text(failure)
296 is_short = failure.is_definition_error and "\n\n" not in text
297 if not (is_short and after_a_short_one):
298 self._print()
299 self._print(f"collection error {failure.file}", fg="red", bold=True)
300 if not is_short:
301 self._print()
302 self._print(textwrap.indent(text, " "))
303 after_a_short_one = is_short
304
305 # Each file's error has the parameters its own tests take. With
306 # more than one such file, this is what they come to.
307 files_taking_parameters = [
308 failure for failure in report.collection_failures if failure.parameters
309 ]
310 if len(files_taking_parameters) > 1:
311 self._print()
312 self._print("parameters nothing passes in", fg="red", bold=True)
313 self._print()
314 self._print(
315 textwrap.indent(
316 _parameters_text(
317 report.parameters_nothing_passes,
318 files=len(files_taking_parameters),
319 ),
320 " ",
321 )
322 )
323
324 for error in run.teardown_errors:
325 self._print()
326 self._print("teardown error", fg="red", bold=True)
327 self._print()
328 self._print(textwrap.indent(_teardown_error_text(error), " "))
329
330 if run.interrupted is not None:
331 self._print()
332 self._print(f"interrupted {run.interrupted.test.id}", fg="red", bold=True)
333 for section in output_sections(run.interrupted.output):
334 self._print()
335 self._print(textwrap.indent(section, " "))
336
337 if report.warnings:
338 self._print()
339 self._print("warnings", fg="yellow", bold=True)
340 self._print()
341 for warning in report.warnings:
342 self._print(textwrap.indent(_warning_text(warning), " "))
343
344 if report.output:
345 self._print()
346 self._print("written outside any test", bold=True)
347 for section in output_sections(report.output):
348 self._print()
349 self._print(textwrap.indent(section, " "))
350
351 self._summary(report)
352
353 def _summary(self, report: RunReport) -> None:
354 run = report.run
355 assert run is not None
356 counts = report.counts
357
358 parts = [f"{counts.passed} passed"]
359 if counts.failed:
360 parts.append(f"{counts.failed} failed")
361 if counts.skipped:
362 parts.append(f"{counts.skipped} skipped")
363 if counts.collection_errors:
364 plural = "" if counts.collection_errors == 1 else "s"
365 parts.append(f"{counts.collection_errors} collection error{plural}")
366 if counts.not_run:
367 parts.append(f"{counts.not_run} not run")
368 if counts.warnings:
369 plural = "" if counts.warnings == 1 else "s"
370 parts.append(f"{counts.warnings} warning{plural}")
371 line = f"{', '.join(parts)} in {run.duration:.2f}s"
372 if run.interrupted is not None:
373 line = f"Interrupted: {line}"
374
375 self._print()
376 self._print(line, fg="green" if report.exit_code == 0 else "red", bold=True)
377
378 outside = seconds_outside_the_tests(report.phases)
379 if self.verbose or outside > SECONDS_OUTSIDE_THE_TESTS_WORTH_A_WORD:
380 self._print()
381 self._print(phases_heading(report.phases), bold=True)
382 self._print(textwrap.indent(phases_text(report.phases), " "))
383
384
385def phases_heading(phases: tuple[Phase, ...]) -> str:
386 return f"where the {seconds_in_all(phases):.2f}s went"
387
388
389def phases_text(phases: tuple[Phase, ...]) -> str:
390 """
391 The phases as a timeline, one line each, with a phase's parts under it:
392
393 python startup 0.04s cpu time, before Plain was imported
394 command 0.05s
395 runtime setup 0.77s
396 hook dev-setup 0.30s
397 import app.agents 0.40s
398 tests 0.02s
399 """
400 names = max(len(_phase_label(phase.name)) for phase in phases)
401 for phase in phases:
402 for part in phase.parts:
403 names = max(names, len(part.name) + 2)
404 for inner in part.parts:
405 names = max(names, len(inner.name) + 4)
406
407 lines = []
408 for phase in phases:
409 line = f"{_phase_label(phase.name).ljust(names)} {phase.seconds:6.2f}s"
410 if phase.measured == "cpu":
411 line += " cpu time, before Plain was imported"
412 lines.append(line)
413 for part in phase.parts:
414 lines.append(f" {part.name.ljust(names - 2)} {part.seconds:6.2f}s")
415 for inner in part.parts:
416 lines.append(
417 f" {inner.name.ljust(names - 4)} {inner.seconds:6.2f}s"
418 )
419 return "\n".join(lines)
420
421
422def _phase_label(name: str) -> str:
423 return name.replace("_", " ")
424
425
426def _teardown_error_text(error: TeardownError) -> str:
427 sections = [error.traceback.rstrip(), *output_sections(error.output)]
428 return "\n\n".join(sections)
429
430
431def _warning_text(warning: RaisedWarning) -> str:
432 """The warning, and under it how often it was raised and where first."""
433 times = "once" if warning.count == 1 else f"{warning.count:,} times"
434 return (
435 f"{warning.category}: {warning.message}\n"
436 f" raised {times}, first at {warning.file}:{warning.line}"
437 f" in {warning.first_test}"
438 )
439
440
441def _parameters_text(
442 parameters: tuple[ParameterNothingPasses, ...], *, files: int
443) -> str:
444 """
445 The parameters tests take across a run, as a table:
446
447 Tests in 63 files take these:
448
449 db 605 tests in 61 files
450 client 96 tests in 12 files
451 """
452 names = max(len(parameter.name) for parameter in parameters)
453 how_many = [
454 "1 test" if parameter.tests == 1 else f"{parameter.tests} tests"
455 for parameter in parameters
456 ]
457 widest = max(len(tests) for tests in how_many)
458
459 lines = [f"Tests in {files} files take these:", ""]
460 for parameter, tests in zip(parameters, how_many, strict=True):
461 in_files = "1 file" if parameter.files == 1 else f"{parameter.files} files"
462 lines.append(
463 f"{parameter.name.ljust(names)} {tests.rjust(widest)} in {in_files}"
464 )
465 return "\n".join(lines)
466
467
468def _collection_failure_text(failure: CollectionFailure) -> str:
469 text = failure.message if failure.is_definition_error else failure.traceback
470 assert text is not None
471 sections = [
472 text,
473 *output_sections(Output(stdout=failure.stdout, stderr=failure.stderr)),
474 ]
475 return "\n\n".join(sections)