Merge pull request #2801 from nicoddemus/capture-fixture
Allow fixtures to use capsys and capfd
This commit is contained in:
Ronny Pfannschmidt
GitHub
commit
9b0ce535c9
|
|
@ -43,7 +43,7 @@ def pytest_load_initial_conftests(early_config, parser, args):
|
|||
pluginmanager.register(capman, "capturemanager")
|
||||
|
||||
# make sure that capturemanager is properly reset at final shutdown
|
||||
early_config.add_cleanup(capman.reset_capturings)
|
||||
early_config.add_cleanup(capman.stop_global_capturing)
|
||||
|
||||
# make sure logging does not raise exceptions at the end
|
||||
def silence_logging_at_shutdown():
|
||||
|
|
@ -52,17 +52,30 @@ def pytest_load_initial_conftests(early_config, parser, args):
|
|||
early_config.add_cleanup(silence_logging_at_shutdown)
|
||||
|
||||
# finally trigger conftest loading but while capturing (issue93)
|
||||
capman.init_capturings()
|
||||
capman.start_global_capturing()
|
||||
outcome = yield
|
||||
out, err = capman.suspendcapture()
|
||||
out, err = capman.suspend_global_capture()
|
||||
if outcome.excinfo is not None:
|
||||
sys.stdout.write(out)
|
||||
sys.stderr.write(err)
|
||||
|
||||
|
||||
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):
|
||||
self._method = method
|
||||
self._global_capturing = None
|
||||
|
||||
def _getcapture(self, method):
|
||||
if method == "fd":
|
||||
|
|
@ -74,23 +87,24 @@ class CaptureManager:
|
|||
else:
|
||||
raise ValueError("unknown capturing method: %r" % method)
|
||||
|
||||
def init_capturings(self):
|
||||
assert not hasattr(self, "_capturing")
|
||||
self._capturing = self._getcapture(self._method)
|
||||
self._capturing.start_capturing()
|
||||
def start_global_capturing(self):
|
||||
assert self._global_capturing is None
|
||||
self._global_capturing = self._getcapture(self._method)
|
||||
self._global_capturing.start_capturing()
|
||||
|
||||
def reset_capturings(self):
|
||||
cap = self.__dict__.pop("_capturing", None)
|
||||
if cap is not None:
|
||||
cap.pop_outerr_to_orig()
|
||||
cap.stop_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 resumecapture(self):
|
||||
self._capturing.resume_capturing()
|
||||
def resume_global_capture(self):
|
||||
self._global_capturing.resume_capturing()
|
||||
|
||||
def suspendcapture(self, in_=False):
|
||||
self.deactivate_funcargs()
|
||||
cap = getattr(self, "_capturing", None)
|
||||
def suspend_global_capture(self, item=None, in_=False):
|
||||
if item is not None:
|
||||
self.deactivate_fixture(item)
|
||||
cap = getattr(self, "_global_capturing", None)
|
||||
if cap is not None:
|
||||
try:
|
||||
outerr = cap.readouterr()
|
||||
|
|
@ -98,23 +112,26 @@ class CaptureManager:
|
|||
cap.suspend_capturing(in_=in_)
|
||||
return outerr
|
||||
|
||||
def activate_funcargs(self, pyfuncitem):
|
||||
capfuncarg = pyfuncitem.__dict__.pop("_capfuncarg", None)
|
||||
if capfuncarg is not None:
|
||||
capfuncarg._start()
|
||||
self._capfuncarg = capfuncarg
|
||||
def activate_fixture(self, item):
|
||||
"""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()
|
||||
|
||||
def deactivate_funcargs(self):
|
||||
capfuncarg = self.__dict__.pop("_capfuncarg", None)
|
||||
if capfuncarg is not None:
|
||||
capfuncarg.close()
|
||||
def deactivate_fixture(self, item):
|
||||
"""Deactivates the ``capsys`` or ``capfd`` fixture of this item, if any."""
|
||||
fixture = getattr(item, "_capture_fixture", None)
|
||||
if fixture is not None:
|
||||
fixture.close()
|
||||
|
||||
@pytest.hookimpl(hookwrapper=True)
|
||||
def pytest_make_collect_report(self, collector):
|
||||
if isinstance(collector, pytest.File):
|
||||
self.resumecapture()
|
||||
self.resume_global_capture()
|
||||
outcome = yield
|
||||
out, err = self.suspendcapture()
|
||||
out, err = self.suspend_global_capture()
|
||||
rep = outcome.get_result()
|
||||
if out:
|
||||
rep.sections.append(("Captured stdout", out))
|
||||
|
|
@ -125,34 +142,39 @@ class CaptureManager:
|
|||
|
||||
@pytest.hookimpl(hookwrapper=True)
|
||||
def pytest_runtest_setup(self, item):
|
||||
self.resumecapture()
|
||||
self.resume_global_capture()
|
||||
# no need to activate a capture fixture because they activate themselves during creation; this
|
||||
# only makes sense when a fixture uses a capture fixture, otherwise the capture fixture will
|
||||
# be activated during pytest_runtest_call
|
||||
yield
|
||||
self.suspendcapture_item(item, "setup")
|
||||
self.suspend_capture_item(item, "setup")
|
||||
|
||||
@pytest.hookimpl(hookwrapper=True)
|
||||
def pytest_runtest_call(self, item):
|
||||
self.resumecapture()
|
||||
self.activate_funcargs(item)
|
||||
self.resume_global_capture()
|
||||
# it is important to activate this fixture during the call phase so it overwrites the "global"
|
||||
# capture
|
||||
self.activate_fixture(item)
|
||||
yield
|
||||
# self.deactivate_funcargs() called from suspendcapture()
|
||||
self.suspendcapture_item(item, "call")
|
||||
self.suspend_capture_item(item, "call")
|
||||
|
||||
@pytest.hookimpl(hookwrapper=True)
|
||||
def pytest_runtest_teardown(self, item):
|
||||
self.resumecapture()
|
||||
self.resume_global_capture()
|
||||
self.activate_fixture(item)
|
||||
yield
|
||||
self.suspendcapture_item(item, "teardown")
|
||||
self.suspend_capture_item(item, "teardown")
|
||||
|
||||
@pytest.hookimpl(tryfirst=True)
|
||||
def pytest_keyboard_interrupt(self, excinfo):
|
||||
self.reset_capturings()
|
||||
self.stop_global_capturing()
|
||||
|
||||
@pytest.hookimpl(tryfirst=True)
|
||||
def pytest_internalerror(self, excinfo):
|
||||
self.reset_capturings()
|
||||
self.stop_global_capturing()
|
||||
|
||||
def suspendcapture_item(self, item, when, in_=False):
|
||||
out, err = self.suspendcapture(in_=in_)
|
||||
def suspend_capture_item(self, item, when, in_=False):
|
||||
out, err = self.suspend_global_capture(item, in_=in_)
|
||||
item.add_report_section(when, "stdout", out)
|
||||
item.add_report_section(when, "stderr", err)
|
||||
|
||||
|
|
@ -168,8 +190,8 @@ def capsys(request):
|
|||
"""
|
||||
if "capfd" in request.fixturenames:
|
||||
raise request.raiseerror(error_capsysfderror)
|
||||
request.node._capfuncarg = c = CaptureFixture(SysCapture, request)
|
||||
return c
|
||||
with _install_capture_fixture_on_item(request, SysCapture) as fixture:
|
||||
yield fixture
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -181,9 +203,29 @@ def capfd(request):
|
|||
if "capsys" in request.fixturenames:
|
||||
request.raiseerror(error_capsysfderror)
|
||||
if not hasattr(os, 'dup'):
|
||||
pytest.skip("capfd funcarg needs os.dup")
|
||||
request.node._capfuncarg = c = CaptureFixture(FDCapture, request)
|
||||
return c
|
||||
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:
|
||||
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:
|
||||
|
|
@ -210,12 +252,14 @@ class CaptureFixture:
|
|||
|
||||
@contextlib.contextmanager
|
||||
def disabled(self):
|
||||
self._capture.suspend_capturing()
|
||||
capmanager = self.request.config.pluginmanager.getplugin('capturemanager')
|
||||
capmanager.suspendcapture_item(self.request.node, "call", in_=True)
|
||||
capmanager.suspend_global_capture(item=None, in_=False)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
capmanager.resumecapture()
|
||||
capmanager.resume_global_capture()
|
||||
self._capture.resume_capturing()
|
||||
|
||||
|
||||
def safe_text_dupfile(f, mode, default_encoding="UTF8"):
|
||||
|
|
|
|||
|
|
@ -54,7 +54,7 @@ class pytestPDB:
|
|||
if cls._pluginmanager is not None:
|
||||
capman = cls._pluginmanager.getplugin("capturemanager")
|
||||
if capman:
|
||||
capman.suspendcapture(in_=True)
|
||||
capman.suspend_global_capture(in_=True)
|
||||
tw = _pytest.config.create_terminal_writer(cls._config)
|
||||
tw.line()
|
||||
tw.sep(">", "PDB set_trace (IO-capturing turned off)")
|
||||
|
|
@ -66,7 +66,7 @@ class PdbInvoke:
|
|||
def pytest_exception_interact(self, node, call, report):
|
||||
capman = node.config.pluginmanager.getplugin("capturemanager")
|
||||
if capman:
|
||||
out, err = capman.suspendcapture(in_=True)
|
||||
out, err = capman.suspend_global_capture(in_=True)
|
||||
sys.stdout.write(out)
|
||||
sys.stdout.write(err)
|
||||
_enter_pdb(node, call.excinfo, report)
|
||||
|
|
|
|||
|
|
@ -44,7 +44,7 @@ def _show_fixture_action(fixturedef, msg):
|
|||
config = fixturedef._fixturemanager.config
|
||||
capman = config.pluginmanager.getplugin('capturemanager')
|
||||
if capman:
|
||||
out, err = capman.suspendcapture()
|
||||
out, err = capman.suspend_global_capture()
|
||||
|
||||
tw = config.get_terminal_writer()
|
||||
tw.line()
|
||||
|
|
@ -63,7 +63,7 @@ def _show_fixture_action(fixturedef, msg):
|
|||
tw.write('[{0}]'.format(fixturedef.cached_param))
|
||||
|
||||
if capman:
|
||||
capman.resumecapture()
|
||||
capman.resume_global_capture()
|
||||
sys.stdout.write(out)
|
||||
sys.stderr.write(err)
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
Resume output capturing after ``capsys/capfd.disabled()`` context manager.
|
||||
|
|
@ -0,0 +1 @@
|
|||
``capsys`` and ``capfd`` can now be used by other fixtures.
|
||||
|
|
@ -78,23 +78,23 @@ class TestCaptureManager(object):
|
|||
old = sys.stdout, sys.stderr, sys.stdin
|
||||
try:
|
||||
capman = CaptureManager(method)
|
||||
capman.init_capturings()
|
||||
outerr = capman.suspendcapture()
|
||||
capman.start_global_capturing()
|
||||
outerr = capman.suspend_global_capture()
|
||||
assert outerr == ("", "")
|
||||
outerr = capman.suspendcapture()
|
||||
outerr = capman.suspend_global_capture()
|
||||
assert outerr == ("", "")
|
||||
print("hello")
|
||||
out, err = capman.suspendcapture()
|
||||
out, err = capman.suspend_global_capture()
|
||||
if method == "no":
|
||||
assert old == (sys.stdout, sys.stderr, sys.stdin)
|
||||
else:
|
||||
assert not out
|
||||
capman.resumecapture()
|
||||
capman.resume_global_capture()
|
||||
print("hello")
|
||||
out, err = capman.suspendcapture()
|
||||
out, err = capman.suspend_global_capture()
|
||||
if method != "no":
|
||||
assert out == "hello\n"
|
||||
capman.reset_capturings()
|
||||
capman.stop_global_capturing()
|
||||
finally:
|
||||
capouter.stop_capturing()
|
||||
|
||||
|
|
@ -103,9 +103,9 @@ class TestCaptureManager(object):
|
|||
capouter = StdCaptureFD()
|
||||
try:
|
||||
capman = CaptureManager("fd")
|
||||
capman.init_capturings()
|
||||
pytest.raises(AssertionError, "capman.init_capturings()")
|
||||
capman.reset_capturings()
|
||||
capman.start_global_capturing()
|
||||
pytest.raises(AssertionError, "capman.start_global_capturing()")
|
||||
capman.stop_global_capturing()
|
||||
finally:
|
||||
capouter.stop_capturing()
|
||||
|
||||
|
|
@ -502,20 +502,64 @@ class TestCaptureFixture(object):
|
|||
assert 'closed' not in result.stderr.str()
|
||||
|
||||
@pytest.mark.parametrize('fixture', ['capsys', 'capfd'])
|
||||
def test_disabled_capture_fixture(self, testdir, fixture):
|
||||
@pytest.mark.parametrize('no_capture', [True, False])
|
||||
def test_disabled_capture_fixture(self, testdir, fixture, no_capture):
|
||||
testdir.makepyfile("""
|
||||
def test_disabled({fixture}):
|
||||
print('captured before')
|
||||
with {fixture}.disabled():
|
||||
print('while capture is disabled')
|
||||
print('captured after')
|
||||
assert {fixture}.readouterr() == ('captured before\\ncaptured after\\n', '')
|
||||
|
||||
def test_normal():
|
||||
print('test_normal executed')
|
||||
""".format(fixture=fixture))
|
||||
result = testdir.runpytest_subprocess()
|
||||
args = ('-s',) if no_capture else ()
|
||||
result = testdir.runpytest_subprocess(*args)
|
||||
result.stdout.fnmatch_lines("""
|
||||
*while capture is disabled*
|
||||
""")
|
||||
assert 'captured before' not in result.stdout.str()
|
||||
assert 'captured after' not in result.stdout.str()
|
||||
if no_capture:
|
||||
assert 'test_normal executed' in result.stdout.str()
|
||||
else:
|
||||
assert 'test_normal executed' not in result.stdout.str()
|
||||
|
||||
@pytest.mark.parametrize('fixture', ['capsys', 'capfd'])
|
||||
def test_fixture_use_by_other_fixtures(self, testdir, fixture):
|
||||
"""
|
||||
Ensure that capsys and capfd can be used by other fixtures during setup and teardown.
|
||||
"""
|
||||
testdir.makepyfile("""
|
||||
from __future__ import print_function
|
||||
import sys
|
||||
import pytest
|
||||
|
||||
@pytest.fixture
|
||||
def captured_print({fixture}):
|
||||
print('stdout contents begin')
|
||||
print('stderr contents begin', file=sys.stderr)
|
||||
out, err = {fixture}.readouterr()
|
||||
|
||||
yield out, err
|
||||
|
||||
print('stdout contents end')
|
||||
print('stderr contents end', file=sys.stderr)
|
||||
out, err = {fixture}.readouterr()
|
||||
assert out == 'stdout contents end\\n'
|
||||
assert err == 'stderr contents end\\n'
|
||||
|
||||
def test_captured_print(captured_print):
|
||||
out, err = captured_print
|
||||
assert out == 'stdout contents begin\\n'
|
||||
assert err == 'stderr contents begin\\n'
|
||||
""".format(fixture=fixture))
|
||||
result = testdir.runpytest_subprocess()
|
||||
result.stdout.fnmatch_lines("*1 passed*")
|
||||
assert 'stdout contents begin' not in result.stdout.str()
|
||||
assert 'stderr contents begin' not in result.stdout.str()
|
||||
|
||||
|
||||
def test_setup_failure_does_not_kill_capturing(testdir):
|
||||
|
|
|
|||
Loading…
Reference in New Issue