diff --git a/changelog/5134.bugfix.rst b/changelog/5134.bugfix.rst new file mode 100644 index 00000000000..ffa40abf92d --- /dev/null +++ b/changelog/5134.bugfix.rst @@ -0,0 +1 @@ +Fix buffering with capsys fixture on Python 2. diff --git a/src/_pytest/capture.py b/src/_pytest/capture.py index 68c17772f39..8a8d8617b40 100644 --- a/src/_pytest/capture.py +++ b/src/_pytest/capture.py @@ -685,11 +685,24 @@ def done(self): self._state = "done" def suspend(self): + if ( + six.PY2 + and self.name in ("stdout", "stderr") + and getattr(sys, "__{}__".format(self.name)) is self._old + and not hasattr(self, "_fdopen_done") + ): + self._fdopen_done = True + try: + # Ensure fd is unbuffered (#5134). + self._old = os.fdopen(self._old.fileno(), "wb+", 0) + except (UnsupportedOperation, OSError): + pass setattr(sys, self.name, self._old) self._state = "suspended" def resume(self): setattr(sys, self.name, self.tmpfile) + # setattr(sys, self.name, os.fdopen(self.tmpfile.fileno(), "w", 0)) self._state = "resumed" def writeorg(self, data): diff --git a/testing/test_capture.py b/testing/test_capture.py index 01123bf5b40..d12eae194a2 100644 --- a/testing/test_capture.py +++ b/testing/test_capture.py @@ -1577,3 +1577,54 @@ def test_fails(): ) else: assert result_with_capture.ret == 0 + + +def test_syscapture_is_unbuffered_when_suspended(testdir, LineMatcher): + import time + + stampfile = testdir.tmpdir.join("stampfile") + testdir.makepyfile( + **{ + "conftest.py": """ + import ctypes + + libc = ctypes.CDLL(None) + libc.puts(b'this comes from C via conftest') + """, + "test_pass.py": """ + import os + import time + + def test_capfd(capfd): + print("test_capfd_start") + with capfd.disabled(): + for i in range(0, 50): + print("test_capfd_loop: %d" % i) + time.sleep(0.1) + if os.path.exists({stampfile!r}): + break + """.format( + stampfile=str(stampfile) + ), + } + ) + + child = testdir.spawn_pytest("-s --color=no -vv test_pass.py") + start = time.time() + child.expect_exact("test_capfd_loop: 0\r\n") + duration = time.time() - start + stampfile.ensure() + out = child.before + child.buffer + child.after + + out += child.read() + lm = LineMatcher(out.decode().splitlines()) + lm.fnmatch_lines( + [ + "this comes from C via conftest", + "test_pass.py::test_capfd test_capfd_loop: 0", + "*= 1 passed in *", + ] + ) + assert duration < 5 + child.wait() + assert child.exitstatus == 0