Add capfdbinary fixture

`capfdbinary` works like `capfd` but produces bytes for `readouterr()`.
This commit is contained in:
Anthony Sottile 2017-11-14 14:08:23 -08:00
parent 685387a43e
commit 8f90812481
4 changed files with 94 additions and 26 deletions

View File

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

2
changelog/2923.feature Normal file
View File

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

View File

@ -85,9 +85,9 @@ of the failing function and hide the other one::
Accessing captured output from a test function Accessing captured output from a test function
--------------------------------------------------- ---------------------------------------------------
The ``capsys`` and ``capfd`` fixtures allow to access stdout/stderr The ``capsys``, ``capfd``, and ``capfdbinary`` fixtures allow access to
output created during test execution. Here is an example test function stdout/stderr output created during test execution. Here is an example test
that performs some output related checks: function that performs some output related checks:
.. code-block:: python .. code-block:: python
@ -110,11 +110,17 @@ output streams and also interacts well with pytest's
own per-test capturing. own per-test capturing.
If you want to capture on filedescriptor level you can use 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 same interface but allows to also capture output from
libraries or subprocesses that directly write to operating libraries or subprocesses that directly write to operating
system level output streams (FD1 and FD2). 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 .. versionadded:: 3.0

View File

@ -398,7 +398,7 @@ class TestCaptureFixture(object):
result = testdir.runpytest(p) result = testdir.runpytest(p)
result.stdout.fnmatch_lines([ result.stdout.fnmatch_lines([
"*ERROR*setup*test_one*", "*ERROR*setup*test_one*",
"E*capsys*capfd*same*time*", "E*capfd*capsys*same*time*",
"*ERROR*setup*test_two*", "*ERROR*setup*test_two*",
"E*capsys*capfd*same*time*", "E*capsys*capfd*same*time*",
"*2 error*"]) "*2 error*"])
@ -418,10 +418,21 @@ class TestCaptureFixture(object):
"*test_one*", "*test_one*",
"*capsys*capfd*same*time*", "*capsys*capfd*same*time*",
"*test_two*", "*test_two*",
"*capsys*capfd*same*time*", "*capfd*capsys*same*time*",
"*2 failed in*", "*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"]) @pytest.mark.parametrize("method", ["sys", "fd"])
def test_capture_is_represented_on_failure_issue128(self, testdir, method): def test_capture_is_represented_on_failure_issue128(self, testdir, method):
p = testdir.makepyfile(""" p = testdir.makepyfile("""
@ -446,6 +457,19 @@ class TestCaptureFixture(object):
""") """)
reprec.assertoutcome(passed=1) 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): def test_partial_setup_failure(self, testdir):
p = testdir.makepyfile(""" p = testdir.makepyfile("""
def test_hello(capsys, missingarg): def test_hello(capsys, missingarg):