capture.py

"""
per-test stdout/stderr capturing mechanism.

"""
import collections
import contextlib
import io
import os
import sys
from io import UnsupportedOperation
from tempfile import TemporaryFile
from typing import BinaryIO
from typing import Generator
from typing import Iterable
from typing import Optional

import pytest
from _pytest.compat import CaptureAndPassthroughIO
from _pytest.compat import CaptureIO
from _pytest.compat import TYPE_CHECKING
from _pytest.config import Config
from _pytest.fixtures import SubRequest

if TYPE_CHECKING:
    from typing_extensions import Literal

    _CaptureMethod = Literal["fd", "sys", "no", "tee-sys"]

patchsysdict = {0: "stdin", 1: "stdout", 2: "stderr"}


def pytest_addoption(parser):
    group = parser.getgroup("general")
    group._addoption(
        "--capture",
        action="store",
        default="fd" if hasattr(os, "dup") else "sys",
        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.",
    )


@pytest.hookimpl(hookwrapper=True)
def pytest_load_initial_conftests(early_config: Config):
    ns = early_config.known_args_namespace
    if ns.capture == "fd":
        _py36_windowsconsoleio_workaround(sys.stdout)
    _colorama_workaround()
    _readline_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 (issue93)
    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)


def _get_multicapture(method: "_CaptureMethod") -> "MultiCapture":
    if method == "fd":
        return MultiCapture(out=True, err=True, Capture=FDCapture)
    elif method == "sys":
        return MultiCapture(out=True, err=True, Capture=SysCapture)
    elif method == "no":
        return MultiCapture(out=False, err=False, in_=False)
    elif method == "tee-sys":
        return MultiCapture(out=True, err=True, in_=False, Capture=TeeSysCapture)
    raise ValueError("unknown capturing method: {!r}".format(method))


class CaptureManager:
    """
    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: which is 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 = method
        self._global_capturing = None
        self._capture_fixture = None  # type: Optional[CaptureFixture]

    def __repr__(self):
        return "<CaptureManager _method={!r} _global_capturing={!r} _capture_fixture={!r}>".format(
            self._method, self._global_capturing, self._capture_fixture
        )

    def is_capturing(self):
        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):
        return self._method != "no"

    def start_global_capturing(self):
        assert self._global_capturing is None
        self._global_capturing = _get_multicapture(self._method)
        self._global_capturing.start_capturing()

    def stop_global_capturing(self):
        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):
        # 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_=False):
        cap = getattr(self, "_global_capturing", None)
        if cap is not None:
            cap.suspend_capturing(in_=in_)

    def suspend(self, in_=False):
        # 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):
        self.resume_global_capture()
        self.resume_fixture()

    def read_global_capture(self):
        return self._global_capturing.readouterr()

    # Fixture Control (it's just forwarding, think about removing this later)

    @contextlib.contextmanager
    def _capturing_for_request(
        self, request: SubRequest,
    ) -> Generator["CaptureFixture", None, None]:
        """
        Context manager that creates a ``CaptureFixture`` instance for the
        given ``request``, ensuring there is only a single one being requested
        at the same time.

        This is used as a helper with ``capsys``, ``capfd`` etc.
        """
        if self._capture_fixture:
            other_name = next(
                k
                for k, v in map_fixname_class.items()
                if v is self._capture_fixture.captureclass
            )
            raise request.raiseerror(
                "cannot use {} and {} at the same time".format(
                    request.fixturename, other_name
                )
            )
        capture_class = map_fixname_class[request.fixturename]
        self._capture_fixture = CaptureFixture(capture_class, request)
        self.activate_fixture()
        yield self._capture_fixture
        self._capture_fixture.close()
        self._capture_fixture = None

    def activate_fixture(self):
        """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):
        """Deactivates the ``capsys`` or ``capfd`` fixture of this item, if any."""
        if self._capture_fixture:
            self._capture_fixture.close()

    def suspend_fixture(self):
        if self._capture_fixture:
            self._capture_fixture._suspend()

    def resume_fixture(self):
        if self._capture_fixture:
            self._capture_fixture._resume()

    # Helper context managers

    @contextlib.contextmanager
    def global_and_fixture_disabled(self):
        """Context manager to temporarily disable global and current fixture capturing."""
        self.suspend()
        try:
            yield
        finally:
            self.resume()

    @contextlib.contextmanager
    def item_capture(self, when, item):
        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

    @pytest.hookimpl(hookwrapper=True)
    def pytest_make_collect_report(self, collector):
        if isinstance(collector, pytest.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

    @pytest.hookimpl(hookwrapper=True)
    def pytest_runtest_setup(self, item):
        with self.item_capture("setup", item):
            yield

    @pytest.hookimpl(hookwrapper=True)
    def pytest_runtest_call(self, item):
        with self.item_capture("call", item):
            yield

    @pytest.hookimpl(hookwrapper=True)
    def pytest_runtest_teardown(self, item):
        with self.item_capture("teardown", item):
            yield

    @pytest.hookimpl(tryfirst=True)
    def pytest_keyboard_interrupt(self, excinfo):
        self.stop_global_capturing()

    @pytest.hookimpl(tryfirst=True)
    def pytest_internalerror(self, excinfo):
        self.stop_global_capturing()


@pytest.fixture
def capsys(request):
    """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.
    """
    capman = request.config.pluginmanager.getplugin("capturemanager")
    with capman._capturing_for_request(request) as fixture:
        yield fixture


@pytest.fixture
def capsysbinary(request):
    """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.
    """
    capman = request.config.pluginmanager.getplugin("capturemanager")
    with capman._capturing_for_request(request) as fixture:
        yield fixture


@pytest.fixture
def capfd(request):
    """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.
    """
    if not hasattr(os, "dup"):
        pytest.skip(
            "capfd fixture needs os.dup function which is not available in this system"
        )
    capman = request.config.pluginmanager.getplugin("capturemanager")
    with capman._capturing_for_request(request) as fixture:
        yield fixture


@pytest.fixture
def capfdbinary(request):
    """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.
    """
    if not hasattr(os, "dup"):
        pytest.skip(
            "capfdbinary fixture needs os.dup function which is not available in this system"
        )
    capman = request.config.pluginmanager.getplugin("capturemanager")
    with capman._capturing_for_request(request) as fixture:
        yield fixture


class CaptureFixture:
    """
    Object returned by :py:func:`capsys`, :py:func:`capsysbinary`, :py:func:`capfd` and :py:func:`capfdbinary`
    fixtures.
    """

    def __init__(self, captureclass, request):
        self.captureclass = captureclass
        self.request = request
        self._capture = None
        self._captured_out = self.captureclass.EMPTY_BUFFER
        self._captured_err = self.captureclass.EMPTY_BUFFER

    def _start(self):
        if self._capture is None:
            self._capture = MultiCapture(
                out=True, err=True, in_=False, Capture=self.captureclass
            )
            self._capture.start_capturing()

    def close(self):
        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):
        """Read and return the captured output so far, resetting the internal buffer.

        :return: 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):
        """Suspends this fixture's own capturing temporarily."""
        if self._capture is not None:
            self._capture.suspend_capturing()

    def _resume(self):
        """Resumes this fixture's own capturing temporarily."""
        if self._capture is not None:
            self._capture.resume_capturing()

    @contextlib.contextmanager
    def disabled(self):
        """Temporarily disables capture while inside the 'with' block."""
        capmanager = self.request.config.pluginmanager.getplugin("capturemanager")
        with capmanager.global_and_fixture_disabled():
            yield


def safe_text_dupfile(f, mode, default_encoding="UTF8"):
    """ return an open text file object that's a duplicate of f on the
        FD-level if possible.
    """
    encoding = getattr(f, "encoding", None)
    try:
        fd = f.fileno()
    except Exception:
        if "b" not in getattr(f, "mode", "") and hasattr(f, "encoding"):
            # we seem to have a text stream, let's just use it
            return f
    else:
        newfd = os.dup(fd)
        if "b" not in mode:
            mode += "b"
        f = os.fdopen(newfd, mode, 0)  # no buffering
    return EncodedFile(f, encoding or default_encoding)


class EncodedFile:
    errors = "strict"  # possibly needed by py3 code (issue555)

    def __init__(self, buffer: BinaryIO, encoding: str) -> None:
        self.buffer = buffer
        self.encoding = encoding

    def write(self, s: str) -> int:
        if not isinstance(s, str):
            raise TypeError(
                "write() argument must be str, not {}".format(type(s).__name__)
            )
        return self.buffer.write(s.encode(self.encoding, "replace"))

    def writelines(self, lines: Iterable[str]) -> None:
        self.buffer.writelines(x.encode(self.encoding, "replace") for x in lines)

    @property
    def name(self) -> str:
        """Ensure that file.name is a string."""
        return repr(self.buffer)

    @property
    def mode(self) -> str:
        return self.buffer.mode.replace("b", "")

    def __getattr__(self, name):
        return getattr(object.__getattribute__(self, "buffer"), name)


CaptureResult = collections.namedtuple("CaptureResult", ["out", "err"])


class MultiCapture:
    out = err = in_ = None
    _state = None
    _in_suspended = False

    def __init__(self, out=True, err=True, in_=True, Capture=None):
        if in_:
            self.in_ = Capture(0)
        if out:
            self.out = Capture(1)
        if err:
            self.err = Capture(2)

    def __repr__(self):
        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):
        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):
        """ pop current snapshot out/err capture and flush to orig streams. """
        out, err = self.readouterr()
        if out:
            self.out.writeorg(out)
        if err:
            self.err.writeorg(err)
        return out, err

    def suspend_capturing(self, in_=False):
        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):
        self._state = "resumed"
        if self.out:
            self.out.resume()
        if self.err:
            self.err.resume()
        if self._in_suspended:
            self.in_.resume()
            self._in_suspended = False

    def stop_capturing(self):
        """ 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 readouterr(self):
        """ return snapshot unicode value of stdout/stderr capturings. """
        return CaptureResult(
            self.out.snap() if self.out is not None else "",
            self.err.snap() if self.err is not None else "",
        )


class NoCapture:
    EMPTY_BUFFER = None
    __init__ = start = done = suspend = resume = lambda *args: None


class FDCaptureBinary:
    """Capture IO to/from a given os-level filedescriptor.

    snap() produces `bytes`
    """

    EMPTY_BUFFER = b""
    _state = None

    def __init__(self, targetfd, tmpfile=None):
        self.targetfd = targetfd
        try:
            self.targetfd_save = os.dup(self.targetfd)
        except OSError:
            self.start = lambda: None
            self.done = lambda: None
        else:
            self.start = self._start
            self.done = self._done
            if targetfd == 0:
                assert not tmpfile, "cannot set tmpfile with stdin"
                tmpfile = open(os.devnull, "r")
                self.syscapture = SysCapture(targetfd)
            else:
                if tmpfile is None:
                    f = TemporaryFile()
                    with f:
                        tmpfile = safe_text_dupfile(f, mode="wb+")
                if targetfd in patchsysdict:
                    self.syscapture = SysCapture(targetfd, tmpfile)
                else:
                    self.syscapture = NoCapture()
            self.tmpfile = tmpfile
            self.tmpfile_fd = tmpfile.fileno()

    def __repr__(self):
        return "<{} {} oldfd={} _state={!r} tmpfile={}>".format(
            self.__class__.__name__,
            self.targetfd,
            getattr(self, "targetfd_save", "<UNSET>"),
            self._state,
            hasattr(self, "tmpfile") and repr(self.tmpfile) or "<UNSET>",
        )

    def _start(self):
        """ Start capturing on targetfd using memorized tmpfile. """
        try:
            os.fstat(self.targetfd_save)
        except (AttributeError, OSError):
            raise ValueError("saved filedescriptor not valid anymore")
        os.dup2(self.tmpfile_fd, self.targetfd)
        self.syscapture.start()
        self._state = "started"

    def snap(self):
        self.tmpfile.seek(0)
        res = self.tmpfile.read()
        self.tmpfile.seek(0)
        self.tmpfile.truncate()
        return res

    def _done(self):
        """ stop capturing, restore streams, return original capture file,
        seeked to position zero. """
        targetfd_save = self.__dict__.pop("targetfd_save")
        os.dup2(targetfd_save, self.targetfd)
        os.close(targetfd_save)
        self.syscapture.done()
        self.tmpfile.close()
        self._state = "done"

    def suspend(self):
        self.syscapture.suspend()
        os.dup2(self.targetfd_save, self.targetfd)
        self._state = "suspended"

    def resume(self):
        self.syscapture.resume()
        os.dup2(self.tmpfile_fd, self.targetfd)
        self._state = "resumed"

    def writeorg(self, data):
        """ write to original file descriptor. """
        if isinstance(data, str):
            data = data.encode("utf8")  # XXX use encoding of original stream
        os.write(self.targetfd_save, data)


class FDCapture(FDCaptureBinary):
    """Capture IO to/from a given os-level filedescriptor.

    snap() produces text
    """

    # Ignore type because it doesn't match the type in the superclass (bytes).
    EMPTY_BUFFER = str()  # type: ignore

    def snap(self):
        res = super().snap()
        enc = getattr(self.tmpfile, "encoding", None)
        if enc and isinstance(res, bytes):
            res = str(res, enc, "replace")
        return res


class SysCaptureBinary:

    EMPTY_BUFFER = b""
    _state = None

    def __init__(self, fd, tmpfile=None):
        name = patchsysdict[fd]
        self._old = getattr(sys, name)
        self.name = name
        if tmpfile is None:
            if name == "stdin":
                tmpfile = DontReadFromInput()
            else:
                tmpfile = CaptureIO()
        self.tmpfile = tmpfile

    def __repr__(self):
        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 start(self):
        setattr(sys, self.name, self.tmpfile)
        self._state = "started"

    def snap(self):
        res = self.tmpfile.buffer.getvalue()
        self.tmpfile.seek(0)
        self.tmpfile.truncate()
        return res

    def done(self):
        setattr(sys, self.name, self._old)
        del self._old
        self.tmpfile.close()
        self._state = "done"

    def suspend(self):
        setattr(sys, self.name, self._old)
        self._state = "suspended"

    def resume(self):
        setattr(sys, self.name, self.tmpfile)
        self._state = "resumed"

    def writeorg(self, data):
        self._old.write(data)
        self._old.flush()


class SysCapture(SysCaptureBinary):
    EMPTY_BUFFER = str()  # type: ignore[assignment]  # noqa: F821

    def snap(self):
        res = self.tmpfile.getvalue()
        self.tmpfile.seek(0)
        self.tmpfile.truncate()
        return res


class TeeSysCapture(SysCapture):
    def __init__(self, fd, tmpfile=None):
        name = patchsysdict[fd]
        self._old = getattr(sys, name)
        self.name = name
        if tmpfile is None:
            if name == "stdin":
                tmpfile = DontReadFromInput()
            else:
                tmpfile = CaptureAndPassthroughIO(self._old)
        self.tmpfile = tmpfile


map_fixname_class = {
    "capfd": FDCapture,
    "capfdbinary": FDCaptureBinary,
    "capsys": SysCapture,
    "capsysbinary": SysCaptureBinary,
}


class DontReadFromInput:
    encoding = None

    def read(self, *args):
        raise IOError(
            "pytest: reading from stdin while output is captured!  Consider using `-s`."
        )

    readline = read
    readlines = read
    __next__ = read

    def __iter__(self):
        return self

    def fileno(self):
        raise UnsupportedOperation("redirected stdin is pseudofile, has no fileno()")

    def isatty(self):
        return False

    def close(self):
        pass

    @property
    def buffer(self):
        return self


def _colorama_workaround():
    """
    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 _readline_workaround():
    """
    Ensure readline is imported so that it attaches to the correct stdio
    handles on Windows.

    Pdb uses readline support where available--when not running from the Python
    prompt, the readline module is not imported until running the pdb REPL.  If
    running pytest with the --pdb option this means the readline module is not
    imported until after I/O capture has been started.

    This is a problem for pyreadline, which is often used to implement readline
    support on Windows, as it does not attach to the correct handles for stdout
    and/or stdin if they have been redirected by the FDCapture mechanism.  This
    workaround ensures that readline is imported before I/O capture is setup so
    that it can attach to the actual stdin/out for the console.

    See https://github.com/pytest-dev/pytest/pull/1281
    """
    if sys.platform.startswith("win32"):
        try:
            import readline  # noqa: F401
        except ImportError:
            pass


def _py36_windowsconsoleio_workaround(stream):
    """
    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 sys.version_info[:2] < (3, 6)
        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"):
        return

    buffered = hasattr(stream.buffer, "raw")
    raw_stdout = stream.buffer.raw if buffered else stream.buffer

    if not isinstance(raw_stdout, io._WindowsConsoleIO):
        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")