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

Merge pull request #3832 from Sup3rGeo/bugfix/capsys-with-cli-logging 

Bugfix/capsys with cli logging (again)
This commit is contained in:
Bruno Oliveira 2018-08-21 20:12:31 -03:00 committed by GitHub
parent 43657f252f 70ebab3537
commit f1079a8222
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 154 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

130
src/_pytest/capture.py
View File

@ -14,7 +14,7 @@ from tempfile import TemporaryFile
import six
import pytest
from _pytest.compat import CaptureIO, dummy_context_manager
from _pytest.compat import CaptureIO
patchsysdict = {0: "stdin", 1: "stdout", 2: "stderr"}
@ -62,8 +62,9 @@ def pytest_load_initial_conftests(early_config, parser, args):
# finally trigger conftest loading but while capturing (issue93)
capman.start_global_capturing()
outcome = yield
out, err = capman.suspend_global_capture()
capman.suspend_global_capture()
if outcome.excinfo is not None:
out, err = capman.read_global_capture()
sys.stdout.write(out)
sys.stderr.write(err)
@ -96,6 +97,8 @@ class CaptureManager(object):
else:
raise ValueError("unknown capturing method: %r" % method)
# Global capturing control
def start_global_capturing(self):
assert self._global_capturing is None
self._global_capturing = self._getcapture(self._method)
@ -110,29 +113,15 @@ class CaptureManager(object):
def resume_global_capture(self):
self._global_capturing.resume_capturing()
def suspend_global_capture(self, item=None, in_=False):
if item is not None:
self.deactivate_fixture(item)
def suspend_global_capture(self, in_=False):
cap = getattr(self, "_global_capturing", None)
if cap is not None:
try:
outerr = cap.readouterr()
finally:
cap.suspend_capturing(in_=in_)
return outerr
cap.suspend_capturing(in_=in_)
@contextlib.contextmanager
def global_and_fixture_disabled(self):
"""Context manager to temporarily disables global and current fixture capturing."""
# Need to undo local capsys-et-al if exists before disabling global capture
fixture = getattr(self._current_item, "_capture_fixture", None)
ctx_manager = fixture._suspend() if fixture else dummy_context_manager()
with ctx_manager:
self.suspend_global_capture(item=None, in_=False)
try:
yield
finally:
self.resume_global_capture()
def read_global_capture(self):
return self._global_capturing.readouterr()
# Fixture Control (its just forwarding, think about removing this later)
def activate_fixture(self, item):
"""If the current item is using ``capsys`` or ``capfd``, activate them so they take precedence over
@ -148,12 +137,53 @@ class CaptureManager(object):
if fixture is not None:
fixture.close()
def suspend_fixture(self, item):
fixture = getattr(item, "_capture_fixture", None)
if fixture is not None:
fixture._suspend()
def resume_fixture(self, item):
fixture = getattr(item, "_capture_fixture", None)
if fixture is not None:
fixture._resume()
# Helper context managers
@contextlib.contextmanager
def global_and_fixture_disabled(self):
"""Context manager to temporarily disables global and current fixture capturing."""
# Need to undo local capsys-et-al if exists before disabling global capture
self.suspend_fixture(self._current_item)
self.suspend_global_capture(in_=False)
try:
yield
finally:
self.resume_global_capture()
self.resume_fixture(self._current_item)
@contextlib.contextmanager
def item_capture(self, when, item):
self.resume_global_capture()
self.activate_fixture(item)
try:
yield
finally:
self.deactivate_fixture(item)
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
out, err = self.suspend_global_capture()
self.suspend_global_capture()
out, err = self.read_global_capture()
rep = outcome.get_result()
if out:
rep.sections.append(("Captured stdout", out))
@ -163,35 +193,25 @@ class CaptureManager(object):
yield
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_setup(self, item):
def pytest_runtest_protocol(self, item):
self._current_item = item
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.suspend_capture_item(item, "setup")
self._current_item = None
@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):
self._current_item = 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.suspend_capture_item(item, "call")
self._current_item = None
with self.item_capture("call", item):
yield
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_teardown(self, item):
self._current_item = item
self.resume_global_capture()
self.activate_fixture(item)
yield
self.suspend_capture_item(item, "teardown")
self._current_item = None
with self.item_capture("teardown", item):
yield
@pytest.hookimpl(tryfirst=True)
def pytest_keyboard_interrupt(self, excinfo):
@ -201,11 +221,6 @@ class CaptureManager(object):
def pytest_internalerror(self, excinfo):
self.stop_global_capturing()
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)
capture_fixtures = {"capfd", "capfdbinary", "capsys", "capsysbinary"}
@ -314,10 +329,12 @@ class CaptureFixture(object):
self._captured_err = self.captureclass.EMPTY_BUFFER
def _start(self):
self._capture = MultiCapture(
out=True, err=True, in_=False, Capture=self.captureclass
)
self._capture.start_capturing()
# Start if not started yet
if getattr(self, "_capture", None) 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:
@ -341,14 +358,13 @@ class CaptureFixture(object):
self._captured_err = self.captureclass.EMPTY_BUFFER
return CaptureResult(captured_out, captured_err)
@contextlib.contextmanager
def _suspend(self):
"""Suspends this fixture's own capturing temporarily."""
self._capture.suspend_capturing()
try:
yield
finally:
self._capture.resume_capturing()
def _resume(self):
"""Resumes this fixture's own capturing temporarily."""
self._capture.resume_capturing()
@contextlib.contextmanager
def disabled(self):

3
src/_pytest/debugging.py
View File

@ -102,7 +102,8 @@ class PdbInvoke(object):
def pytest_exception_interact(self, node, call, report):
capman = node.config.pluginmanager.getplugin("capturemanager")
if capman:
out, err = capman.suspend_global_capture(in_=True)
capman.suspend_global_capture(in_=True)
out, err = capman.read_global_capture()
sys.stdout.write(out)
sys.stdout.write(err)
_enter_pdb(node, call.excinfo, report)

3
src/_pytest/setuponly.py
View File

@ -51,7 +51,8 @@ def _show_fixture_action(fixturedef, msg):
config = fixturedef._fixturemanager.config
capman = config.pluginmanager.getplugin("capturemanager")
if capman:
out, err = capman.suspend_global_capture()
capman.suspend_global_capture()
out, err = capman.read_global_capture()
tw = config.get_terminal_writer()
tw.line()

5
testing/logging/test_reporting.py
View File

@ -878,7 +878,6 @@ def test_live_logging_suspends_capture(has_capture_manager, request):
import logging
import contextlib
from functools import partial
from _pytest.capture import CaptureManager
from _pytest.logging import _LiveLoggingStreamHandler
class MockCaptureManager:
@ -890,10 +889,6 @@ def test_live_logging_suspends_capture(has_capture_manager, request):
yield
self.calls.append("exit disabled")
# sanity check
assert CaptureManager.suspend_capture_item
assert CaptureManager.resume_global_capture
class DummyTerminal(six.StringIO):
def section(self, *args, **kwargs):
pass

89
testing/test_capture.py
View File

@ -70,19 +70,23 @@ class TestCaptureManager(object):
try:
capman = CaptureManager(method)
capman.start_global_capturing()
outerr = capman.suspend_global_capture()
capman.suspend_global_capture()
outerr = capman.read_global_capture()
assert outerr == ("", "")
outerr = capman.suspend_global_capture()
capman.suspend_global_capture()
outerr = capman.read_global_capture()
assert outerr == ("", "")
print("hello")
out, err = capman.suspend_global_capture()
capman.suspend_global_capture()
out, err = capman.read_global_capture()
if method == "no":
assert old == (sys.stdout, sys.stderr, sys.stdin)
else:
assert not out
capman.resume_global_capture()
print("hello")
out, err = capman.suspend_global_capture()
capman.suspend_global_capture()
out, err = capman.read_global_capture()
if method != "no":
assert out == "hello\n"
capman.stop_global_capturing()
@ -1415,9 +1419,69 @@ def test_pickling_and_unpickling_encoded_file():
pickle.loads(ef_as_str)
def test_capsys_with_cli_logging(testdir):
def test_global_capture_with_live_logging(testdir):
# Issue 3819
# capsys should work with real-time cli logging
# capture should work with live cli logging
# Teardown report seems to have the capture for the whole process (setup, capture, teardown)
testdir.makeconftest(
"""
def pytest_runtest_logreport(report):
if "test_global" in report.nodeid:
if report.when == "teardown":
with open("caplog", "w") as f:
f.write(report.caplog)
with open("capstdout", "w") as f:
f.write(report.capstdout)
"""
)
testdir.makepyfile(
"""
import logging
import sys
import pytest
logger = logging.getLogger(__name__)
@pytest.fixture
def fix1():
print("fix setup")
logging.info("fix setup")
yield
logging.info("fix teardown")
print("fix teardown")
def test_global(fix1):
print("begin test")
logging.info("something in test")
print("end test")
"""
)
result = testdir.runpytest_subprocess("--log-cli-level=INFO")
assert result.ret == 0
with open("caplog", "r") as f:
caplog = f.read()
assert "fix setup" in caplog
assert "something in test" in caplog
assert "fix teardown" in caplog
with open("capstdout", "r") as f:
capstdout = f.read()
assert "fix setup" in capstdout
assert "begin test" in capstdout
assert "end test" in capstdout
assert "fix teardown" in capstdout
@pytest.mark.parametrize("capture_fixture", ["capsys", "capfd"])
def test_capture_with_live_logging(testdir, capture_fixture):
# Issue 3819
# capture should work with live cli logging
testdir.makepyfile(
"""
import logging
@ -1425,22 +1489,23 @@ def test_capsys_with_cli_logging(testdir):
logger = logging.getLogger(__name__)
def test_myoutput(capsys): # or use "capfd" for fd-level
def test_capture({0}):
print("hello")
sys.stderr.write("world\\n")
captured = capsys.readouterr()
captured = {0}.readouterr()
assert captured.out == "hello\\n"
assert captured.err == "world\\n"
logging.info("something")
print("next")
logging.info("something")
captured = capsys.readouterr()
captured = {0}.readouterr()
assert captured.out == "next\\n"
"""
""".format(
capture_fixture
)
)
result = testdir.runpytest_subprocess("--log-cli-level=INFO")
assert result.ret == 0