Allow fixtures to use capsys and capfd

_pytest/capture.py

@@ -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 silence_logging_at_shutdown():
     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,47 +87,51 @@ def _getcapture(self, method):
         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()
             finally:
                 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 @@ def pytest_make_collect_report(self, collector):
 
     @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 @@ def readouterr(self):
 
     @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"):

_pytest/debugging.py

@@ -54,7 +54,7 @@ def set_trace(cls):
         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)

_pytest/setuponly.py

@@ -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)
 

changelog/1993.bugfix

@@ -0,0 +1 @@
+Resume output capturing after ``capsys/capfd.disabled()`` context manager.

changelog/2709.bugfix

@@ -0,0 +1 @@
+``capsys`` and ``capfd`` can now be used by other fixtures.

testing/test_capture.py

@@ -78,23 +78,23 @@ def test_capturing_basic_api(self, method):
         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 @@ def test_init_capturing(self):
         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 @@ def test_log(capsys):
         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):