Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog/6871.bugfix.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Fix crash with captured output when using the :fixture:`capsysbinary fixture <capsysbinary>`.
14 changes: 11 additions & 3 deletions src/_pytest/capture.py
Original file line number Diff line number Diff line change
Expand Up @@ -610,8 +610,6 @@ def resume(self):

def writeorg(self, data):
""" write to original file descriptor. """
if isinstance(data, str):
data = data.encode("utf8") # XXX use encoding of original stream
os.write(self.targetfd_save, data)


Expand All @@ -631,6 +629,11 @@ def snap(self):
res = str(res, enc, "replace")
return res

def writeorg(self, data):
""" write to original file descriptor. """
data = data.encode("utf-8") # XXX use encoding of original stream
os.write(self.targetfd_save, data)


class SysCaptureBinary:

Expand Down Expand Up @@ -682,8 +685,9 @@ def resume(self):
self._state = "resumed"

def writeorg(self, data):
self._old.write(data)
self._old.flush()
self._old.buffer.write(data)
self._old.buffer.flush()


class SysCapture(SysCaptureBinary):
Expand All @@ -695,6 +699,10 @@ def snap(self):
self.tmpfile.truncate()
return res

def writeorg(self, data):
self._old.write(data)
self._old.flush()


class TeeSysCapture(SysCapture):
def __init__(self, fd, tmpfile=None):
Expand Down
36 changes: 29 additions & 7 deletions testing/test_capture.py
Original file line number Diff line number Diff line change
Expand Up @@ -542,18 +542,40 @@ def test_hello(capfdbinary):
reprec.assertoutcome(passed=1)

def test_capsysbinary(self, testdir):
reprec = testdir.inline_runsource(
"""\
p1 = testdir.makepyfile(
r"""
def test_hello(capsysbinary):
import sys
# some likely un-decodable bytes
sys.stdout.buffer.write(b'\\xfe\\x98\\x20')

sys.stdout.buffer.write(b'hello')

# Some likely un-decodable bytes.
sys.stdout.buffer.write(b'\xfe\x98\x20')

sys.stdout.buffer.flush()

# Ensure writing in text mode still works and is captured.
# https://github.com/pytest-dev/pytest/issues/6871
print("world", flush=True)

out, err = capsysbinary.readouterr()
assert out == b'\\xfe\\x98\\x20'
assert out == b'hello\xfe\x98\x20world\n'
assert err == b''

print("stdout after")
print("stderr after", file=sys.stderr)
"""
)
reprec.assertoutcome(passed=1)
result = testdir.runpytest(str(p1), "-rA")
result.stdout.fnmatch_lines(
[
"*- Captured stdout call -*",
"stdout after",
"*- Captured stderr call -*",
"stderr after",
"*= 1 passed in *",
]
)

def test_partial_setup_failure(self, testdir):
p = testdir.makepyfile(
Expand Down Expand Up @@ -977,7 +999,7 @@ def test_writeorg(self, tmpfile):
cap.start()
tmpfile.write(data1)
tmpfile.flush()
cap.writeorg(data2)
cap.writeorg(data2.decode("ascii"))
scap = cap.snap()
cap.done()
assert scap == data1.decode("ascii")
Expand Down