Merge pull request #6926 from bluetech/fix-capsysbin-print

Fix crash when printing while capsysbinary is active

changelog/6871.bugfix.rst

@@ -0,0 +1 @@
+Fix crash with captured output when using the :fixture:`capsysbinary fixture <capsysbinary>`.

src/_pytest/capture.py

@@ -570,8 +570,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)
 
 
@@ -591,6 +589,11 @@ def snap(self):
         self.tmpfile.truncate()
         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:
 
@@ -642,8 +645,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):
@@ -655,6 +659,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):

testing/test_capture.py

@@ -515,18 +515,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(
@@ -890,7 +912,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")