1"""
2Test output as one JSON document, for `--json`.
3
4The document is a `RunReport` written out. Everything the runner knew
5separately is a field of its own: nothing has to be read out of a string
6to find a file, a line, a test or a value.
7
8Every object of one kind has the same fields, whatever happened. A field
9with nothing to say is `null`, or an empty list, or an empty string, and is
10still there.
11"""
12
13import json
14from typing import Any, TextIO
15
16from .execution import TestResult
17from .failure import CollectionFailure, Failure, where_defined
18from .output_capture import StreamOutput
19from .phases import Part
20from .printing import PrintedValue
21from .report import RunReport
22
23__all__ = []
24
25# Raised when a field is renamed, removed, or changes what it means. A
26# field that is added doesn't raise it.
27DOCUMENT_VERSION = 1
28
29
30def document_text(report: RunReport, *, list_passed: bool) -> str:
31 return json.dumps(document(report, list_passed=list_passed), indent=2)
32
33
34def document(report: RunReport, *, list_passed: bool) -> dict[str, Any]:
35 run = report.run
36 counts = report.counts
37 command = report.command
38
39 tests = []
40 interrupted = None
41 teardown_errors = []
42 if run is not None:
43 tests = [
44 _test(result)
45 for result in run.results
46 if list_passed or result.outcome != "passed"
47 ]
48 if run.interrupted is not None:
49 file, line = where_defined(run.interrupted.test)
50 interrupted = {
51 "id": run.interrupted.test.id,
52 "file": file,
53 "line": line,
54 "stdout": _stream(run.interrupted.output.stdout),
55 "stderr": _stream(run.interrupted.output.stderr),
56 }
57 teardown_errors = [
58 {
59 "traceback": error.traceback,
60 "stdout": _stream(error.output.stdout),
61 "stderr": _stream(error.output.stderr),
62 }
63 for error in run.teardown_errors
64 ]
65
66 stopped = None
67 if report.stopped is not None:
68 stopped = {
69 "reason": report.stopped.reason,
70 "message": report.stopped.message,
71 "traceback": report.stopped.traceback,
72 "stdout": _stream(report.stopped.output.stdout),
73 "stderr": _stream(report.stopped.output.stderr),
74 }
75
76 return {
77 "version": DOCUMENT_VERSION,
78 "outcome": report.outcome,
79 "exit_code": report.exit_code,
80 "duration": round(run.duration, 4) if run is not None else 0.0,
81 "command": {
82 "argv": list(command.argv),
83 "directory": command.directory,
84 "targets": list(command.targets),
85 "match": command.match,
86 "tags": list(command.tags),
87 "exclude_tags": list(command.exclude_tags),
88 "fail_fast": command.fail_fast,
89 "full_values": command.full_values,
90 },
91 "counts": {
92 "selected": counts.selected,
93 "passed": counts.passed,
94 "failed": counts.failed,
95 "skipped": counts.skipped,
96 "not_run": counts.not_run,
97 "collection_errors": counts.collection_errors,
98 "warnings": counts.warnings,
99 },
100 "tests_listed": "all" if list_passed else "failed_and_skipped",
101 "tests": tests,
102 "collection_errors": [
103 _collection_error(failure) for failure in report.collection_failures
104 ],
105 # What the collection errors come to: each parameter that tests take
106 # and nothing passes in, over the whole run.
107 "parameters_nothing_passes": [
108 {
109 "name": parameter.name,
110 "tests": parameter.tests,
111 "files": parameter.files,
112 }
113 for parameter in report.parameters_nothing_passes
114 ],
115 "warnings": [
116 {
117 "category": warning.category,
118 "message": warning.message,
119 "count": warning.count,
120 "first_test": warning.first_test,
121 "file": warning.file,
122 "line": warning.line,
123 }
124 for warning in report.warnings
125 ],
126 "stopped": stopped,
127 "interrupted": interrupted,
128 "teardown_errors": teardown_errors,
129 # What was written outside any test: setting up the app and the
130 # lifecycles, and taking them down. A stopped run's is in `stopped`.
131 "stdout": _stream(report.output.stdout),
132 "stderr": _stream(report.output.stderr),
133 # Where the time went, in the order it went there.
134 "phases": [
135 {
136 "name": phase.name,
137 "seconds": round(phase.seconds, 4),
138 "measured": phase.measured,
139 "parts": _parts(phase.parts),
140 }
141 for phase in report.phases
142 ],
143 }
144
145
146def _parts(parts: tuple[Part, ...]) -> list[dict[str, Any]]:
147 return [
148 {
149 "name": part.name,
150 "seconds": round(part.seconds, 4),
151 "parts": _parts(part.parts),
152 }
153 for part in parts
154 ]
155
156
157def _test(result: TestResult) -> dict[str, Any]:
158 file, line = where_defined(result.test)
159 return {
160 "id": result.test.id,
161 "file": file,
162 "line": line,
163 "name": result.test.name,
164 "tags": list(result.test.tags),
165 "outcome": result.outcome,
166 "duration": round(result.duration, 4),
167 "skip_reason": result.skip_reason,
168 "failure": _failure(result.failure) if result.failure is not None else None,
169 }
170
171
172def _failure(failure: Failure) -> dict[str, Any]:
173 failed_assert = None
174 if failure.failed_assert is not None:
175 diff = None
176 if failure.failed_assert.diff is not None:
177 diff = {
178 "lines": list(failure.failed_assert.diff.lines),
179 "cut_lines": failure.failed_assert.diff.cut_lines,
180 }
181 failed_assert = {
182 "expression": failure.failed_assert.expression,
183 "message": failure.failed_assert.message,
184 "parts": [
185 {
186 "source": part.source,
187 "depth": part.depth,
188 "evaluated": part.value is not None,
189 "value": _value(part.value) if part.value is not None else None,
190 }
191 for part in failure.failed_assert.parts
192 ],
193 "diff": diff,
194 }
195
196 return {
197 "error_type": failure.error_type,
198 "error_message": failure.error_message,
199 "file": failure.file,
200 "line": failure.line,
201 "traceback": failure.traceback,
202 "frames": [
203 {"file": frame.file, "line": frame.line, "function": frame.function}
204 for frame in failure.frames
205 ],
206 "assert": failed_assert,
207 "locals": [
208 {"name": local.name, "value": _value(local.value)}
209 for local in failure.locals
210 ],
211 "stdout": _stream(failure.stdout),
212 "stderr": _stream(failure.stderr),
213 "rerun_command": failure.rerun_command,
214 }
215
216
217def _collection_error(failure: CollectionFailure) -> dict[str, Any]:
218 return {
219 "file": failure.file,
220 "line": failure.line,
221 "is_definition_error": failure.is_definition_error,
222 "error_type": failure.error_type,
223 "message": failure.message,
224 "traceback": failure.traceback,
225 "stdout": _stream(failure.stdout),
226 "stderr": _stream(failure.stderr),
227 }
228
229
230def _value(value: PrintedValue) -> dict[str, Any]:
231 return {
232 "text": value.text,
233 "cut_characters": value.cut_characters,
234 "same_as": value.same_as,
235 }
236
237
238def _stream(stream: StreamOutput) -> dict[str, Any]:
239 return {"text": stream.text, "cut_characters": stream.cut_characters}
240
241
242class JsonReporter:
243 """
244 Prints nothing while the run goes, and the document when it is over.
245
246 The document is all that is written to `out`.
247 """
248
249 def __init__(self, *, out: TextIO, list_passed: bool) -> None:
250 self.out = out
251 self.list_passed = list_passed
252
253 def collected(self, count: int) -> None:
254 pass
255
256 def result(self, result: TestResult) -> None:
257 pass
258
259 def finished(self, report: RunReport) -> None:
260 self.out.write(document_text(report, list_passed=self.list_passed))
261 self.out.write("\n")
262 self.out.flush()