Skip to content

Commit

Permalink
[4.6] capsys: ensure fd is unbuffered
Browse files Browse the repository at this point in the history
  • Loading branch information
blueyed committed Oct 26, 2019
1 parent 3edf417 commit cb80252
Show file tree
Hide file tree
Showing 3 changed files with 66 additions and 0 deletions.
1 change: 1 addition & 0 deletions changelog/5134.bugfix.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Fix buffering with capsys fixture on Python 2.
14 changes: 14 additions & 0 deletions src/_pytest/capture.py
Original file line number Diff line number Diff line change
Expand Up @@ -685,11 +685,25 @@ 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")
and not sys.platform.startswith("win32")
):
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):
Expand Down
51 changes: 51 additions & 0 deletions testing/test_capture.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

0 comments on commit cb80252

Please sign in to comment.