1082 lines
34 KiB
Python
1082 lines
34 KiB
Python
"""Per-test stdout/stderr capturing mechanism."""
|
|
import abc
|
|
import collections
|
|
import contextlib
|
|
import io
|
|
import os
|
|
import sys
|
|
from io import UnsupportedOperation
|
|
from tempfile import TemporaryFile
|
|
from types import TracebackType
|
|
from typing import Any
|
|
from typing import AnyStr
|
|
from typing import BinaryIO
|
|
from typing import Generator
|
|
from typing import Generic
|
|
from typing import Iterable
|
|
from typing import Iterator
|
|
from typing import List
|
|
from typing import NamedTuple
|
|
from typing import Optional
|
|
from typing import TextIO
|
|
from typing import Tuple
|
|
from typing import Type
|
|
from typing import TYPE_CHECKING
|
|
from typing import Union
|
|
|
|
from _pytest.compat import final
|
|
from _pytest.config import Config
|
|
from _pytest.config import hookimpl
|
|
from _pytest.config.argparsing import Parser
|
|
from _pytest.deprecated import check_ispytest
|
|
from _pytest.fixtures import fixture
|
|
from _pytest.fixtures import SubRequest
|
|
from _pytest.nodes import Collector
|
|
from _pytest.nodes import File
|
|
from _pytest.nodes import Item
|
|
|
|
if TYPE_CHECKING:
|
|
from typing_extensions import Final
|
|
from typing_extensions import Literal
|
|
|
|
_CaptureMethod = Literal["fd", "sys", "no", "tee-sys"]
|
|
|
|
|
|
def pytest_addoption(parser: Parser) -> None:
|
|
group = parser.getgroup("general")
|
|
group._addoption(
|
|
"--capture",
|
|
action="store",
|
|
default="fd",
|
|
metavar="method",
|
|
choices=["fd", "sys", "no", "tee-sys"],
|
|
help="Per-test capturing method: one of fd|sys|no|tee-sys",
|
|
)
|
|
group._addoption(
|
|
"-s",
|
|
action="store_const",
|
|
const="no",
|
|
dest="capture",
|
|
help="Shortcut for --capture=no",
|
|
)
|
|
|
|
|
|
def _colorama_workaround() -> None:
|
|
"""Ensure colorama is imported so that it attaches to the correct stdio
|
|
handles on Windows.
|
|
|
|
colorama uses the terminal on import time. So if something does the
|
|
first import of colorama while I/O capture is active, colorama will
|
|
fail in various ways.
|
|
"""
|
|
if sys.platform.startswith("win32"):
|
|
try:
|
|
import colorama # noqa: F401
|
|
except ImportError:
|
|
pass
|
|
|
|
|
|
def _windowsconsoleio_workaround(stream: TextIO) -> None:
|
|
"""Workaround for Windows Unicode console handling.
|
|
|
|
Python 3.6 implemented Unicode console handling for Windows. This works
|
|
by reading/writing to the raw console handle using
|
|
``{Read,Write}ConsoleW``.
|
|
|
|
The problem is that we are going to ``dup2`` over the stdio file
|
|
descriptors when doing ``FDCapture`` and this will ``CloseHandle`` the
|
|
handles used by Python to write to the console. Though there is still some
|
|
weirdness and the console handle seems to only be closed randomly and not
|
|
on the first call to ``CloseHandle``, or maybe it gets reopened with the
|
|
same handle value when we suspend capturing.
|
|
|
|
The workaround in this case will reopen stdio with a different fd which
|
|
also means a different handle by replicating the logic in
|
|
"Py_lifecycle.c:initstdio/create_stdio".
|
|
|
|
:param stream:
|
|
In practice ``sys.stdout`` or ``sys.stderr``, but given
|
|
here as parameter for unittesting purposes.
|
|
|
|
See https://github.com/pytest-dev/py/issues/103.
|
|
"""
|
|
if not sys.platform.startswith("win32") or hasattr(sys, "pypy_version_info"):
|
|
return
|
|
|
|
# Bail out if ``stream`` doesn't seem like a proper ``io`` stream (#2666).
|
|
if not hasattr(stream, "buffer"): # type: ignore[unreachable]
|
|
return
|
|
|
|
buffered = hasattr(stream.buffer, "raw")
|
|
raw_stdout = stream.buffer.raw if buffered else stream.buffer # type: ignore[attr-defined]
|
|
|
|
if not isinstance(raw_stdout, io._WindowsConsoleIO): # type: ignore[attr-defined]
|
|
return
|
|
|
|
def _reopen_stdio(f, mode):
|
|
if not buffered and mode[0] == "w":
|
|
buffering = 0
|
|
else:
|
|
buffering = -1
|
|
|
|
return io.TextIOWrapper(
|
|
open(os.dup(f.fileno()), mode, buffering),
|
|
f.encoding,
|
|
f.errors,
|
|
f.newlines,
|
|
f.line_buffering,
|
|
)
|
|
|
|
sys.stdin = _reopen_stdio(sys.stdin, "rb")
|
|
sys.stdout = _reopen_stdio(sys.stdout, "wb")
|
|
sys.stderr = _reopen_stdio(sys.stderr, "wb")
|
|
|
|
|
|
@hookimpl(hookwrapper=True)
|
|
def pytest_load_initial_conftests(early_config: Config):
|
|
ns = early_config.known_args_namespace
|
|
if ns.capture == "fd":
|
|
_windowsconsoleio_workaround(sys.stdout)
|
|
_colorama_workaround()
|
|
pluginmanager = early_config.pluginmanager
|
|
capman = CaptureManager(ns.capture)
|
|
pluginmanager.register(capman, "capturemanager")
|
|
|
|
# Make sure that capturemanager is properly reset at final shutdown.
|
|
early_config.add_cleanup(capman.stop_global_capturing)
|
|
|
|
# Finally trigger conftest loading but while capturing (issue #93).
|
|
capman.start_global_capturing()
|
|
outcome = yield
|
|
capman.suspend_global_capture()
|
|
if outcome.excinfo is not None:
|
|
out, err = capman.read_global_capture()
|
|
sys.stdout.write(out)
|
|
sys.stderr.write(err)
|
|
|
|
|
|
# IO Helpers.
|
|
|
|
|
|
class EncodedFile(io.TextIOWrapper):
|
|
__slots__ = ()
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
# Ensure that file.name is a string. Workaround for a Python bug
|
|
# fixed in >=3.7.4: https://bugs.python.org/issue36015
|
|
return repr(self.buffer)
|
|
|
|
@property
|
|
def mode(self) -> str:
|
|
# TextIOWrapper doesn't expose a mode, but at least some of our
|
|
# tests check it.
|
|
return self.buffer.mode.replace("b", "")
|
|
|
|
|
|
class CaptureIO(io.TextIOWrapper):
|
|
def __init__(self) -> None:
|
|
super().__init__(io.BytesIO(), encoding="UTF-8", newline="", write_through=True)
|
|
|
|
def getvalue(self) -> str:
|
|
assert isinstance(self.buffer, io.BytesIO)
|
|
return self.buffer.getvalue().decode("UTF-8")
|
|
|
|
|
|
class TeeCaptureIO(CaptureIO):
|
|
def __init__(self, other: TextIO) -> None:
|
|
self._other = other
|
|
super().__init__()
|
|
|
|
def write(self, s: str) -> int:
|
|
super().write(s)
|
|
return self._other.write(s)
|
|
|
|
|
|
class DontReadFromInput(TextIO):
|
|
@property
|
|
def encoding(self) -> str:
|
|
return sys.__stdin__.encoding
|
|
|
|
def read(self, size: int = -1) -> str:
|
|
raise OSError(
|
|
"pytest: reading from stdin while output is captured! Consider using `-s`."
|
|
)
|
|
|
|
readline = read
|
|
|
|
def __next__(self) -> str:
|
|
return self.readline()
|
|
|
|
def readlines(self, hint: Optional[int] = -1) -> List[str]:
|
|
raise OSError(
|
|
"pytest: reading from stdin while output is captured! Consider using `-s`."
|
|
)
|
|
|
|
def __iter__(self) -> Iterator[str]:
|
|
return self
|
|
|
|
def fileno(self) -> int:
|
|
raise UnsupportedOperation("redirected stdin is pseudofile, has no fileno()")
|
|
|
|
def flush(self) -> None:
|
|
raise UnsupportedOperation("redirected stdin is pseudofile, has no flush()")
|
|
|
|
def isatty(self) -> bool:
|
|
return False
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
def readable(self) -> bool:
|
|
return False
|
|
|
|
def seek(self, offset: int, whence: int = 0) -> int:
|
|
raise UnsupportedOperation("redirected stdin is pseudofile, has no seek(int)")
|
|
|
|
def seekable(self) -> bool:
|
|
return False
|
|
|
|
def tell(self) -> int:
|
|
raise UnsupportedOperation("redirected stdin is pseudofile, has no tell()")
|
|
|
|
def truncate(self, size: Optional[int] = None) -> int:
|
|
raise UnsupportedOperation("cannot truncate stdin")
|
|
|
|
def write(self, data: str) -> int:
|
|
raise UnsupportedOperation("cannot write to stdin")
|
|
|
|
def writelines(self, lines: Iterable[str]) -> None:
|
|
raise UnsupportedOperation("Cannot write to stdin")
|
|
|
|
def writable(self) -> bool:
|
|
return False
|
|
|
|
def __enter__(self) -> "DontReadFromInput":
|
|
return self
|
|
|
|
def __exit__(
|
|
self,
|
|
type: Optional[Type[BaseException]],
|
|
value: Optional[BaseException],
|
|
traceback: Optional[TracebackType],
|
|
) -> None:
|
|
pass
|
|
|
|
@property
|
|
def buffer(self) -> BinaryIO:
|
|
# The str/bytes doesn't actually matter in this type, so OK to fake.
|
|
return self # type: ignore[return-value]
|
|
|
|
|
|
# Capture classes.
|
|
|
|
|
|
class CaptureBase(abc.ABC, Generic[AnyStr]):
|
|
EMPTY_BUFFER: AnyStr
|
|
|
|
@abc.abstractmethod
|
|
def __init__(self, fd: int) -> None:
|
|
raise NotImplementedError()
|
|
|
|
@abc.abstractmethod
|
|
def start(self) -> None:
|
|
raise NotImplementedError()
|
|
|
|
@abc.abstractmethod
|
|
def done(self) -> None:
|
|
raise NotImplementedError()
|
|
|
|
@abc.abstractmethod
|
|
def suspend(self) -> None:
|
|
raise NotImplementedError()
|
|
|
|
@abc.abstractmethod
|
|
def resume(self) -> None:
|
|
raise NotImplementedError()
|
|
|
|
@abc.abstractmethod
|
|
def writeorg(self, data: AnyStr) -> None:
|
|
raise NotImplementedError()
|
|
|
|
@abc.abstractmethod
|
|
def snap(self) -> AnyStr:
|
|
raise NotImplementedError()
|
|
|
|
|
|
patchsysdict = {0: "stdin", 1: "stdout", 2: "stderr"}
|
|
|
|
|
|
class NoCapture(CaptureBase[str]):
|
|
EMPTY_BUFFER = ""
|
|
|
|
def __init__(self, fd: int) -> None:
|
|
pass
|
|
|
|
def start(self) -> None:
|
|
pass
|
|
|
|
def done(self) -> None:
|
|
pass
|
|
|
|
def suspend(self) -> None:
|
|
pass
|
|
|
|
def resume(self) -> None:
|
|
pass
|
|
|
|
def snap(self) -> str:
|
|
return ""
|
|
|
|
def writeorg(self, data: str) -> None:
|
|
pass
|
|
|
|
|
|
class SysCaptureBase(CaptureBase[AnyStr]):
|
|
def __init__(
|
|
self, fd: int, tmpfile: Optional[TextIO] = None, *, tee: bool = False
|
|
) -> None:
|
|
name = patchsysdict[fd]
|
|
self._old: TextIO = getattr(sys, name)
|
|
self.name = name
|
|
if tmpfile is None:
|
|
if name == "stdin":
|
|
tmpfile = DontReadFromInput()
|
|
else:
|
|
tmpfile = CaptureIO() if not tee else TeeCaptureIO(self._old)
|
|
self.tmpfile = tmpfile
|
|
self._state = "initialized"
|
|
|
|
def repr(self, class_name: str) -> str:
|
|
return "<{} {} _old={} _state={!r} tmpfile={!r}>".format(
|
|
class_name,
|
|
self.name,
|
|
hasattr(self, "_old") and repr(self._old) or "<UNSET>",
|
|
self._state,
|
|
self.tmpfile,
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return "<{} {} _old={} _state={!r} tmpfile={!r}>".format(
|
|
self.__class__.__name__,
|
|
self.name,
|
|
hasattr(self, "_old") and repr(self._old) or "<UNSET>",
|
|
self._state,
|
|
self.tmpfile,
|
|
)
|
|
|
|
def _assert_state(self, op: str, states: Tuple[str, ...]) -> None:
|
|
assert (
|
|
self._state in states
|
|
), "cannot {} in state {!r}: expected one of {}".format(
|
|
op, self._state, ", ".join(states)
|
|
)
|
|
|
|
def start(self) -> None:
|
|
self._assert_state("start", ("initialized",))
|
|
setattr(sys, self.name, self.tmpfile)
|
|
self._state = "started"
|
|
|
|
def done(self) -> None:
|
|
self._assert_state("done", ("initialized", "started", "suspended", "done"))
|
|
if self._state == "done":
|
|
return
|
|
setattr(sys, self.name, self._old)
|
|
del self._old
|
|
self.tmpfile.close()
|
|
self._state = "done"
|
|
|
|
def suspend(self) -> None:
|
|
self._assert_state("suspend", ("started", "suspended"))
|
|
setattr(sys, self.name, self._old)
|
|
self._state = "suspended"
|
|
|
|
def resume(self) -> None:
|
|
self._assert_state("resume", ("started", "suspended"))
|
|
if self._state == "started":
|
|
return
|
|
setattr(sys, self.name, self.tmpfile)
|
|
self._state = "started"
|
|
|
|
|
|
class SysCaptureBinary(SysCaptureBase[bytes]):
|
|
EMPTY_BUFFER = b""
|
|
|
|
def snap(self) -> bytes:
|
|
self._assert_state("snap", ("started", "suspended"))
|
|
self.tmpfile.seek(0)
|
|
res = self.tmpfile.buffer.read()
|
|
self.tmpfile.seek(0)
|
|
self.tmpfile.truncate()
|
|
return res
|
|
|
|
def writeorg(self, data: bytes) -> None:
|
|
self._assert_state("writeorg", ("started", "suspended"))
|
|
self._old.flush()
|
|
self._old.buffer.write(data)
|
|
self._old.buffer.flush()
|
|
|
|
|
|
class SysCapture(SysCaptureBase[str]):
|
|
EMPTY_BUFFER = ""
|
|
|
|
def snap(self) -> str:
|
|
self._assert_state("snap", ("started", "suspended"))
|
|
assert isinstance(self.tmpfile, CaptureIO)
|
|
res = self.tmpfile.getvalue()
|
|
self.tmpfile.seek(0)
|
|
self.tmpfile.truncate()
|
|
return res
|
|
|
|
def writeorg(self, data: str) -> None:
|
|
self._assert_state("writeorg", ("started", "suspended"))
|
|
self._old.write(data)
|
|
self._old.flush()
|
|
|
|
|
|
class FDCaptureBase(CaptureBase[AnyStr]):
|
|
def __init__(self, targetfd: int) -> None:
|
|
self.targetfd = targetfd
|
|
|
|
try:
|
|
os.fstat(targetfd)
|
|
except OSError:
|
|
# FD capturing is conceptually simple -- create a temporary file,
|
|
# redirect the FD to it, redirect back when done. But when the
|
|
# target FD is invalid it throws a wrench into this lovely scheme.
|
|
#
|
|
# Tests themselves shouldn't care if the FD is valid, FD capturing
|
|
# should work regardless of external circumstances. So falling back
|
|
# to just sys capturing is not a good option.
|
|
#
|
|
# Further complications are the need to support suspend() and the
|
|
# possibility of FD reuse (e.g. the tmpfile getting the very same
|
|
# target FD). The following approach is robust, I believe.
|
|
self.targetfd_invalid: Optional[int] = os.open(os.devnull, os.O_RDWR)
|
|
os.dup2(self.targetfd_invalid, targetfd)
|
|
else:
|
|
self.targetfd_invalid = None
|
|
self.targetfd_save = os.dup(targetfd)
|
|
|
|
if targetfd == 0:
|
|
self.tmpfile = open(os.devnull, encoding="utf-8")
|
|
self.syscapture: CaptureBase[str] = SysCapture(targetfd)
|
|
else:
|
|
self.tmpfile = EncodedFile(
|
|
TemporaryFile(buffering=0),
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
newline="",
|
|
write_through=True,
|
|
)
|
|
if targetfd in patchsysdict:
|
|
self.syscapture = SysCapture(targetfd, self.tmpfile)
|
|
else:
|
|
self.syscapture = NoCapture(targetfd)
|
|
|
|
self._state = "initialized"
|
|
|
|
def __repr__(self) -> str:
|
|
return "<{} {} oldfd={} _state={!r} tmpfile={!r}>".format(
|
|
self.__class__.__name__,
|
|
self.targetfd,
|
|
self.targetfd_save,
|
|
self._state,
|
|
self.tmpfile,
|
|
)
|
|
|
|
def _assert_state(self, op: str, states: Tuple[str, ...]) -> None:
|
|
assert (
|
|
self._state in states
|
|
), "cannot {} in state {!r}: expected one of {}".format(
|
|
op, self._state, ", ".join(states)
|
|
)
|
|
|
|
def start(self) -> None:
|
|
"""Start capturing on targetfd using memorized tmpfile."""
|
|
self._assert_state("start", ("initialized",))
|
|
os.dup2(self.tmpfile.fileno(), self.targetfd)
|
|
self.syscapture.start()
|
|
self._state = "started"
|
|
|
|
def done(self) -> None:
|
|
"""Stop capturing, restore streams, return original capture file,
|
|
seeked to position zero."""
|
|
self._assert_state("done", ("initialized", "started", "suspended", "done"))
|
|
if self._state == "done":
|
|
return
|
|
os.dup2(self.targetfd_save, self.targetfd)
|
|
os.close(self.targetfd_save)
|
|
if self.targetfd_invalid is not None:
|
|
if self.targetfd_invalid != self.targetfd:
|
|
os.close(self.targetfd)
|
|
os.close(self.targetfd_invalid)
|
|
self.syscapture.done()
|
|
self.tmpfile.close()
|
|
self._state = "done"
|
|
|
|
def suspend(self) -> None:
|
|
self._assert_state("suspend", ("started", "suspended"))
|
|
if self._state == "suspended":
|
|
return
|
|
self.syscapture.suspend()
|
|
os.dup2(self.targetfd_save, self.targetfd)
|
|
self._state = "suspended"
|
|
|
|
def resume(self) -> None:
|
|
self._assert_state("resume", ("started", "suspended"))
|
|
if self._state == "started":
|
|
return
|
|
self.syscapture.resume()
|
|
os.dup2(self.tmpfile.fileno(), self.targetfd)
|
|
self._state = "started"
|
|
|
|
|
|
class FDCaptureBinary(FDCaptureBase[bytes]):
|
|
"""Capture IO to/from a given OS-level file descriptor.
|
|
|
|
snap() produces `bytes`.
|
|
"""
|
|
|
|
EMPTY_BUFFER = b""
|
|
|
|
def snap(self) -> bytes:
|
|
self._assert_state("snap", ("started", "suspended"))
|
|
self.tmpfile.seek(0)
|
|
res = self.tmpfile.buffer.read()
|
|
self.tmpfile.seek(0)
|
|
self.tmpfile.truncate()
|
|
return res
|
|
|
|
def writeorg(self, data: bytes) -> None:
|
|
"""Write to original file descriptor."""
|
|
self._assert_state("writeorg", ("started", "suspended"))
|
|
os.write(self.targetfd_save, data)
|
|
|
|
|
|
class FDCapture(FDCaptureBase[str]):
|
|
"""Capture IO to/from a given OS-level file descriptor.
|
|
|
|
snap() produces text.
|
|
"""
|
|
|
|
EMPTY_BUFFER = ""
|
|
|
|
def snap(self) -> str:
|
|
self._assert_state("snap", ("started", "suspended"))
|
|
self.tmpfile.seek(0)
|
|
res = self.tmpfile.read()
|
|
self.tmpfile.seek(0)
|
|
self.tmpfile.truncate()
|
|
return res
|
|
|
|
def writeorg(self, data: str) -> None:
|
|
"""Write to original file descriptor."""
|
|
self._assert_state("writeorg", ("started", "suspended"))
|
|
# XXX use encoding of original stream
|
|
os.write(self.targetfd_save, data.encode("utf-8"))
|
|
|
|
|
|
# MultiCapture
|
|
|
|
|
|
# Generic NamedTuple only supported since Python 3.11.
|
|
if sys.version_info >= (3, 11) or TYPE_CHECKING:
|
|
|
|
@final
|
|
class CaptureResult(NamedTuple, Generic[AnyStr]):
|
|
"""The result of :method:`CaptureFixture.readouterr`."""
|
|
|
|
out: AnyStr
|
|
err: AnyStr
|
|
|
|
else:
|
|
|
|
class CaptureResult(
|
|
collections.namedtuple("CaptureResult", ["out", "err"]), Generic[AnyStr]
|
|
):
|
|
"""The result of :method:`CaptureFixture.readouterr`."""
|
|
|
|
__slots__ = ()
|
|
|
|
|
|
class MultiCapture(Generic[AnyStr]):
|
|
_state = None
|
|
_in_suspended = False
|
|
|
|
def __init__(
|
|
self,
|
|
in_: Optional[CaptureBase[AnyStr]],
|
|
out: Optional[CaptureBase[AnyStr]],
|
|
err: Optional[CaptureBase[AnyStr]],
|
|
) -> None:
|
|
self.in_: Optional[CaptureBase[AnyStr]] = in_
|
|
self.out: Optional[CaptureBase[AnyStr]] = out
|
|
self.err: Optional[CaptureBase[AnyStr]] = err
|
|
|
|
def __repr__(self) -> str:
|
|
return "<MultiCapture out={!r} err={!r} in_={!r} _state={!r} _in_suspended={!r}>".format(
|
|
self.out,
|
|
self.err,
|
|
self.in_,
|
|
self._state,
|
|
self._in_suspended,
|
|
)
|
|
|
|
def start_capturing(self) -> None:
|
|
self._state = "started"
|
|
if self.in_:
|
|
self.in_.start()
|
|
if self.out:
|
|
self.out.start()
|
|
if self.err:
|
|
self.err.start()
|
|
|
|
def pop_outerr_to_orig(self) -> Tuple[AnyStr, AnyStr]:
|
|
"""Pop current snapshot out/err capture and flush to orig streams."""
|
|
out, err = self.readouterr()
|
|
if out:
|
|
assert self.out is not None
|
|
self.out.writeorg(out)
|
|
if err:
|
|
assert self.err is not None
|
|
self.err.writeorg(err)
|
|
return out, err
|
|
|
|
def suspend_capturing(self, in_: bool = False) -> None:
|
|
self._state = "suspended"
|
|
if self.out:
|
|
self.out.suspend()
|
|
if self.err:
|
|
self.err.suspend()
|
|
if in_ and self.in_:
|
|
self.in_.suspend()
|
|
self._in_suspended = True
|
|
|
|
def resume_capturing(self) -> None:
|
|
self._state = "started"
|
|
if self.out:
|
|
self.out.resume()
|
|
if self.err:
|
|
self.err.resume()
|
|
if self._in_suspended:
|
|
assert self.in_ is not None
|
|
self.in_.resume()
|
|
self._in_suspended = False
|
|
|
|
def stop_capturing(self) -> None:
|
|
"""Stop capturing and reset capturing streams."""
|
|
if self._state == "stopped":
|
|
raise ValueError("was already stopped")
|
|
self._state = "stopped"
|
|
if self.out:
|
|
self.out.done()
|
|
if self.err:
|
|
self.err.done()
|
|
if self.in_:
|
|
self.in_.done()
|
|
|
|
def is_started(self) -> bool:
|
|
"""Whether actively capturing -- not suspended or stopped."""
|
|
return self._state == "started"
|
|
|
|
def readouterr(self) -> CaptureResult[AnyStr]:
|
|
out = self.out.snap() if self.out else ""
|
|
err = self.err.snap() if self.err else ""
|
|
# TODO: This type error is real, need to fix.
|
|
return CaptureResult(out, err) # type: ignore[arg-type]
|
|
|
|
|
|
def _get_multicapture(method: "_CaptureMethod") -> MultiCapture[str]:
|
|
if method == "fd":
|
|
return MultiCapture(in_=FDCapture(0), out=FDCapture(1), err=FDCapture(2))
|
|
elif method == "sys":
|
|
return MultiCapture(in_=SysCapture(0), out=SysCapture(1), err=SysCapture(2))
|
|
elif method == "no":
|
|
return MultiCapture(in_=None, out=None, err=None)
|
|
elif method == "tee-sys":
|
|
return MultiCapture(
|
|
in_=None, out=SysCapture(1, tee=True), err=SysCapture(2, tee=True)
|
|
)
|
|
raise ValueError(f"unknown capturing method: {method!r}")
|
|
|
|
|
|
# CaptureManager and CaptureFixture
|
|
|
|
|
|
class CaptureManager:
|
|
"""The capture plugin.
|
|
|
|
Manages that the appropriate capture method is enabled/disabled during
|
|
collection and each test phase (setup, call, teardown). After each of
|
|
those points, the captured output is obtained and attached to the
|
|
collection/runtest report.
|
|
|
|
There are two levels of capture:
|
|
|
|
* global: enabled by default and can be suppressed by the ``-s``
|
|
option. This is always enabled/disabled during collection and each test
|
|
phase.
|
|
|
|
* fixture: when a test function or one of its fixture depend on the
|
|
``capsys`` or ``capfd`` fixtures. In this case special handling is
|
|
needed to ensure the fixtures take precedence over the global capture.
|
|
"""
|
|
|
|
def __init__(self, method: "_CaptureMethod") -> None:
|
|
self._method: Final = method
|
|
self._global_capturing: Optional[MultiCapture[str]] = None
|
|
self._capture_fixture: Optional[CaptureFixture[Any]] = None
|
|
|
|
def __repr__(self) -> str:
|
|
return "<CaptureManager _method={!r} _global_capturing={!r} _capture_fixture={!r}>".format(
|
|
self._method, self._global_capturing, self._capture_fixture
|
|
)
|
|
|
|
def is_capturing(self) -> Union[str, bool]:
|
|
if self.is_globally_capturing():
|
|
return "global"
|
|
if self._capture_fixture:
|
|
return "fixture %s" % self._capture_fixture.request.fixturename
|
|
return False
|
|
|
|
# Global capturing control
|
|
|
|
def is_globally_capturing(self) -> bool:
|
|
return self._method != "no"
|
|
|
|
def start_global_capturing(self) -> None:
|
|
assert self._global_capturing is None
|
|
self._global_capturing = _get_multicapture(self._method)
|
|
self._global_capturing.start_capturing()
|
|
|
|
def stop_global_capturing(self) -> None:
|
|
if self._global_capturing is not None:
|
|
self._global_capturing.pop_outerr_to_orig()
|
|
self._global_capturing.stop_capturing()
|
|
self._global_capturing = None
|
|
|
|
def resume_global_capture(self) -> None:
|
|
# During teardown of the python process, and on rare occasions, capture
|
|
# attributes can be `None` while trying to resume global capture.
|
|
if self._global_capturing is not None:
|
|
self._global_capturing.resume_capturing()
|
|
|
|
def suspend_global_capture(self, in_: bool = False) -> None:
|
|
if self._global_capturing is not None:
|
|
self._global_capturing.suspend_capturing(in_=in_)
|
|
|
|
def suspend(self, in_: bool = False) -> None:
|
|
# Need to undo local capsys-et-al if it exists before disabling global capture.
|
|
self.suspend_fixture()
|
|
self.suspend_global_capture(in_)
|
|
|
|
def resume(self) -> None:
|
|
self.resume_global_capture()
|
|
self.resume_fixture()
|
|
|
|
def read_global_capture(self) -> CaptureResult[str]:
|
|
assert self._global_capturing is not None
|
|
return self._global_capturing.readouterr()
|
|
|
|
# Fixture Control
|
|
|
|
def set_fixture(self, capture_fixture: "CaptureFixture[Any]") -> None:
|
|
if self._capture_fixture:
|
|
current_fixture = self._capture_fixture.request.fixturename
|
|
requested_fixture = capture_fixture.request.fixturename
|
|
capture_fixture.request.raiseerror(
|
|
"cannot use {} and {} at the same time".format(
|
|
requested_fixture, current_fixture
|
|
)
|
|
)
|
|
self._capture_fixture = capture_fixture
|
|
|
|
def unset_fixture(self) -> None:
|
|
self._capture_fixture = None
|
|
|
|
def activate_fixture(self) -> None:
|
|
"""If the current item is using ``capsys`` or ``capfd``, activate
|
|
them so they take precedence over the global capture."""
|
|
if self._capture_fixture:
|
|
self._capture_fixture._start()
|
|
|
|
def deactivate_fixture(self) -> None:
|
|
"""Deactivate the ``capsys`` or ``capfd`` fixture of this item, if any."""
|
|
if self._capture_fixture:
|
|
self._capture_fixture.close()
|
|
|
|
def suspend_fixture(self) -> None:
|
|
if self._capture_fixture:
|
|
self._capture_fixture._suspend()
|
|
|
|
def resume_fixture(self) -> None:
|
|
if self._capture_fixture:
|
|
self._capture_fixture._resume()
|
|
|
|
# Helper context managers
|
|
|
|
@contextlib.contextmanager
|
|
def global_and_fixture_disabled(self) -> Generator[None, None, None]:
|
|
"""Context manager to temporarily disable global and current fixture capturing."""
|
|
do_fixture = self._capture_fixture and self._capture_fixture._is_started()
|
|
if do_fixture:
|
|
self.suspend_fixture()
|
|
do_global = self._global_capturing and self._global_capturing.is_started()
|
|
if do_global:
|
|
self.suspend_global_capture()
|
|
try:
|
|
yield
|
|
finally:
|
|
if do_global:
|
|
self.resume_global_capture()
|
|
if do_fixture:
|
|
self.resume_fixture()
|
|
|
|
@contextlib.contextmanager
|
|
def item_capture(self, when: str, item: Item) -> Generator[None, None, None]:
|
|
self.resume_global_capture()
|
|
self.activate_fixture()
|
|
try:
|
|
yield
|
|
finally:
|
|
self.deactivate_fixture()
|
|
self.suspend_global_capture(in_=False)
|
|
|
|
out, err = self.read_global_capture()
|
|
item.add_report_section(when, "stdout", out)
|
|
item.add_report_section(when, "stderr", err)
|
|
|
|
# Hooks
|
|
|
|
@hookimpl(hookwrapper=True)
|
|
def pytest_make_collect_report(self, collector: Collector):
|
|
if isinstance(collector, File):
|
|
self.resume_global_capture()
|
|
outcome = yield
|
|
self.suspend_global_capture()
|
|
out, err = self.read_global_capture()
|
|
rep = outcome.get_result()
|
|
if out:
|
|
rep.sections.append(("Captured stdout", out))
|
|
if err:
|
|
rep.sections.append(("Captured stderr", err))
|
|
else:
|
|
yield
|
|
|
|
@hookimpl(hookwrapper=True)
|
|
def pytest_runtest_setup(self, item: Item) -> Generator[None, None, None]:
|
|
with self.item_capture("setup", item):
|
|
yield
|
|
|
|
@hookimpl(hookwrapper=True)
|
|
def pytest_runtest_call(self, item: Item) -> Generator[None, None, None]:
|
|
with self.item_capture("call", item):
|
|
yield
|
|
|
|
@hookimpl(hookwrapper=True)
|
|
def pytest_runtest_teardown(self, item: Item) -> Generator[None, None, None]:
|
|
with self.item_capture("teardown", item):
|
|
yield
|
|
|
|
@hookimpl(tryfirst=True)
|
|
def pytest_keyboard_interrupt(self) -> None:
|
|
self.stop_global_capturing()
|
|
|
|
@hookimpl(tryfirst=True)
|
|
def pytest_internalerror(self) -> None:
|
|
self.stop_global_capturing()
|
|
|
|
|
|
class CaptureFixture(Generic[AnyStr]):
|
|
"""Object returned by the :fixture:`capsys`, :fixture:`capsysbinary`,
|
|
:fixture:`capfd` and :fixture:`capfdbinary` fixtures."""
|
|
|
|
def __init__(
|
|
self,
|
|
captureclass: Type[CaptureBase[AnyStr]],
|
|
request: SubRequest,
|
|
*,
|
|
_ispytest: bool = False,
|
|
) -> None:
|
|
check_ispytest(_ispytest)
|
|
self.captureclass: Type[CaptureBase[AnyStr]] = captureclass
|
|
self.request = request
|
|
self._capture: Optional[MultiCapture[AnyStr]] = None
|
|
self._captured_out: AnyStr = self.captureclass.EMPTY_BUFFER
|
|
self._captured_err: AnyStr = self.captureclass.EMPTY_BUFFER
|
|
|
|
def _start(self) -> None:
|
|
if self._capture is None:
|
|
self._capture = MultiCapture(
|
|
in_=None,
|
|
out=self.captureclass(1),
|
|
err=self.captureclass(2),
|
|
)
|
|
self._capture.start_capturing()
|
|
|
|
def close(self) -> None:
|
|
if self._capture is not None:
|
|
out, err = self._capture.pop_outerr_to_orig()
|
|
self._captured_out += out
|
|
self._captured_err += err
|
|
self._capture.stop_capturing()
|
|
self._capture = None
|
|
|
|
def readouterr(self) -> CaptureResult[AnyStr]:
|
|
"""Read and return the captured output so far, resetting the internal
|
|
buffer.
|
|
|
|
:returns:
|
|
The captured content as a namedtuple with ``out`` and ``err``
|
|
string attributes.
|
|
"""
|
|
captured_out, captured_err = self._captured_out, self._captured_err
|
|
if self._capture is not None:
|
|
out, err = self._capture.readouterr()
|
|
captured_out += out
|
|
captured_err += err
|
|
self._captured_out = self.captureclass.EMPTY_BUFFER
|
|
self._captured_err = self.captureclass.EMPTY_BUFFER
|
|
return CaptureResult(captured_out, captured_err)
|
|
|
|
def _suspend(self) -> None:
|
|
"""Suspend this fixture's own capturing temporarily."""
|
|
if self._capture is not None:
|
|
self._capture.suspend_capturing()
|
|
|
|
def _resume(self) -> None:
|
|
"""Resume this fixture's own capturing temporarily."""
|
|
if self._capture is not None:
|
|
self._capture.resume_capturing()
|
|
|
|
def _is_started(self) -> bool:
|
|
"""Whether actively capturing -- not disabled or closed."""
|
|
if self._capture is not None:
|
|
return self._capture.is_started()
|
|
return False
|
|
|
|
@contextlib.contextmanager
|
|
def disabled(self) -> Generator[None, None, None]:
|
|
"""Temporarily disable capturing while inside the ``with`` block."""
|
|
capmanager: CaptureManager = self.request.config.pluginmanager.getplugin(
|
|
"capturemanager"
|
|
)
|
|
with capmanager.global_and_fixture_disabled():
|
|
yield
|
|
|
|
|
|
# The fixtures.
|
|
|
|
|
|
@fixture
|
|
def capsys(request: SubRequest) -> Generator[CaptureFixture[str], None, None]:
|
|
r"""Enable text capturing of writes to ``sys.stdout`` and ``sys.stderr``.
|
|
|
|
The captured output is made available via ``capsys.readouterr()`` method
|
|
calls, which return a ``(out, err)`` namedtuple.
|
|
``out`` and ``err`` will be ``text`` objects.
|
|
|
|
Returns an instance of :class:`CaptureFixture[str] <pytest.CaptureFixture>`.
|
|
|
|
Example:
|
|
|
|
.. code-block:: python
|
|
|
|
def test_output(capsys):
|
|
print("hello")
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "hello\n"
|
|
"""
|
|
capman: CaptureManager = request.config.pluginmanager.getplugin("capturemanager")
|
|
capture_fixture = CaptureFixture(SysCapture, request, _ispytest=True)
|
|
capman.set_fixture(capture_fixture)
|
|
capture_fixture._start()
|
|
yield capture_fixture
|
|
capture_fixture.close()
|
|
capman.unset_fixture()
|
|
|
|
|
|
@fixture
|
|
def capsysbinary(request: SubRequest) -> Generator[CaptureFixture[bytes], None, None]:
|
|
r"""Enable bytes capturing of writes to ``sys.stdout`` and ``sys.stderr``.
|
|
|
|
The captured output is made available via ``capsysbinary.readouterr()``
|
|
method calls, which return a ``(out, err)`` namedtuple.
|
|
``out`` and ``err`` will be ``bytes`` objects.
|
|
|
|
Returns an instance of :class:`CaptureFixture[bytes] <pytest.CaptureFixture>`.
|
|
|
|
Example:
|
|
|
|
.. code-block:: python
|
|
|
|
def test_output(capsysbinary):
|
|
print("hello")
|
|
captured = capsysbinary.readouterr()
|
|
assert captured.out == b"hello\n"
|
|
"""
|
|
capman: CaptureManager = request.config.pluginmanager.getplugin("capturemanager")
|
|
capture_fixture = CaptureFixture(SysCaptureBinary, request, _ispytest=True)
|
|
capman.set_fixture(capture_fixture)
|
|
capture_fixture._start()
|
|
yield capture_fixture
|
|
capture_fixture.close()
|
|
capman.unset_fixture()
|
|
|
|
|
|
@fixture
|
|
def capfd(request: SubRequest) -> Generator[CaptureFixture[str], None, None]:
|
|
r"""Enable text capturing of writes to file descriptors ``1`` and ``2``.
|
|
|
|
The captured output is made available via ``capfd.readouterr()`` method
|
|
calls, which return a ``(out, err)`` namedtuple.
|
|
``out`` and ``err`` will be ``text`` objects.
|
|
|
|
Returns an instance of :class:`CaptureFixture[str] <pytest.CaptureFixture>`.
|
|
|
|
Example:
|
|
|
|
.. code-block:: python
|
|
|
|
def test_system_echo(capfd):
|
|
os.system('echo "hello"')
|
|
captured = capfd.readouterr()
|
|
assert captured.out == "hello\n"
|
|
"""
|
|
capman: CaptureManager = request.config.pluginmanager.getplugin("capturemanager")
|
|
capture_fixture = CaptureFixture(FDCapture, request, _ispytest=True)
|
|
capman.set_fixture(capture_fixture)
|
|
capture_fixture._start()
|
|
yield capture_fixture
|
|
capture_fixture.close()
|
|
capman.unset_fixture()
|
|
|
|
|
|
@fixture
|
|
def capfdbinary(request: SubRequest) -> Generator[CaptureFixture[bytes], None, None]:
|
|
r"""Enable bytes capturing of writes to file descriptors ``1`` and ``2``.
|
|
|
|
The captured output is made available via ``capfd.readouterr()`` method
|
|
calls, which return a ``(out, err)`` namedtuple.
|
|
``out`` and ``err`` will be ``byte`` objects.
|
|
|
|
Returns an instance of :class:`CaptureFixture[bytes] <pytest.CaptureFixture>`.
|
|
|
|
Example:
|
|
|
|
.. code-block:: python
|
|
|
|
def test_system_echo(capfdbinary):
|
|
os.system('echo "hello"')
|
|
captured = capfdbinary.readouterr()
|
|
assert captured.out == b"hello\n"
|
|
|
|
"""
|
|
capman: CaptureManager = request.config.pluginmanager.getplugin("capturemanager")
|
|
capture_fixture = CaptureFixture(FDCaptureBinary, request, _ispytest=True)
|
|
capman.set_fixture(capture_fixture)
|
|
capture_fixture._start()
|
|
yield capture_fixture
|
|
capture_fixture.close()
|
|
capman.unset_fixture()
|