1"""
2What a run writes to stdout and stderr, held until it is known whether
3anyone needs to read it.
4
5A test that passes wrote nothing worth reading, and what it wrote is thrown
6away. A test that fails has what it wrote printed with its failure, and
7nowhere else.
8
9The two descriptors themselves are pointed at files, 1 and 2, not
10`sys.stdout` and `sys.stderr`. Replacing those two names would miss most of
11what a test writes: a logging handler keeps the stream object it was given
12when logging was configured, before any test ran; a subprocess and a C
13extension write to the descriptor and never look at `sys`. Everything that
14writes ends up at the descriptor, so that is where it is held.
15
16The runner's own output goes to the descriptors as they were, through
17`real_stdout` and `real_stderr`.
18"""
19
20import codecs
21import faulthandler
22import os
23import sys
24import tempfile
25from dataclasses import dataclass
26from types import TracebackType
27from typing import IO, Self, TextIO
28
29__all__ = []
30
31# The most of one stream that is kept for one test, in characters. What is
32# kept is the end: the last thing written is the nearest to the failure.
33OUTPUT_CAP = 10_000
34
35# The flag that lets output through as it is written.
36SHOW_OUTPUT_FLAG = "--show-output"
37
38_READ_AT_ONCE = 1024 * 1024
39
40
41@dataclass(frozen=True, kw_only=True)
42class StreamOutput:
43 """What was written to one stream, as a report prints it."""
44
45 text: str
46 # How many characters the cap left out, from the start. 0 when `text`
47 # is everything that was written.
48 cut_characters: int = 0
49
50
51@dataclass(frozen=True, kw_only=True)
52class Output:
53 """What was written to stdout and to stderr, each in its own order."""
54
55 stdout: StreamOutput
56 stderr: StreamOutput
57
58 def __bool__(self) -> bool:
59 return bool(self.stdout.text or self.stderr.text)
60
61
62NO_OUTPUT = Output(stdout=StreamOutput(text=""), stderr=StreamOutput(text=""))
63
64
65def joined(outputs: list[Output]) -> Output:
66 """Several outputs as one, each stream in the order the outputs are in."""
67 return Output(
68 stdout=_joined_stream([output.stdout for output in outputs]),
69 stderr=_joined_stream([output.stderr for output in outputs]),
70 )
71
72
73def _joined_stream(streams: list[StreamOutput]) -> StreamOutput:
74 return StreamOutput(
75 text="".join(stream.text for stream in streams),
76 cut_characters=sum(stream.cut_characters for stream in streams),
77 )
78
79
80class _HeldDescriptor:
81 """One of the process's output descriptors, pointed at a file."""
82
83 def __init__(self, descriptor: int) -> None:
84 self.descriptor = descriptor
85 self.real = os.dup(descriptor)
86 self.file, self.path_still_to_remove = _file_that_appends()
87 self.hold()
88
89 def hold(self) -> None:
90 os.dup2(self.file.fileno(), self.descriptor)
91
92 def let_through(self) -> None:
93 os.dup2(self.real, self.descriptor)
94
95 def close(self) -> None:
96 self.let_through()
97 os.close(self.real)
98 self.file.close()
99 if self.path_still_to_remove is not None:
100 os.unlink(self.path_still_to_remove)
101
102 def empty(self) -> None:
103 self.file.truncate(0)
104
105 def written(self, *, full_output: bool) -> StreamOutput:
106 size = os.fstat(self.file.fileno()).st_size
107 if size == 0:
108 return StreamOutput(text="")
109
110 if full_output:
111 self.file.seek(0)
112 return StreamOutput(text=self.file.read().decode(errors="replace"))
113
114 # A character is at most four bytes, so the last OUTPUT_CAP
115 # characters are within the last four times as many bytes.
116 kept_from = max(0, size - OUTPUT_CAP * 4)
117 self.file.seek(kept_from)
118 text = self.file.read().decode(errors="replace")
119 if kept_from == 0 and len(text) <= OUTPUT_CAP:
120 return StreamOutput(text=text)
121
122 kept = text[-OUTPUT_CAP:]
123 return StreamOutput(
124 text=kept, cut_characters=self._characters_in_file() - len(kept)
125 )
126
127 def _characters_in_file(self) -> int:
128 """Counted a piece at a time: the file can be any size."""
129 decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
130 characters = 0
131 self.file.seek(0)
132 while piece := self.file.read(_READ_AT_ONCE):
133 characters += len(decoder.decode(piece))
134 return characters + len(decoder.decode(b"", final=True))
135
136
137class OutputCapture:
138 """
139 Holds what is written to stdout and stderr from entering to exiting.
140
141 with OutputCapture() as capture:
142 ...
143 output = capture.take()
144
145 With `show_output`, nothing is held: output goes where it always went,
146 as it is written, and `take()` has nothing to give.
147 """
148
149 def __init__(self, *, show_output: bool = False, full_output: bool = False) -> None:
150 self.show_output = show_output
151 self.full_output = full_output
152 self.real_stdout: TextIO = sys.stdout
153 self.real_stderr: TextIO = sys.stderr
154 self._held: list[_HeldDescriptor] = []
155 self._let_through_for_a_debugger = False
156 self._fault_handler_was_enabled = False
157 self._stdout_wrote_a_line_at_a_time = False
158 self._previous_breakpoint_hook = sys.breakpointhook
159
160 def __enter__(self) -> Self:
161 if self.show_output:
162 return self
163
164 _flush_what_python_is_holding()
165 stdout = _HeldDescriptor(1)
166 stderr = _HeldDescriptor(2)
167 self._held = [stdout, stderr]
168
169 # Python writes what is printed when it has a few thousand
170 # characters of it, unless stdout is a terminal. A line printed
171 # before a subprocess ran would be written after what the
172 # subprocess wrote. Written a line at a time, what is held is in
173 # the order it was written in.
174 self._stdout_wrote_a_line_at_a_time = _line_buffering_of(sys.stdout)
175 _write_a_line_at_a_time(sys.stdout, True)
176 self.real_stdout = _text_stream(stdout.real, like=sys.stdout)
177 self.real_stderr = _text_stream(stderr.real, like=sys.stderr)
178
179 # A crash of the interpreter itself is reported on stderr, by
180 # `faulthandler`. It must reach the terminal: nothing would be left
181 # running to print what was held.
182 self._fault_handler_was_enabled = faulthandler.is_enabled()
183 faulthandler.enable(file=self.real_stderr)
184
185 # A debugger needs the terminal.
186 self._previous_breakpoint_hook = sys.breakpointhook
187 sys.breakpointhook = self._breakpoint # ty: ignore[invalid-assignment]
188 return self
189
190 def __exit__(
191 self,
192 error_type: type[BaseException] | None,
193 error: BaseException | None,
194 traceback: TracebackType | None,
195 ) -> None:
196 if self.show_output:
197 return
198
199 # The run reports what it held, and leaves nothing here. Something
200 # is left only when the run was left without reporting: an error in
201 # the runner itself, or code that wrote its reason and called
202 # `sys.exit()`. Nothing of the runner's is left to print it then, so
203 # it is printed here.
204 held = self.take()
205
206 sys.breakpointhook = self._previous_breakpoint_hook # ty: ignore[invalid-assignment]
207 if self._fault_handler_was_enabled:
208 faulthandler.enable()
209 else:
210 faulthandler.disable()
211
212 _flush_what_python_is_holding()
213 _write_a_line_at_a_time(sys.stdout, self._stdout_wrote_a_line_at_a_time)
214 self.real_stdout.flush()
215 self.real_stderr.flush()
216 for descriptor in self._held:
217 descriptor.close()
218 self._held = []
219 self.real_stdout = sys.stdout
220 self.real_stderr = sys.stderr
221
222 if held:
223 _print_what_nothing_reported(held)
224
225 def take(self) -> Output:
226 """What has been written since the last `take()` or `discard()`."""
227 if not self._held:
228 return NO_OUTPUT
229
230 _flush_what_python_is_holding()
231 stdout, stderr = self._held
232 output = Output(
233 stdout=stdout.written(full_output=self.full_output),
234 stderr=stderr.written(full_output=self.full_output),
235 )
236 self._empty()
237 return output
238
239 def discard(self) -> None:
240 """Throw away what has been written since the last `take()`."""
241 if not self._held:
242 return
243
244 _flush_what_python_is_holding()
245 self._empty()
246
247 def _empty(self) -> None:
248 for descriptor in self._held:
249 descriptor.empty()
250 # A debugger had the terminal until the test it was in was over.
251 if self._let_through_for_a_debugger:
252 self._let_through_for_a_debugger = False
253 for descriptor in self._held:
254 descriptor.hold()
255
256 def _breakpoint(self, *args: object, **kwargs: object) -> object:
257 """
258 What `breakpoint()` calls during a run. Output is let through from
259 here until the test is over, and then the debugger is started as
260 it would have been.
261 """
262 if os.environ.get("PYTHONBREAKPOINT") == "0":
263 return None
264
265 self._let_output_through_for_a_debugger()
266
267 if os.environ.get("PYTHONBREAKPOINT", "pdb.set_trace") != "pdb.set_trace":
268 return sys.__breakpointhook__(*args, **kwargs) # noqa: T100 — starting the debugger that was asked for
269
270 # Imported here: pdb takes longer to import than most runs take to
271 # collect, and nearly no run uses it.
272 import pdb # noqa: T100
273
274 # Started on the frame that called `breakpoint()`, as `pdb.set_trace()`
275 # starts it. Left to find the frame for itself, pdb would find this
276 # one.
277 debugger = pdb.Pdb(mode="inline", backend="monitoring", colorize=True)
278 debugger.set_trace(sys._getframe(1))
279 return None
280
281 def _let_output_through_for_a_debugger(self) -> None:
282 if self._let_through_for_a_debugger:
283 return
284 _flush_what_python_is_holding()
285 for descriptor in self._held:
286 descriptor.let_through()
287 self._let_through_for_a_debugger = True
288 self.real_stderr.write(
289 "\nbreakpoint(): output is let through until this test is over.\n"
290 )
291 self.real_stderr.flush()
292
293
294def _print_what_nothing_reported(held: Output) -> None:
295 """
296 To stderr, both streams: stdout is where `--json` writes its document,
297 and a run that ended this way has no document to write.
298 """
299 for name, stream in (("stdout", held.stdout), ("stderr", held.stderr)):
300 if not stream.text:
301 continue
302 sys.stderr.write(f"Written to {name} before the run ended:\n")
303 if stream.cut_characters:
304 sys.stderr.write(f"... {stream.cut_characters:,} characters before this\n")
305 sys.stderr.write(stream.text)
306 if not stream.text.endswith("\n"):
307 sys.stderr.write("\n")
308 sys.stderr.flush()
309
310
311def _file_that_appends() -> tuple[IO[bytes], str | None]:
312 """
313 A temporary file that every write is added to the end of, whoever
314 writes and wherever they think they are in it. Once the file has been
315 emptied, the end is the start again.
316
317 It is opened by its path for that. `tempfile.TemporaryFile(mode="a+b")`
318 opens the file first and is given the mode afterwards, which is too
319 late to make the descriptor an appending one.
320 """
321 descriptor, path = tempfile.mkstemp(prefix="plain-testing-output-")
322 os.close(descriptor)
323 file = open(path, "a+b", buffering=0) # noqa: SIM115 — it is held open for the run
324 # Nothing else needs its name, and without one the file is gone when it
325 # is closed. Windows won't take the name off a file that is open: there
326 # the path is returned too, to be removed after closing.
327 try:
328 os.unlink(path)
329 except PermissionError:
330 return file, path
331 return file, None
332
333
334def _text_stream(descriptor: int, *, like: TextIO) -> TextIO:
335 """A stream that writes text to a descriptor, a line at a time."""
336 return os.fdopen(
337 descriptor,
338 "w",
339 buffering=1,
340 encoding=getattr(like, "encoding", None) or "utf-8",
341 errors="backslashreplace",
342 closefd=False,
343 )
344
345
346def _line_buffering_of(stream: TextIO) -> bool:
347 return bool(getattr(stream, "line_buffering", False))
348
349
350def _write_a_line_at_a_time(stream: TextIO, line_buffering: bool) -> None:
351 # A test run inside another test's `redirect_stdout` has a stream here
352 # that can't be told how to write, and doesn't write to the descriptor.
353 reconfigure = getattr(stream, "reconfigure", None)
354 if reconfigure is not None:
355 reconfigure(line_buffering=line_buffering)
356
357
358def _flush_what_python_is_holding() -> None:
359 """
360 Python keeps what is written to `sys.stdout` for a while before writing
361 it to the descriptor. It has to be written before the descriptor is
362 read, emptied or pointed somewhere else.
363 """
364 for stream in (sys.stdout, sys.stderr, sys.__stdout__, sys.__stderr__):
365 if stream is None:
366 continue
367 try:
368 stream.flush()
369 except ValueError, OSError:
370 # Closed, by the test that replaced it.
371 pass