Merge pull request #2925 from asottile/capfdbinary

Add capfdbinary fixture

Author: nicoddemus

_pytest/capture.py

@@ -180,17 +180,29 @@ def suspend_capture_item(self, item, when, in_=False):
         item.add_report_section(when, "stderr", err)
 
 
-error_capsysfderror = "cannot use capsys and capfd at the same time"
+capture_fixtures = {'capfd', 'capfdbinary', 'capsys'}
+
+
+def _ensure_only_one_capture_fixture(request, name):
+    fixtures = set(request.fixturenames) & capture_fixtures - set((name,))
+    if fixtures:
+        fixtures = sorted(fixtures)
+        fixtures = fixtures[0] if len(fixtures) == 1 else fixtures
+        raise request.raiseerror(
+            "cannot use {0} and {1} at the same time".format(
+                fixtures, name,
+            ),
+        )
 
 
 @pytest.fixture
 def capsys(request):
     """Enable capturing of writes to sys.stdout/sys.stderr and make
     captured output available via ``capsys.readouterr()`` method calls
-    which return a ``(out, err)`` tuple.
+    which return a ``(out, err)`` tuple.  ``out`` and ``err`` will be ``text``
+    objects.
     """
-    if "capfd" in request.fixturenames:
-        raise request.raiseerror(error_capsysfderror)
+    _ensure_only_one_capture_fixture(request, 'capsys')
     with _install_capture_fixture_on_item(request, SysCapture) as fixture:
         yield fixture
 
@@ -199,16 +211,30 @@ def capsys(request):
 def capfd(request):
     """Enable capturing of writes to file descriptors 1 and 2 and make
     captured output available via ``capfd.readouterr()`` method calls
-    which return a ``(out, err)`` tuple.
+    which return a ``(out, err)`` tuple.  ``out`` and ``err`` will be ``text``
+    objects.
     """
-    if "capsys" in request.fixturenames:
-        request.raiseerror(error_capsysfderror)
+    _ensure_only_one_capture_fixture(request, 'capfd')
     if not hasattr(os, 'dup'):
         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
 
 
+@pytest.fixture
+def capfdbinary(request):
+    """Enable capturing of write to file descriptors 1 and 2 and make
+    captured output available via ``capfdbinary.readouterr`` method calls
+    which return a ``(out, err)`` tuple.  ``out`` and ``err`` will be
+    ``bytes`` objects.
+    """
+    _ensure_only_one_capture_fixture(request, 'capfdbinary')
+    if not hasattr(os, 'dup'):
+        pytest.skip("capfdbinary fixture needs os.dup function which is not available in this system")
+    with _install_capture_fixture_on_item(request, FDCaptureBinary) as fixture:
+        yield fixture
+
+
 @contextlib.contextmanager
 def _install_capture_fixture_on_item(request, capture_class):
     """
@@ -378,8 +404,11 @@ class NoCapture:
     __init__ = start = done = suspend = resume = lambda *args: None
 
 
-class FDCapture:
-    """ Capture IO to/from a given os-level filedescriptor. """
+class FDCaptureBinary:
+    """Capture IO to/from a given os-level filedescriptor.
+
+    snap() produces `bytes`
+    """
 
     def __init__(self, targetfd, tmpfile=None):
         self.targetfd = targetfd
@@ -418,17 +447,11 @@ def start(self):
         self.syscapture.start()
 
     def snap(self):
-        f = self.tmpfile
-        f.seek(0)
-        res = f.read()
-        if res:
-            enc = getattr(f, "encoding", None)
-            if enc and isinstance(res, bytes):
-                res = six.text_type(res, enc, "replace")
-            f.truncate(0)
-            f.seek(0)
-            return res
-        return ''
+        self.tmpfile.seek(0)
+        res = self.tmpfile.read()
+        self.tmpfile.seek(0)
+        self.tmpfile.truncate()
+        return res
 
     def done(self):
         """ stop capturing, restore streams, return original capture file,
@@ -454,6 +477,19 @@ def writeorg(self, data):
         os.write(self.targetfd_save, data)
 
 
+class FDCapture(FDCaptureBinary):
+    """Capture IO to/from a given os-level filedescriptor.
+
+    snap() produces text
+    """
+    def snap(self):
+        res = FDCaptureBinary.snap(self)
+        enc = getattr(self.tmpfile, "encoding", None)
+        if enc and isinstance(res, bytes):
+            res = six.text_type(res, enc, "replace")
+        return res
+
+
 class SysCapture:
     def __init__(self, fd, tmpfile=None):
         name = patchsysdict[fd]

changelog/2923.feature

@@ -0,0 +1,2 @@
+Add ``capfdbinary`` a version of ``capfd`` which returns bytes from
+``readouterr()``.

doc/en/capture.rst

@@ -85,9 +85,9 @@ of the failing function and hide the other one::
 Accessing captured output from a test function
 ---------------------------------------------------
 
-The ``capsys`` and ``capfd`` fixtures allow to access stdout/stderr
-output created during test execution.  Here is an example test function
-that performs some output related checks:
+The ``capsys``, ``capfd``, and ``capfdbinary`` fixtures allow access to
+stdout/stderr output created during test execution.  Here is an example test
+function that performs some output related checks:
 
 .. code-block:: python
 
@@ -110,11 +110,17 @@ output streams and also interacts well with pytest's
 own per-test capturing.
 
 If you want to capture on filedescriptor level you can use
-the ``capfd`` function argument which offers the exact
+the ``capfd`` fixture which offers the exact
 same interface but allows to also capture output from
 libraries or subprocesses that directly write to operating
 system level output streams (FD1 and FD2).
 
+.. versionadded:: 3.3
+
+If the code under test writes non-textual data, you can capture this using
+the ``capfdbinary`` fixture which instead returns ``bytes`` from
+the ``readouterr`` method.
+
 
 .. versionadded:: 3.0
 

testing/test_capture.py

@@ -398,7 +398,7 @@ def test_two(capfd, capsys):
         result = testdir.runpytest(p)
         result.stdout.fnmatch_lines([
             "*ERROR*setup*test_one*",
-            "E*capsys*capfd*same*time*",
+            "E*capfd*capsys*same*time*",
             "*ERROR*setup*test_two*",
             "E*capsys*capfd*same*time*",
             "*2 error*"])
@@ -418,10 +418,21 @@ def test_two(capfd, request):
             "*test_one*",
             "*capsys*capfd*same*time*",
             "*test_two*",
-            "*capsys*capfd*same*time*",
+            "*capfd*capsys*same*time*",
             "*2 failed in*",
         ])
 
+    def test_capsyscapfdbinary(self, testdir):
+        p = testdir.makepyfile("""
+            def test_one(capsys, capfdbinary):
+                pass
+        """)
+        result = testdir.runpytest(p)
+        result.stdout.fnmatch_lines([
+            "*ERROR*setup*test_one*",
+            "E*capfdbinary*capsys*same*time*",
+            "*1 error*"])
+
     @pytest.mark.parametrize("method", ["sys", "fd"])
     def test_capture_is_represented_on_failure_issue128(self, testdir, method):
         p = testdir.makepyfile("""
@@ -446,6 +457,19 @@ def test_hello(capfd):
         """)
         reprec.assertoutcome(passed=1)
 
+    @needsosdup
+    def test_capfdbinary(self, testdir):
+        reprec = testdir.inline_runsource("""
+            def test_hello(capfdbinary):
+                import os
+                # some likely un-decodable bytes
+                os.write(1, b'\\xfe\\x98\\x20')
+                out, err = capfdbinary.readouterr()
+                assert out == b'\\xfe\\x98\\x20'
+                assert err == b''
+        """)
+        reprec.assertoutcome(passed=1)
+
     def test_partial_setup_failure(self, testdir):
         p = testdir.makepyfile("""
             def test_hello(capsys, missingarg):