Capture and display stdout/stderr while running subtests

Fix pytest-dev#18

CHANGELOG.rst

@@ -1,6 +1,14 @@
 CHANGELOG
 =========
 
+0.3.0 (2020-01-22)
+------------------
+
+* Dropped support for Python 3.4.
+* ``subtests`` now correctly captures and displays stdout/stderr (`#18`_).
+
+.. _#18: https://github.com/pytest-dev/pytest-subtests/issues/18
+
 0.2.1 (2019-04-04)
 ------------------
 
@@ -11,7 +19,7 @@ CHANGELOG
 0.2.0 (2019-04-03)
 ------------------
 
-* Sub tests are correctly reported with ``pytest-xdist>=1.28``.
+* Subtests are correctly reported with ``pytest-xdist>=1.28``.
 
 0.1.0 (2019-04-01)
 ------------------

pytest_subtests.py

@@ -1,10 +1,13 @@
 import sys
 from contextlib import contextmanager
-from time import time
+from time import monotonic
 
 import attr
 import pytest
 from _pytest._code import ExceptionInfo
+from _pytest.capture import CaptureFixture
+from _pytest.capture import FDCapture
+from _pytest.capture import SysCapture
 from _pytest.outcomes import OutcomeException
 from _pytest.reports import TestReport
 from _pytest.runner import CallInfo
@@ -96,31 +99,81 @@ def subtests(request):
         suspend_capture_ctx = capmam.global_and_fixture_disabled
     else:
         suspend_capture_ctx = nullcontext
-    yield SubTests(request.node.ihook, request.node, suspend_capture_ctx)
+    yield SubTests(request.node.ihook, suspend_capture_ctx, request)
 
 
 @attr.s
 class SubTests(object):
     ihook = attr.ib()
-    item = attr.ib()
     suspend_capture_ctx = attr.ib()
+    request = attr.ib()
+
+    @property
+    def item(self):
+        return self.request.node
+
+    @contextmanager
+    def _capturing_output(self):
+        option = self.request.config.getoption("capture", None)
+
+        # capsys or capfd are active, subtest should not capture
+        capture_fixture_active = hasattr(self.request.node, "_capture_fixture")
+
+        if option == "sys" and not capture_fixture_active:
+            fixture = CaptureFixture(SysCapture, self.request)
+        elif option == "fd" and not capture_fixture_active:
+            fixture = CaptureFixture(FDCapture, self.request)
+        else:
+            fixture = None
+
+        if fixture is not None:
+            fixture._start()
+
+        captured = Captured()
+        try:
+            yield captured
+        finally:
+            if fixture is not None:
+                out, err = fixture.readouterr()
+                fixture.close()
+                captured.out = out
+                captured.err = err
 
     @contextmanager
     def test(self, msg=None, **kwargs):
-        start = time()
+        start = monotonic()
         exc_info = None
-        try:
-            yield
-        except (Exception, OutcomeException):
-            exc_info = ExceptionInfo.from_current()
-        stop = time()
+
+        with self._capturing_output() as captured:
+            try:
+                yield
+            except (Exception, OutcomeException):
+                exc_info = ExceptionInfo.from_current()
+
+        stop = monotonic()
+
         call_info = CallInfo(None, exc_info, start, stop, when="call")
         sub_report = SubTestReport.from_item_and_call(item=self.item, call=call_info)
         sub_report.context = SubTestContext(msg, kwargs.copy())
+
+        captured.update_report(sub_report)
+
         with self.suspend_capture_ctx():
             self.ihook.pytest_runtest_logreport(report=sub_report)
 
 
+@attr.s
+class Captured:
+    out = attr.ib(default="", type=str)
+    err = attr.ib(default="", type=str)
+
+    def update_report(self, report):
+        if self.out:
+            report.sections.append(("Captured stdout call", self.out))
+        if self.err:
+            report.sections.append(("Captured stderr call", self.err))
+
+
 def pytest_report_to_serializable(report):
     if isinstance(report, SubTestReport):
         return report._to_json()

tests/test_subtests.py

@@ -219,3 +219,91 @@ def test_foo(self):
             result.stdout.fnmatch_lines(
                 ["collected 1 item", "* 3 skipped, 1 passed in *"]
             )
+
+
+class TestCapture:
+    def create_file(self, testdir):
+        testdir.makepyfile(
+            """
+                    import sys
+                    def test(subtests):
+                        print()
+                        print('start test')
+
+                        with subtests.test(i='A'):
+                            print("hello stdout A")
+                            print("hello stderr A", file=sys.stderr)
+                            assert 0
+
+                        with subtests.test(i='B'):
+                            print("hello stdout B")
+                            print("hello stderr B", file=sys.stderr)
+                            assert 0
+
+                        print('end test')
+                        assert 0
+                """
+        )
+
+    def test_capturing(self, testdir):
+        self.create_file(testdir)
+        result = testdir.runpytest()
+        result.stdout.fnmatch_lines(
+            [
+                "*__ test (i='A') __*",
+                "*Captured stdout call*",
+                "hello stdout A",
+                "*Captured stderr call*",
+                "hello stderr A",
+                "*__ test (i='B') __*",
+                "*Captured stdout call*",
+                "hello stdout B",
+                "*Captured stderr call*",
+                "hello stderr B",
+                "*__ test __*",
+                "*Captured stdout call*",
+                "start test",
+                "end test",
+            ]
+        )
+
+    def test_no_capture(self, testdir):
+        self.create_file(testdir)
+        result = testdir.runpytest("-s")
+        result.stdout.fnmatch_lines(
+            [
+                "start test",
+                "hello stdout A",
+                "Fhello stdout B",
+                "Fend test",
+                "*__ test (i='A') __*",
+                "*__ test (i='B') __*",
+                "*__ test __*",
+            ]
+        )
+        result.stderr.fnmatch_lines(["hello stderr A", "hello stderr B"])
+
+    @pytest.mark.parametrize("fixture", ["capsys", "capfd"])
+    def test_capture_with_fixture(self, testdir, fixture):
+        testdir.makepyfile(
+            r"""
+            import sys
+
+            def test(subtests, {fixture}):
+                print('start test')
+
+                with subtests.test(i='A'):
+                    print("hello stdout A")
+                    print("hello stderr A", file=sys.stderr)
+
+                out, err = {fixture}.readouterr()
+                assert out == 'start test\nhello stdout A\n'
+                assert err == 'hello stderr A\n'
+        """.format(
+                fixture=fixture
+            )
+        )
+        result = testdir.runpytest()
+        result.stdout.fnmatch_lines(
+            ["*1 passed*",]
+        )