p53209761
/
test_ok2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

capture: cleanup item fixture handling (#6663) 

This started by looking at how to get the current test item in general,
and then I noticed that it is not necessary for the capture plugin to
track it manually in the first place.
This commit is contained in:
Daniel Hahler 2020-02-07 19:23:37 +01:00 committed by GitHub
parent 9e262038c8
commit b4ace46c42
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 66 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

140
src/_pytest/capture.py
View File

@ -10,11 +10,14 @@ 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.config import Config
from _pytest.fixtures import FixtureRequest
patchsysdict = {0: "stdin", 1: "stdout", 2: "stderr"}
@ -40,7 +43,7 @@ def pytest_addoption(parser):
@pytest.hookimpl(hookwrapper=True)
def pytest_load_initial_conftests(early_config, parser, args):
def pytest_load_initial_conftests(early_config: Config):
ns = early_config.known_args_namespace
if ns.capture == "fd":
_py36_windowsconsoleio_workaround(sys.stdout)
@ -76,14 +79,14 @@ class CaptureManager:
case special handling is needed to ensure the fixtures take precedence over the global capture.
"""
def __init__(self, method):
def __init__(self, method) -> None:
self._method = method
self._global_capturing = None
self._current_item = None
self._capture_fixture = None # type: Optional[CaptureFixture]
def __repr__(self):
return "<CaptureManager _method={!r} _global_capturing={!r} _current_item={!r}>".format(
self._method, self._global_capturing, self._current_item
return "<CaptureManager _method={!r} _global_capturing={!r} _item_capture={!r}>".format(
self._method, self._global_capturing, self._capture_fixture
)
def _getcapture(self, method):
@ -100,11 +103,8 @@ class CaptureManager:
def is_capturing(self):
if self.is_globally_capturing():
return "global"
capture_fixture = getattr(self._current_item, "_capture_fixture", None)
if capture_fixture is not None:
return (
"fixture %s" % self._current_item._capture_fixture.request.fixturename
)
if self._capture_fixture:
return "fixture %s" % self._capture_fixture.request.fixturename
return False
# Global capturing control
@ -136,41 +136,59 @@ class CaptureManager:
def suspend(self, in_=False):
# Need to undo local capsys-et-al if it exists before disabling global capture.
self.suspend_fixture(self._current_item)
self.suspend_fixture()
self.suspend_global_capture(in_)
def resume(self):
self.resume_global_capture()
self.resume_fixture(self._current_item)
self.resume_fixture()
def read_global_capture(self):
return self._global_capturing.readouterr()
# Fixture Control (it's just forwarding, think about removing this later)
def activate_fixture(self, item):
@contextlib.contextmanager
def _capturing_for_request(
self, request: FixtureRequest
) -> Generator["CaptureFixture", None, None]:
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.
"""
fixture = getattr(item, "_capture_fixture", None)
if fixture is not None:
fixture._start()
if self._capture_fixture:
self._capture_fixture._start()
def deactivate_fixture(self, item):
def deactivate_fixture(self):
"""Deactivates the ``capsys`` or ``capfd`` fixture of this item, if any."""
fixture = getattr(item, "_capture_fixture", None)
if fixture is not None:
fixture.close()
if self._capture_fixture:
self._capture_fixture.close()
def suspend_fixture(self, item):
fixture = getattr(item, "_capture_fixture", None)
if fixture is not None:
fixture._suspend()
def suspend_fixture(self):
if self._capture_fixture:
self._capture_fixture._suspend()
def resume_fixture(self, item):
fixture = getattr(item, "_capture_fixture", None)
if fixture is not None:
fixture._resume()
def resume_fixture(self):
if self._capture_fixture:
self._capture_fixture._resume()
# Helper context managers
@ -186,11 +204,11 @@ class CaptureManager:
@contextlib.contextmanager
def item_capture(self, when, item):
self.resume_global_capture()
self.activate_fixture(item)
self.activate_fixture()
try:
yield
finally:
self.deactivate_fixture(item)
self.deactivate_fixture()
self.suspend_global_capture(in_=False)
out, err = self.read_global_capture()
@ -214,12 +232,6 @@ class CaptureManager:
else:
yield
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_protocol(self, item):
self._current_item = item
yield
self._current_item = None
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_setup(self, item):
with self.item_capture("setup", item):
@ -244,18 +256,6 @@ class CaptureManager:
self.stop_global_capturing()
capture_fixtures = {"capfd", "capfdbinary", "capsys", "capsysbinary"}
def _ensure_only_one_capture_fixture(request: FixtureRequest, name):
fixtures = sorted(set(request.fixturenames) & capture_fixtures - {name})
if fixtures:
arg = fixtures[0] if len(fixtures) == 1 else fixtures
raise request.raiseerror(
"cannot use {} and {} at the same time".format(arg, name)
)
@pytest.fixture
def capsys(request):
"""Enable text capturing of writes to ``sys.stdout`` and ``sys.stderr``.
@ -264,8 +264,8 @@ def capsys(request):
calls, which return a ``(out, err)`` namedtuple.
``out`` and ``err`` will be ``text`` objects.
"""
_ensure_only_one_capture_fixture(request, "capsys")
with _install_capture_fixture_on_item(request, SysCapture) as fixture:
capman = request.config.pluginmanager.getplugin("capturemanager")
with capman._capturing_for_request(request) as fixture:
yield fixture
@ -277,8 +277,8 @@ def capsysbinary(request):
method calls, which return a ``(out, err)`` namedtuple.
``out`` and ``err`` will be ``bytes`` objects.
"""
_ensure_only_one_capture_fixture(request, "capsysbinary")
with _install_capture_fixture_on_item(request, SysCaptureBinary) as fixture:
capman = request.config.pluginmanager.getplugin("capturemanager")
with capman._capturing_for_request(request) as fixture:
yield fixture
@ -290,12 +290,12 @@ def capfd(request):
calls, which return a ``(out, err)`` namedtuple.
``out`` and ``err`` will be ``text`` objects.
"""
_ensure_only_one_capture_fixture(request, "capfd")
if not hasattr(os, "dup"):
pytest.skip(
"capfd fixture needs os.dup function which is not available in this system"
)
with _install_capture_fixture_on_item(request, FDCapture) as fixture:
capman = request.config.pluginmanager.getplugin("capturemanager")
with capman._capturing_for_request(request) as fixture:
yield fixture
@ -307,35 +307,15 @@ def capfdbinary(request):
calls, which return a ``(out, err)`` namedtuple.
``out`` and ``err`` will be ``byte`` objects.
"""
_ensure_only_one_capture_fixture(request, "capfdbinary")
if not hasattr(os, "dup"):
pytest.skip(
"capfdbinary fixture needs os.dup function which is not available in this system"
)
with _install_capture_fixture_on_item(request, FDCaptureBinary) as fixture:
capman = request.config.pluginmanager.getplugin("capturemanager")
with capman._capturing_for_request(request) as fixture:
yield fixture
@contextlib.contextmanager
def _install_capture_fixture_on_item(request, capture_class):
"""
Context manager which creates a ``CaptureFixture`` instance and "installs" it on
the item/node of the given request. Used by ``capsys`` and ``capfd``.
The CaptureFixture is added as attribute of the item because it needs to accessed
by ``CaptureManager`` during its ``pytest_runtest_*`` hooks.
"""
request.node._capture_fixture = fixture = CaptureFixture(capture_class, request)
capmanager = request.config.pluginmanager.getplugin("capturemanager")
# Need to active this fixture right away in case it is being used by another fixture (setup phase).
# If this fixture is being used only by a test function (call phase), then we wouldn't need this
# activation, but it doesn't hurt.
capmanager.activate_fixture(request.node)
yield fixture
fixture.close()
del request.node._capture_fixture
class CaptureFixture:
"""
Object returned by :py:func:`capsys`, :py:func:`capsysbinary`, :py:func:`capfd` and :py:func:`capfdbinary`
@ -707,6 +687,14 @@ class SysCaptureBinary(SysCapture):
return res
map_fixname_class = {
"capfd": FDCapture,
"capfdbinary": FDCaptureBinary,
"capsys": SysCapture,
"capsysbinary": SysCaptureBinary,
}
class DontReadFromInput:
encoding = None

4
testing/test_capture.py
View File

@ -477,9 +477,9 @@ class TestCaptureFixture:
result.stdout.fnmatch_lines(
[
"*test_one*",
"*capsys*capfd*same*time*",
"E * cannot use capfd and capsys at the same time",
"*test_two*",
"*capfd*capsys*same*time*",
"E * cannot use capsys and capfd at the same time",
"*2 failed in*",
]
)