capture.py 36 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144
  1. # mypy: allow-untyped-defs
  2. """Per-test stdout/stderr capturing mechanism."""
  3. from __future__ import annotations
  4. import abc
  5. import collections
  6. from collections.abc import Generator
  7. from collections.abc import Iterable
  8. from collections.abc import Iterator
  9. import contextlib
  10. import io
  11. from io import UnsupportedOperation
  12. import os
  13. import sys
  14. from tempfile import TemporaryFile
  15. from types import TracebackType
  16. from typing import Any
  17. from typing import AnyStr
  18. from typing import BinaryIO
  19. from typing import cast
  20. from typing import Final
  21. from typing import final
  22. from typing import Generic
  23. from typing import Literal
  24. from typing import NamedTuple
  25. from typing import TextIO
  26. from typing import TYPE_CHECKING
  27. if TYPE_CHECKING:
  28. from typing_extensions import Self
  29. from _pytest.config import Config
  30. from _pytest.config import hookimpl
  31. from _pytest.config.argparsing import Parser
  32. from _pytest.deprecated import check_ispytest
  33. from _pytest.fixtures import fixture
  34. from _pytest.fixtures import SubRequest
  35. from _pytest.nodes import Collector
  36. from _pytest.nodes import File
  37. from _pytest.nodes import Item
  38. from _pytest.reports import CollectReport
  39. _CaptureMethod = Literal["fd", "sys", "no", "tee-sys"]
  40. def pytest_addoption(parser: Parser) -> None:
  41. group = parser.getgroup("general")
  42. group.addoption(
  43. "--capture",
  44. action="store",
  45. default="fd",
  46. metavar="method",
  47. choices=["fd", "sys", "no", "tee-sys"],
  48. help="Per-test capturing method: one of fd|sys|no|tee-sys",
  49. )
  50. group._addoption( # private to use reserved lower-case short option
  51. "-s",
  52. action="store_const",
  53. const="no",
  54. dest="capture",
  55. help="Shortcut for --capture=no",
  56. )
  57. def _colorama_workaround() -> None:
  58. """Ensure colorama is imported so that it attaches to the correct stdio
  59. handles on Windows.
  60. colorama uses the terminal on import time. So if something does the
  61. first import of colorama while I/O capture is active, colorama will
  62. fail in various ways.
  63. """
  64. if sys.platform.startswith("win32"):
  65. try:
  66. import colorama # noqa: F401
  67. except ImportError:
  68. pass
  69. def _readline_workaround() -> None:
  70. """Ensure readline is imported early so it attaches to the correct stdio handles.
  71. This isn't a problem with the default GNU readline implementation, but in
  72. some configurations, Python uses libedit instead (on macOS, and for prebuilt
  73. binaries such as used by uv).
  74. In theory this is only needed if readline.backend == "libedit", but the
  75. workaround consists of importing readline here, so we already worked around
  76. the issue by the time we could check if we need to.
  77. """
  78. try:
  79. import readline # noqa: F401
  80. except ImportError:
  81. pass
  82. def _windowsconsoleio_workaround(stream: TextIO) -> None:
  83. """Workaround for Windows Unicode console handling.
  84. Python 3.6 implemented Unicode console handling for Windows. This works
  85. by reading/writing to the raw console handle using
  86. ``{Read,Write}ConsoleW``.
  87. The problem is that we are going to ``dup2`` over the stdio file
  88. descriptors when doing ``FDCapture`` and this will ``CloseHandle`` the
  89. handles used by Python to write to the console. Though there is still some
  90. weirdness and the console handle seems to only be closed randomly and not
  91. on the first call to ``CloseHandle``, or maybe it gets reopened with the
  92. same handle value when we suspend capturing.
  93. The workaround in this case will reopen stdio with a different fd which
  94. also means a different handle by replicating the logic in
  95. "Py_lifecycle.c:initstdio/create_stdio".
  96. :param stream:
  97. In practice ``sys.stdout`` or ``sys.stderr``, but given
  98. here as parameter for unittesting purposes.
  99. See https://github.com/pytest-dev/py/issues/103.
  100. """
  101. if not sys.platform.startswith("win32") or hasattr(sys, "pypy_version_info"):
  102. return
  103. # Bail out if ``stream`` doesn't seem like a proper ``io`` stream (#2666).
  104. if not hasattr(stream, "buffer"): # type: ignore[unreachable,unused-ignore]
  105. return
  106. raw_stdout = stream.buffer.raw if hasattr(stream.buffer, "raw") else stream.buffer
  107. if not isinstance(raw_stdout, io._WindowsConsoleIO): # type: ignore[attr-defined,unused-ignore]
  108. return
  109. def _reopen_stdio(f, mode):
  110. if not hasattr(stream.buffer, "raw") and mode[0] == "w":
  111. buffering = 0
  112. else:
  113. buffering = -1
  114. return io.TextIOWrapper(
  115. open(os.dup(f.fileno()), mode, buffering),
  116. f.encoding,
  117. f.errors,
  118. f.newlines,
  119. f.line_buffering,
  120. )
  121. sys.stdin = _reopen_stdio(sys.stdin, "rb")
  122. sys.stdout = _reopen_stdio(sys.stdout, "wb")
  123. sys.stderr = _reopen_stdio(sys.stderr, "wb")
  124. @hookimpl(wrapper=True)
  125. def pytest_load_initial_conftests(early_config: Config) -> Generator[None]:
  126. ns = early_config.known_args_namespace
  127. if ns.capture == "fd":
  128. _windowsconsoleio_workaround(sys.stdout)
  129. _colorama_workaround()
  130. _readline_workaround()
  131. pluginmanager = early_config.pluginmanager
  132. capman = CaptureManager(ns.capture)
  133. pluginmanager.register(capman, "capturemanager")
  134. # Make sure that capturemanager is properly reset at final shutdown.
  135. early_config.add_cleanup(capman.stop_global_capturing)
  136. # Finally trigger conftest loading but while capturing (issue #93).
  137. capman.start_global_capturing()
  138. try:
  139. try:
  140. yield
  141. finally:
  142. capman.suspend_global_capture()
  143. except BaseException:
  144. out, err = capman.read_global_capture()
  145. sys.stdout.write(out)
  146. sys.stderr.write(err)
  147. raise
  148. # IO Helpers.
  149. class EncodedFile(io.TextIOWrapper):
  150. __slots__ = ()
  151. @property
  152. def name(self) -> str:
  153. # Ensure that file.name is a string. Workaround for a Python bug
  154. # fixed in >=3.7.4: https://bugs.python.org/issue36015
  155. return repr(self.buffer)
  156. @property
  157. def mode(self) -> str:
  158. # TextIOWrapper doesn't expose a mode, but at least some of our
  159. # tests check it.
  160. assert hasattr(self.buffer, "mode")
  161. return cast(str, self.buffer.mode.replace("b", ""))
  162. class CaptureIO(io.TextIOWrapper):
  163. def __init__(self) -> None:
  164. super().__init__(io.BytesIO(), encoding="UTF-8", newline="", write_through=True)
  165. def getvalue(self) -> str:
  166. assert isinstance(self.buffer, io.BytesIO)
  167. return self.buffer.getvalue().decode("UTF-8")
  168. class TeeCaptureIO(CaptureIO):
  169. def __init__(self, other: TextIO) -> None:
  170. self._other = other
  171. super().__init__()
  172. def write(self, s: str) -> int:
  173. super().write(s)
  174. return self._other.write(s)
  175. class DontReadFromInput(TextIO):
  176. @property
  177. def encoding(self) -> str:
  178. assert sys.__stdin__ is not None
  179. return sys.__stdin__.encoding
  180. def read(self, size: int = -1) -> str:
  181. raise OSError(
  182. "pytest: reading from stdin while output is captured! Consider using `-s`."
  183. )
  184. readline = read
  185. def __next__(self) -> str:
  186. return self.readline()
  187. def readlines(self, hint: int | None = -1) -> list[str]:
  188. raise OSError(
  189. "pytest: reading from stdin while output is captured! Consider using `-s`."
  190. )
  191. def __iter__(self) -> Iterator[str]:
  192. return self
  193. def fileno(self) -> int:
  194. raise UnsupportedOperation("redirected stdin is pseudofile, has no fileno()")
  195. def flush(self) -> None:
  196. raise UnsupportedOperation("redirected stdin is pseudofile, has no flush()")
  197. def isatty(self) -> bool:
  198. return False
  199. def close(self) -> None:
  200. pass
  201. def readable(self) -> bool:
  202. return False
  203. def seek(self, offset: int, whence: int = 0) -> int:
  204. raise UnsupportedOperation("redirected stdin is pseudofile, has no seek(int)")
  205. def seekable(self) -> bool:
  206. return False
  207. def tell(self) -> int:
  208. raise UnsupportedOperation("redirected stdin is pseudofile, has no tell()")
  209. def truncate(self, size: int | None = None) -> int:
  210. raise UnsupportedOperation("cannot truncate stdin")
  211. def write(self, data: str) -> int:
  212. raise UnsupportedOperation("cannot write to stdin")
  213. def writelines(self, lines: Iterable[str]) -> None:
  214. raise UnsupportedOperation("Cannot write to stdin")
  215. def writable(self) -> bool:
  216. return False
  217. def __enter__(self) -> Self:
  218. return self
  219. def __exit__(
  220. self,
  221. type: type[BaseException] | None,
  222. value: BaseException | None,
  223. traceback: TracebackType | None,
  224. ) -> None:
  225. pass
  226. @property
  227. def buffer(self) -> BinaryIO:
  228. # The str/bytes doesn't actually matter in this type, so OK to fake.
  229. return self # type: ignore[return-value]
  230. # Capture classes.
  231. class CaptureBase(abc.ABC, Generic[AnyStr]):
  232. EMPTY_BUFFER: AnyStr
  233. @abc.abstractmethod
  234. def __init__(self, fd: int) -> None:
  235. raise NotImplementedError()
  236. @abc.abstractmethod
  237. def start(self) -> None:
  238. raise NotImplementedError()
  239. @abc.abstractmethod
  240. def done(self) -> None:
  241. raise NotImplementedError()
  242. @abc.abstractmethod
  243. def suspend(self) -> None:
  244. raise NotImplementedError()
  245. @abc.abstractmethod
  246. def resume(self) -> None:
  247. raise NotImplementedError()
  248. @abc.abstractmethod
  249. def writeorg(self, data: AnyStr) -> None:
  250. raise NotImplementedError()
  251. @abc.abstractmethod
  252. def snap(self) -> AnyStr:
  253. raise NotImplementedError()
  254. patchsysdict = {0: "stdin", 1: "stdout", 2: "stderr"}
  255. class NoCapture(CaptureBase[str]):
  256. EMPTY_BUFFER = ""
  257. def __init__(self, fd: int) -> None:
  258. pass
  259. def start(self) -> None:
  260. pass
  261. def done(self) -> None:
  262. pass
  263. def suspend(self) -> None:
  264. pass
  265. def resume(self) -> None:
  266. pass
  267. def snap(self) -> str:
  268. return ""
  269. def writeorg(self, data: str) -> None:
  270. pass
  271. class SysCaptureBase(CaptureBase[AnyStr]):
  272. def __init__(
  273. self, fd: int, tmpfile: TextIO | None = None, *, tee: bool = False
  274. ) -> None:
  275. name = patchsysdict[fd]
  276. self._old: TextIO = getattr(sys, name)
  277. self.name = name
  278. if tmpfile is None:
  279. if name == "stdin":
  280. tmpfile = DontReadFromInput()
  281. else:
  282. tmpfile = CaptureIO() if not tee else TeeCaptureIO(self._old)
  283. self.tmpfile = tmpfile
  284. self._state = "initialized"
  285. def repr(self, class_name: str) -> str:
  286. return "<{} {} _old={} _state={!r} tmpfile={!r}>".format(
  287. class_name,
  288. self.name,
  289. (hasattr(self, "_old") and repr(self._old)) or "<UNSET>",
  290. self._state,
  291. self.tmpfile,
  292. )
  293. def __repr__(self) -> str:
  294. return "<{} {} _old={} _state={!r} tmpfile={!r}>".format(
  295. self.__class__.__name__,
  296. self.name,
  297. (hasattr(self, "_old") and repr(self._old)) or "<UNSET>",
  298. self._state,
  299. self.tmpfile,
  300. )
  301. def _assert_state(self, op: str, states: tuple[str, ...]) -> None:
  302. assert self._state in states, (
  303. "cannot {} in state {!r}: expected one of {}".format(
  304. op, self._state, ", ".join(states)
  305. )
  306. )
  307. def start(self) -> None:
  308. self._assert_state("start", ("initialized",))
  309. setattr(sys, self.name, self.tmpfile)
  310. self._state = "started"
  311. def done(self) -> None:
  312. self._assert_state("done", ("initialized", "started", "suspended", "done"))
  313. if self._state == "done":
  314. return
  315. setattr(sys, self.name, self._old)
  316. del self._old
  317. self.tmpfile.close()
  318. self._state = "done"
  319. def suspend(self) -> None:
  320. self._assert_state("suspend", ("started", "suspended"))
  321. setattr(sys, self.name, self._old)
  322. self._state = "suspended"
  323. def resume(self) -> None:
  324. self._assert_state("resume", ("started", "suspended"))
  325. if self._state == "started":
  326. return
  327. setattr(sys, self.name, self.tmpfile)
  328. self._state = "started"
  329. class SysCaptureBinary(SysCaptureBase[bytes]):
  330. EMPTY_BUFFER = b""
  331. def snap(self) -> bytes:
  332. self._assert_state("snap", ("started", "suspended"))
  333. self.tmpfile.seek(0)
  334. res = self.tmpfile.buffer.read()
  335. self.tmpfile.seek(0)
  336. self.tmpfile.truncate()
  337. return res
  338. def writeorg(self, data: bytes) -> None:
  339. self._assert_state("writeorg", ("started", "suspended"))
  340. self._old.flush()
  341. self._old.buffer.write(data)
  342. self._old.buffer.flush()
  343. class SysCapture(SysCaptureBase[str]):
  344. EMPTY_BUFFER = ""
  345. def snap(self) -> str:
  346. self._assert_state("snap", ("started", "suspended"))
  347. assert isinstance(self.tmpfile, CaptureIO)
  348. res = self.tmpfile.getvalue()
  349. self.tmpfile.seek(0)
  350. self.tmpfile.truncate()
  351. return res
  352. def writeorg(self, data: str) -> None:
  353. self._assert_state("writeorg", ("started", "suspended"))
  354. self._old.write(data)
  355. self._old.flush()
  356. class FDCaptureBase(CaptureBase[AnyStr]):
  357. def __init__(self, targetfd: int) -> None:
  358. self.targetfd = targetfd
  359. try:
  360. os.fstat(targetfd)
  361. except OSError:
  362. # FD capturing is conceptually simple -- create a temporary file,
  363. # redirect the FD to it, redirect back when done. But when the
  364. # target FD is invalid it throws a wrench into this lovely scheme.
  365. #
  366. # Tests themselves shouldn't care if the FD is valid, FD capturing
  367. # should work regardless of external circumstances. So falling back
  368. # to just sys capturing is not a good option.
  369. #
  370. # Further complications are the need to support suspend() and the
  371. # possibility of FD reuse (e.g. the tmpfile getting the very same
  372. # target FD). The following approach is robust, I believe.
  373. self.targetfd_invalid: int | None = os.open(os.devnull, os.O_RDWR)
  374. os.dup2(self.targetfd_invalid, targetfd)
  375. else:
  376. self.targetfd_invalid = None
  377. self.targetfd_save = os.dup(targetfd)
  378. if targetfd == 0:
  379. self.tmpfile = open(os.devnull, encoding="utf-8")
  380. self.syscapture: CaptureBase[str] = SysCapture(targetfd)
  381. else:
  382. self.tmpfile = EncodedFile(
  383. TemporaryFile(buffering=0),
  384. encoding="utf-8",
  385. errors="replace",
  386. newline="",
  387. write_through=True,
  388. )
  389. if targetfd in patchsysdict:
  390. self.syscapture = SysCapture(targetfd, self.tmpfile)
  391. else:
  392. self.syscapture = NoCapture(targetfd)
  393. self._state = "initialized"
  394. def __repr__(self) -> str:
  395. return (
  396. f"<{self.__class__.__name__} {self.targetfd} oldfd={self.targetfd_save} "
  397. f"_state={self._state!r} tmpfile={self.tmpfile!r}>"
  398. )
  399. def _assert_state(self, op: str, states: tuple[str, ...]) -> None:
  400. assert self._state in states, (
  401. "cannot {} in state {!r}: expected one of {}".format(
  402. op, self._state, ", ".join(states)
  403. )
  404. )
  405. def start(self) -> None:
  406. """Start capturing on targetfd using memorized tmpfile."""
  407. self._assert_state("start", ("initialized",))
  408. os.dup2(self.tmpfile.fileno(), self.targetfd)
  409. self.syscapture.start()
  410. self._state = "started"
  411. def done(self) -> None:
  412. """Stop capturing, restore streams, return original capture file,
  413. seeked to position zero."""
  414. self._assert_state("done", ("initialized", "started", "suspended", "done"))
  415. if self._state == "done":
  416. return
  417. os.dup2(self.targetfd_save, self.targetfd)
  418. os.close(self.targetfd_save)
  419. if self.targetfd_invalid is not None:
  420. if self.targetfd_invalid != self.targetfd:
  421. os.close(self.targetfd)
  422. os.close(self.targetfd_invalid)
  423. self.syscapture.done()
  424. self.tmpfile.close()
  425. self._state = "done"
  426. def suspend(self) -> None:
  427. self._assert_state("suspend", ("started", "suspended"))
  428. if self._state == "suspended":
  429. return
  430. self.syscapture.suspend()
  431. os.dup2(self.targetfd_save, self.targetfd)
  432. self._state = "suspended"
  433. def resume(self) -> None:
  434. self._assert_state("resume", ("started", "suspended"))
  435. if self._state == "started":
  436. return
  437. self.syscapture.resume()
  438. os.dup2(self.tmpfile.fileno(), self.targetfd)
  439. self._state = "started"
  440. class FDCaptureBinary(FDCaptureBase[bytes]):
  441. """Capture IO to/from a given OS-level file descriptor.
  442. snap() produces `bytes`.
  443. """
  444. EMPTY_BUFFER = b""
  445. def snap(self) -> bytes:
  446. self._assert_state("snap", ("started", "suspended"))
  447. self.tmpfile.seek(0)
  448. res = self.tmpfile.buffer.read()
  449. self.tmpfile.seek(0)
  450. self.tmpfile.truncate()
  451. return res # type: ignore[return-value]
  452. def writeorg(self, data: bytes) -> None:
  453. """Write to original file descriptor."""
  454. self._assert_state("writeorg", ("started", "suspended"))
  455. os.write(self.targetfd_save, data)
  456. class FDCapture(FDCaptureBase[str]):
  457. """Capture IO to/from a given OS-level file descriptor.
  458. snap() produces text.
  459. """
  460. EMPTY_BUFFER = ""
  461. def snap(self) -> str:
  462. self._assert_state("snap", ("started", "suspended"))
  463. self.tmpfile.seek(0)
  464. res = self.tmpfile.read()
  465. self.tmpfile.seek(0)
  466. self.tmpfile.truncate()
  467. return res
  468. def writeorg(self, data: str) -> None:
  469. """Write to original file descriptor."""
  470. self._assert_state("writeorg", ("started", "suspended"))
  471. # XXX use encoding of original stream
  472. os.write(self.targetfd_save, data.encode("utf-8"))
  473. # MultiCapture
  474. # Generic NamedTuple only supported since Python 3.11.
  475. if sys.version_info >= (3, 11) or TYPE_CHECKING:
  476. @final
  477. class CaptureResult(NamedTuple, Generic[AnyStr]):
  478. """The result of :method:`caplog.readouterr() <pytest.CaptureFixture.readouterr>`."""
  479. out: AnyStr
  480. err: AnyStr
  481. else:
  482. class CaptureResult(
  483. collections.namedtuple("CaptureResult", ["out", "err"]), # noqa: PYI024
  484. Generic[AnyStr],
  485. ):
  486. """The result of :method:`caplog.readouterr() <pytest.CaptureFixture.readouterr>`."""
  487. __slots__ = ()
  488. class MultiCapture(Generic[AnyStr]):
  489. _state = None
  490. _in_suspended = False
  491. def __init__(
  492. self,
  493. in_: CaptureBase[AnyStr] | None,
  494. out: CaptureBase[AnyStr] | None,
  495. err: CaptureBase[AnyStr] | None,
  496. ) -> None:
  497. self.in_: CaptureBase[AnyStr] | None = in_
  498. self.out: CaptureBase[AnyStr] | None = out
  499. self.err: CaptureBase[AnyStr] | None = err
  500. def __repr__(self) -> str:
  501. return (
  502. f"<MultiCapture out={self.out!r} err={self.err!r} in_={self.in_!r} "
  503. f"_state={self._state!r} _in_suspended={self._in_suspended!r}>"
  504. )
  505. def start_capturing(self) -> None:
  506. self._state = "started"
  507. if self.in_:
  508. self.in_.start()
  509. if self.out:
  510. self.out.start()
  511. if self.err:
  512. self.err.start()
  513. def pop_outerr_to_orig(self) -> tuple[AnyStr, AnyStr]:
  514. """Pop current snapshot out/err capture and flush to orig streams."""
  515. out, err = self.readouterr()
  516. if out:
  517. assert self.out is not None
  518. self.out.writeorg(out)
  519. if err:
  520. assert self.err is not None
  521. self.err.writeorg(err)
  522. return out, err
  523. def suspend_capturing(self, in_: bool = False) -> None:
  524. self._state = "suspended"
  525. if self.out:
  526. self.out.suspend()
  527. if self.err:
  528. self.err.suspend()
  529. if in_ and self.in_:
  530. self.in_.suspend()
  531. self._in_suspended = True
  532. def resume_capturing(self) -> None:
  533. self._state = "started"
  534. if self.out:
  535. self.out.resume()
  536. if self.err:
  537. self.err.resume()
  538. if self._in_suspended:
  539. assert self.in_ is not None
  540. self.in_.resume()
  541. self._in_suspended = False
  542. def stop_capturing(self) -> None:
  543. """Stop capturing and reset capturing streams."""
  544. if self._state == "stopped":
  545. raise ValueError("was already stopped")
  546. self._state = "stopped"
  547. if self.out:
  548. self.out.done()
  549. if self.err:
  550. self.err.done()
  551. if self.in_:
  552. self.in_.done()
  553. def is_started(self) -> bool:
  554. """Whether actively capturing -- not suspended or stopped."""
  555. return self._state == "started"
  556. def readouterr(self) -> CaptureResult[AnyStr]:
  557. out = self.out.snap() if self.out else ""
  558. err = self.err.snap() if self.err else ""
  559. # TODO: This type error is real, need to fix.
  560. return CaptureResult(out, err) # type: ignore[arg-type]
  561. def _get_multicapture(method: _CaptureMethod) -> MultiCapture[str]:
  562. if method == "fd":
  563. return MultiCapture(in_=FDCapture(0), out=FDCapture(1), err=FDCapture(2))
  564. elif method == "sys":
  565. return MultiCapture(in_=SysCapture(0), out=SysCapture(1), err=SysCapture(2))
  566. elif method == "no":
  567. return MultiCapture(in_=None, out=None, err=None)
  568. elif method == "tee-sys":
  569. return MultiCapture(
  570. in_=None, out=SysCapture(1, tee=True), err=SysCapture(2, tee=True)
  571. )
  572. raise ValueError(f"unknown capturing method: {method!r}")
  573. # CaptureManager and CaptureFixture
  574. class CaptureManager:
  575. """The capture plugin.
  576. Manages that the appropriate capture method is enabled/disabled during
  577. collection and each test phase (setup, call, teardown). After each of
  578. those points, the captured output is obtained and attached to the
  579. collection/runtest report.
  580. There are two levels of capture:
  581. * global: enabled by default and can be suppressed by the ``-s``
  582. option. This is always enabled/disabled during collection and each test
  583. phase.
  584. * fixture: when a test function or one of its fixture depend on the
  585. ``capsys`` or ``capfd`` fixtures. In this case special handling is
  586. needed to ensure the fixtures take precedence over the global capture.
  587. """
  588. def __init__(self, method: _CaptureMethod) -> None:
  589. self._method: Final = method
  590. self._global_capturing: MultiCapture[str] | None = None
  591. self._capture_fixture: CaptureFixture[Any] | None = None
  592. def __repr__(self) -> str:
  593. return (
  594. f"<CaptureManager _method={self._method!r} _global_capturing={self._global_capturing!r} "
  595. f"_capture_fixture={self._capture_fixture!r}>"
  596. )
  597. def is_capturing(self) -> str | bool:
  598. if self.is_globally_capturing():
  599. return "global"
  600. if self._capture_fixture:
  601. return f"fixture {self._capture_fixture.request.fixturename}"
  602. return False
  603. # Global capturing control
  604. def is_globally_capturing(self) -> bool:
  605. return self._method != "no"
  606. def start_global_capturing(self) -> None:
  607. assert self._global_capturing is None
  608. self._global_capturing = _get_multicapture(self._method)
  609. self._global_capturing.start_capturing()
  610. def stop_global_capturing(self) -> None:
  611. if self._global_capturing is not None:
  612. self._global_capturing.pop_outerr_to_orig()
  613. self._global_capturing.stop_capturing()
  614. self._global_capturing = None
  615. def resume_global_capture(self) -> None:
  616. # During teardown of the python process, and on rare occasions, capture
  617. # attributes can be `None` while trying to resume global capture.
  618. if self._global_capturing is not None:
  619. self._global_capturing.resume_capturing()
  620. def suspend_global_capture(self, in_: bool = False) -> None:
  621. if self._global_capturing is not None:
  622. self._global_capturing.suspend_capturing(in_=in_)
  623. def suspend(self, in_: bool = False) -> None:
  624. # Need to undo local capsys-et-al if it exists before disabling global capture.
  625. self.suspend_fixture()
  626. self.suspend_global_capture(in_)
  627. def resume(self) -> None:
  628. self.resume_global_capture()
  629. self.resume_fixture()
  630. def read_global_capture(self) -> CaptureResult[str]:
  631. assert self._global_capturing is not None
  632. return self._global_capturing.readouterr()
  633. # Fixture Control
  634. def set_fixture(self, capture_fixture: CaptureFixture[Any]) -> None:
  635. if self._capture_fixture:
  636. current_fixture = self._capture_fixture.request.fixturename
  637. requested_fixture = capture_fixture.request.fixturename
  638. capture_fixture.request.raiseerror(
  639. f"cannot use {requested_fixture} and {current_fixture} at the same time"
  640. )
  641. self._capture_fixture = capture_fixture
  642. def unset_fixture(self) -> None:
  643. self._capture_fixture = None
  644. def activate_fixture(self) -> None:
  645. """If the current item is using ``capsys`` or ``capfd``, activate
  646. them so they take precedence over the global capture."""
  647. if self._capture_fixture:
  648. self._capture_fixture._start()
  649. def deactivate_fixture(self) -> None:
  650. """Deactivate the ``capsys`` or ``capfd`` fixture of this item, if any."""
  651. if self._capture_fixture:
  652. self._capture_fixture.close()
  653. def suspend_fixture(self) -> None:
  654. if self._capture_fixture:
  655. self._capture_fixture._suspend()
  656. def resume_fixture(self) -> None:
  657. if self._capture_fixture:
  658. self._capture_fixture._resume()
  659. # Helper context managers
  660. @contextlib.contextmanager
  661. def global_and_fixture_disabled(self) -> Generator[None]:
  662. """Context manager to temporarily disable global and current fixture capturing."""
  663. do_fixture = self._capture_fixture and self._capture_fixture._is_started()
  664. if do_fixture:
  665. self.suspend_fixture()
  666. do_global = self._global_capturing and self._global_capturing.is_started()
  667. if do_global:
  668. self.suspend_global_capture()
  669. try:
  670. yield
  671. finally:
  672. if do_global:
  673. self.resume_global_capture()
  674. if do_fixture:
  675. self.resume_fixture()
  676. @contextlib.contextmanager
  677. def item_capture(self, when: str, item: Item) -> Generator[None]:
  678. self.resume_global_capture()
  679. self.activate_fixture()
  680. try:
  681. yield
  682. finally:
  683. self.deactivate_fixture()
  684. self.suspend_global_capture(in_=False)
  685. out, err = self.read_global_capture()
  686. item.add_report_section(when, "stdout", out)
  687. item.add_report_section(when, "stderr", err)
  688. # Hooks
  689. @hookimpl(wrapper=True)
  690. def pytest_make_collect_report(
  691. self, collector: Collector
  692. ) -> Generator[None, CollectReport, CollectReport]:
  693. if isinstance(collector, File):
  694. self.resume_global_capture()
  695. try:
  696. rep = yield
  697. finally:
  698. self.suspend_global_capture()
  699. out, err = self.read_global_capture()
  700. if out:
  701. rep.sections.append(("Captured stdout", out))
  702. if err:
  703. rep.sections.append(("Captured stderr", err))
  704. else:
  705. rep = yield
  706. return rep
  707. @hookimpl(wrapper=True)
  708. def pytest_runtest_setup(self, item: Item) -> Generator[None]:
  709. with self.item_capture("setup", item):
  710. return (yield)
  711. @hookimpl(wrapper=True)
  712. def pytest_runtest_call(self, item: Item) -> Generator[None]:
  713. with self.item_capture("call", item):
  714. return (yield)
  715. @hookimpl(wrapper=True)
  716. def pytest_runtest_teardown(self, item: Item) -> Generator[None]:
  717. with self.item_capture("teardown", item):
  718. return (yield)
  719. @hookimpl(tryfirst=True)
  720. def pytest_keyboard_interrupt(self) -> None:
  721. self.stop_global_capturing()
  722. @hookimpl(tryfirst=True)
  723. def pytest_internalerror(self) -> None:
  724. self.stop_global_capturing()
  725. class CaptureFixture(Generic[AnyStr]):
  726. """Object returned by the :fixture:`capsys`, :fixture:`capsysbinary`,
  727. :fixture:`capfd` and :fixture:`capfdbinary` fixtures."""
  728. def __init__(
  729. self,
  730. captureclass: type[CaptureBase[AnyStr]],
  731. request: SubRequest,
  732. *,
  733. config: dict[str, Any] | None = None,
  734. _ispytest: bool = False,
  735. ) -> None:
  736. check_ispytest(_ispytest)
  737. self.captureclass: type[CaptureBase[AnyStr]] = captureclass
  738. self.request = request
  739. self._config = config if config else {}
  740. self._capture: MultiCapture[AnyStr] | None = None
  741. self._captured_out: AnyStr = self.captureclass.EMPTY_BUFFER
  742. self._captured_err: AnyStr = self.captureclass.EMPTY_BUFFER
  743. def _start(self) -> None:
  744. if self._capture is None:
  745. self._capture = MultiCapture(
  746. in_=None,
  747. out=self.captureclass(1, **self._config),
  748. err=self.captureclass(2, **self._config),
  749. )
  750. self._capture.start_capturing()
  751. def close(self) -> None:
  752. if self._capture is not None:
  753. out, err = self._capture.pop_outerr_to_orig()
  754. self._captured_out += out
  755. self._captured_err += err
  756. self._capture.stop_capturing()
  757. self._capture = None
  758. def readouterr(self) -> CaptureResult[AnyStr]:
  759. """Read and return the captured output so far, resetting the internal
  760. buffer.
  761. :returns:
  762. The captured content as a namedtuple with ``out`` and ``err``
  763. string attributes.
  764. """
  765. captured_out, captured_err = self._captured_out, self._captured_err
  766. if self._capture is not None:
  767. out, err = self._capture.readouterr()
  768. captured_out += out
  769. captured_err += err
  770. self._captured_out = self.captureclass.EMPTY_BUFFER
  771. self._captured_err = self.captureclass.EMPTY_BUFFER
  772. return CaptureResult(captured_out, captured_err)
  773. def _suspend(self) -> None:
  774. """Suspend this fixture's own capturing temporarily."""
  775. if self._capture is not None:
  776. self._capture.suspend_capturing()
  777. def _resume(self) -> None:
  778. """Resume this fixture's own capturing temporarily."""
  779. if self._capture is not None:
  780. self._capture.resume_capturing()
  781. def _is_started(self) -> bool:
  782. """Whether actively capturing -- not disabled or closed."""
  783. if self._capture is not None:
  784. return self._capture.is_started()
  785. return False
  786. @contextlib.contextmanager
  787. def disabled(self) -> Generator[None]:
  788. """Temporarily disable capturing while inside the ``with`` block."""
  789. capmanager: CaptureManager = self.request.config.pluginmanager.getplugin(
  790. "capturemanager"
  791. )
  792. with capmanager.global_and_fixture_disabled():
  793. yield
  794. # The fixtures.
  795. @fixture
  796. def capsys(request: SubRequest) -> Generator[CaptureFixture[str]]:
  797. r"""Enable text capturing of writes to ``sys.stdout`` and ``sys.stderr``.
  798. The captured output is made available via ``capsys.readouterr()`` method
  799. calls, which return a ``(out, err)`` namedtuple.
  800. ``out`` and ``err`` will be ``text`` objects.
  801. Returns an instance of :class:`CaptureFixture[str] <pytest.CaptureFixture>`.
  802. Example:
  803. .. code-block:: python
  804. def test_output(capsys):
  805. print("hello")
  806. captured = capsys.readouterr()
  807. assert captured.out == "hello\n"
  808. """
  809. capman: CaptureManager = request.config.pluginmanager.getplugin("capturemanager")
  810. capture_fixture = CaptureFixture(SysCapture, request, _ispytest=True)
  811. capman.set_fixture(capture_fixture)
  812. capture_fixture._start()
  813. yield capture_fixture
  814. capture_fixture.close()
  815. capman.unset_fixture()
  816. @fixture
  817. def capteesys(request: SubRequest) -> Generator[CaptureFixture[str]]:
  818. r"""Enable simultaneous text capturing and pass-through of writes
  819. to ``sys.stdout`` and ``sys.stderr`` as defined by ``--capture=``.
  820. The captured output is made available via ``capteesys.readouterr()`` method
  821. calls, which return a ``(out, err)`` namedtuple.
  822. ``out`` and ``err`` will be ``text`` objects.
  823. The output is also passed-through, allowing it to be "live-printed",
  824. reported, or both as defined by ``--capture=``.
  825. Returns an instance of :class:`CaptureFixture[str] <pytest.CaptureFixture>`.
  826. Example:
  827. .. code-block:: python
  828. def test_output(capteesys):
  829. print("hello")
  830. captured = capteesys.readouterr()
  831. assert captured.out == "hello\n"
  832. """
  833. capman: CaptureManager = request.config.pluginmanager.getplugin("capturemanager")
  834. capture_fixture = CaptureFixture(
  835. SysCapture, request, config=dict(tee=True), _ispytest=True
  836. )
  837. capman.set_fixture(capture_fixture)
  838. capture_fixture._start()
  839. yield capture_fixture
  840. capture_fixture.close()
  841. capman.unset_fixture()
  842. @fixture
  843. def capsysbinary(request: SubRequest) -> Generator[CaptureFixture[bytes]]:
  844. r"""Enable bytes capturing of writes to ``sys.stdout`` and ``sys.stderr``.
  845. The captured output is made available via ``capsysbinary.readouterr()``
  846. method calls, which return a ``(out, err)`` namedtuple.
  847. ``out`` and ``err`` will be ``bytes`` objects.
  848. Returns an instance of :class:`CaptureFixture[bytes] <pytest.CaptureFixture>`.
  849. Example:
  850. .. code-block:: python
  851. def test_output(capsysbinary):
  852. print("hello")
  853. captured = capsysbinary.readouterr()
  854. assert captured.out == b"hello\n"
  855. """
  856. capman: CaptureManager = request.config.pluginmanager.getplugin("capturemanager")
  857. capture_fixture = CaptureFixture(SysCaptureBinary, request, _ispytest=True)
  858. capman.set_fixture(capture_fixture)
  859. capture_fixture._start()
  860. yield capture_fixture
  861. capture_fixture.close()
  862. capman.unset_fixture()
  863. @fixture
  864. def capfd(request: SubRequest) -> Generator[CaptureFixture[str]]:
  865. r"""Enable text capturing of writes to file descriptors ``1`` and ``2``.
  866. The captured output is made available via ``capfd.readouterr()`` method
  867. calls, which return a ``(out, err)`` namedtuple.
  868. ``out`` and ``err`` will be ``text`` objects.
  869. Returns an instance of :class:`CaptureFixture[str] <pytest.CaptureFixture>`.
  870. Example:
  871. .. code-block:: python
  872. def test_system_echo(capfd):
  873. os.system('echo "hello"')
  874. captured = capfd.readouterr()
  875. assert captured.out == "hello\n"
  876. """
  877. capman: CaptureManager = request.config.pluginmanager.getplugin("capturemanager")
  878. capture_fixture = CaptureFixture(FDCapture, request, _ispytest=True)
  879. capman.set_fixture(capture_fixture)
  880. capture_fixture._start()
  881. yield capture_fixture
  882. capture_fixture.close()
  883. capman.unset_fixture()
  884. @fixture
  885. def capfdbinary(request: SubRequest) -> Generator[CaptureFixture[bytes]]:
  886. r"""Enable bytes capturing of writes to file descriptors ``1`` and ``2``.
  887. The captured output is made available via ``capfd.readouterr()`` method
  888. calls, which return a ``(out, err)`` namedtuple.
  889. ``out`` and ``err`` will be ``byte`` objects.
  890. Returns an instance of :class:`CaptureFixture[bytes] <pytest.CaptureFixture>`.
  891. Example:
  892. .. code-block:: python
  893. def test_system_echo(capfdbinary):
  894. os.system('echo "hello"')
  895. captured = capfdbinary.readouterr()
  896. assert captured.out == b"hello\n"
  897. """
  898. capman: CaptureManager = request.config.pluginmanager.getplugin("capturemanager")
  899. capture_fixture = CaptureFixture(FDCaptureBinary, request, _ispytest=True)
  900. capman.set_fixture(capture_fixture)
  901. capture_fixture._start()
  902. yield capture_fixture
  903. capture_fixture.close()
  904. capman.unset_fixture()