parent
de0d19ca09
commit
9919269ed0
|
@ -61,6 +61,18 @@ def pytest_load_initial_conftests(early_config, parser, args):
|
||||||
|
|
||||||
|
|
||||||
class CaptureManager:
|
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):
|
def __init__(self, method):
|
||||||
self._method = method
|
self._method = method
|
||||||
|
|
||||||
|
@ -88,8 +100,9 @@ class CaptureManager:
|
||||||
def resumecapture(self):
|
def resumecapture(self):
|
||||||
self._capturing.resume_capturing()
|
self._capturing.resume_capturing()
|
||||||
|
|
||||||
def suspendcapture(self, in_=False):
|
def suspendcapture(self, item=None, in_=False):
|
||||||
self.deactivate_funcargs()
|
if item is not None:
|
||||||
|
self.deactivate_fixture(item)
|
||||||
cap = getattr(self, "_capturing", None)
|
cap = getattr(self, "_capturing", None)
|
||||||
if cap is not None:
|
if cap is not None:
|
||||||
try:
|
try:
|
||||||
|
@ -98,16 +111,19 @@ class CaptureManager:
|
||||||
cap.suspend_capturing(in_=in_)
|
cap.suspend_capturing(in_=in_)
|
||||||
return outerr
|
return outerr
|
||||||
|
|
||||||
def activate_funcargs(self, pyfuncitem):
|
def activate_fixture(self, item):
|
||||||
capfuncarg = pyfuncitem.__dict__.pop("_capfuncarg", None)
|
"""If the current item is using ``capsys`` or ``capfd``, activate them so they take precedence over
|
||||||
if capfuncarg is not None:
|
the global capture.
|
||||||
capfuncarg._start()
|
"""
|
||||||
self._capfuncarg = capfuncarg
|
fixture = getattr(item, "_capture_fixture", None)
|
||||||
|
if fixture is not None:
|
||||||
|
fixture._start()
|
||||||
|
|
||||||
def deactivate_funcargs(self):
|
def deactivate_fixture(self, item):
|
||||||
capfuncarg = self.__dict__.pop("_capfuncarg", None)
|
"""Deactivates the ``capsys`` or ``capfd`` fixture of this item, if any."""
|
||||||
if capfuncarg is not None:
|
fixture = getattr(item, "_capture_fixture", None)
|
||||||
capfuncarg.close()
|
if fixture is not None:
|
||||||
|
fixture.close()
|
||||||
|
|
||||||
@pytest.hookimpl(hookwrapper=True)
|
@pytest.hookimpl(hookwrapper=True)
|
||||||
def pytest_make_collect_report(self, collector):
|
def pytest_make_collect_report(self, collector):
|
||||||
|
@ -126,20 +142,25 @@ class CaptureManager:
|
||||||
@pytest.hookimpl(hookwrapper=True)
|
@pytest.hookimpl(hookwrapper=True)
|
||||||
def pytest_runtest_setup(self, item):
|
def pytest_runtest_setup(self, item):
|
||||||
self.resumecapture()
|
self.resumecapture()
|
||||||
|
# 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
|
yield
|
||||||
self.suspendcapture_item(item, "setup")
|
self.suspendcapture_item(item, "setup")
|
||||||
|
|
||||||
@pytest.hookimpl(hookwrapper=True)
|
@pytest.hookimpl(hookwrapper=True)
|
||||||
def pytest_runtest_call(self, item):
|
def pytest_runtest_call(self, item):
|
||||||
self.resumecapture()
|
self.resumecapture()
|
||||||
self.activate_funcargs(item)
|
# it is important to activate this fixture during the call phase so it overwrites the "global"
|
||||||
|
# capture
|
||||||
|
self.activate_fixture(item)
|
||||||
yield
|
yield
|
||||||
# self.deactivate_funcargs() called from suspendcapture()
|
|
||||||
self.suspendcapture_item(item, "call")
|
self.suspendcapture_item(item, "call")
|
||||||
|
|
||||||
@pytest.hookimpl(hookwrapper=True)
|
@pytest.hookimpl(hookwrapper=True)
|
||||||
def pytest_runtest_teardown(self, item):
|
def pytest_runtest_teardown(self, item):
|
||||||
self.resumecapture()
|
self.resumecapture()
|
||||||
|
self.activate_fixture(item)
|
||||||
yield
|
yield
|
||||||
self.suspendcapture_item(item, "teardown")
|
self.suspendcapture_item(item, "teardown")
|
||||||
|
|
||||||
|
@ -152,7 +173,7 @@ class CaptureManager:
|
||||||
self.reset_capturings()
|
self.reset_capturings()
|
||||||
|
|
||||||
def suspendcapture_item(self, item, when, in_=False):
|
def suspendcapture_item(self, item, when, in_=False):
|
||||||
out, err = self.suspendcapture(in_=in_)
|
out, err = self.suspendcapture(item, in_=in_)
|
||||||
item.add_report_section(when, "stdout", out)
|
item.add_report_section(when, "stdout", out)
|
||||||
item.add_report_section(when, "stderr", err)
|
item.add_report_section(when, "stderr", err)
|
||||||
|
|
||||||
|
@ -168,8 +189,8 @@ def capsys(request):
|
||||||
"""
|
"""
|
||||||
if "capfd" in request.fixturenames:
|
if "capfd" in request.fixturenames:
|
||||||
raise request.raiseerror(error_capsysfderror)
|
raise request.raiseerror(error_capsysfderror)
|
||||||
request.node._capfuncarg = c = CaptureFixture(SysCapture, request)
|
with _install_capture_fixture_on_item(request, SysCapture) as fixture:
|
||||||
return c
|
yield fixture
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
|
@ -181,9 +202,29 @@ def capfd(request):
|
||||||
if "capsys" in request.fixturenames:
|
if "capsys" in request.fixturenames:
|
||||||
request.raiseerror(error_capsysfderror)
|
request.raiseerror(error_capsysfderror)
|
||||||
if not hasattr(os, 'dup'):
|
if not hasattr(os, 'dup'):
|
||||||
pytest.skip("capfd funcarg needs os.dup")
|
pytest.skip("capfd fixture needs os.dup function which is not available in this system")
|
||||||
request.node._capfuncarg = c = CaptureFixture(FDCapture, request)
|
with _install_capture_fixture_on_item(request, FDCapture) as fixture:
|
||||||
return c
|
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:
|
class CaptureFixture:
|
||||||
|
|
|
@ -0,0 +1 @@
|
||||||
|
``capsys`` and ``capfd`` can now be used by other fixtures.
|
|
@ -517,6 +517,40 @@ class TestCaptureFixture(object):
|
||||||
assert 'captured before' not in result.stdout.str()
|
assert 'captured before' not in result.stdout.str()
|
||||||
assert 'captured after' not in result.stdout.str()
|
assert 'captured after' 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):
|
def test_setup_failure_does_not_kill_capturing(testdir):
|
||||||
sub1 = testdir.mkpydir("sub1")
|
sub1 = testdir.mkpydir("sub1")
|
||||||
|
|
Loading…
Reference in New Issue