Skip to content

Commit

Permalink
Simplify away safe_text_dupfile/EncodedFile
Browse files Browse the repository at this point in the history
I tried to understand what the `safe_text_dupfile()` function and
`EncodedFile` class do. Outside tests, `EncodedFile` is only used by
`safe_text_dupfile`, and `safe_text_dupfile` is only used by
`FDCaptureBinary.__init__()`. I then started to eliminate always-true
conditions based on the single call site, and in the end nothing was
left.

The old code had some workarounds:

- Ensure `errors` property is set: pytest-dev#555.
  It is set now, although now it will return `replace`, which is the
  actual method used also before, instead of `strict`. AFAIU, the issue
  just wanted it to be set to something, not `strict` specifically.

- Infinite recursion trying to pickle `EncodedFile` (f7282b8).
  Not sure, I don't think this object should be pickle-able at all (and
  it isn't...).

- A bug in `logging.Handler.__repr__` with int `file.name`.
  pytest-dev#2555
  Was fixed in Python>=3.7.4: https://bugs.python.org/issue36015.
  TODO: This workaround is probably still needed?
  • Loading branch information
bluetech committed Mar 7, 2020
1 parent 580edc1 commit 60f5531
Show file tree
Hide file tree
Showing 2 changed files with 8 additions and 129 deletions.
65 changes: 6 additions & 59 deletions src/_pytest/capture.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,9 +9,7 @@
import sys
from io import UnsupportedOperation
from tempfile import TemporaryFile
from typing import BinaryIO
from typing import Generator
from typing import Iterable
from typing import Optional

import pytest
Expand Down Expand Up @@ -390,55 +388,6 @@ def disabled(self):
yield


def safe_text_dupfile(f, mode, default_encoding="UTF8"):
""" return an open text file object that's a duplicate of f on the
FD-level if possible.
"""
encoding = getattr(f, "encoding", None)
try:
fd = f.fileno()
except Exception:
if "b" not in getattr(f, "mode", "") and hasattr(f, "encoding"):
# we seem to have a text stream, let's just use it
return f
else:
newfd = os.dup(fd)
if "b" not in mode:
mode += "b"
f = os.fdopen(newfd, mode, 0) # no buffering
return EncodedFile(f, encoding or default_encoding)


class EncodedFile:
errors = "strict" # possibly needed by py3 code (issue555)

def __init__(self, buffer: BinaryIO, encoding: str) -> None:
self.buffer = buffer
self.encoding = encoding

def write(self, s: str) -> int:
if not isinstance(s, str):
raise TypeError(
"write() argument must be str, not {}".format(type(s).__name__)
)
return self.buffer.write(s.encode(self.encoding, "replace"))

def writelines(self, lines: Iterable[str]) -> None:
self.buffer.writelines(x.encode(self.encoding, "replace") for x in lines)

@property
def name(self) -> str:
"""Ensure that file.name is a string."""
return repr(self.buffer)

@property
def mode(self) -> str:
return self.buffer.mode.replace("b", "")

def __getattr__(self, name):
return getattr(object.__getattribute__(self, "buffer"), name)


CaptureResult = collections.namedtuple("CaptureResult", ["out", "err"])


Expand Down Expand Up @@ -548,9 +497,7 @@ def __init__(self, targetfd, tmpfile=None):
self.syscapture = SysCapture(targetfd)
else:
if tmpfile is None:
f = TemporaryFile()
with f:
tmpfile = safe_text_dupfile(f, mode="wb+")
tmpfile = TemporaryFile("wt+", encoding="utf-8", errors="replace")
if targetfd in patchsysdict:
self.syscapture = SysCapture(targetfd, tmpfile)
else:
Expand Down Expand Up @@ -579,7 +526,7 @@ def _start(self):

def snap(self):
self.tmpfile.seek(0)
res = self.tmpfile.read()
res = self.tmpfile.buffer.read()
self.tmpfile.seek(0)
self.tmpfile.truncate()
return res
Expand Down Expand Up @@ -621,10 +568,10 @@ class FDCapture(FDCaptureBinary):
EMPTY_BUFFER = str() # type: ignore

def snap(self):
res = super().snap()
enc = getattr(self.tmpfile, "encoding", None)
if enc and isinstance(res, bytes):
res = str(res, enc, "replace")
self.tmpfile.seek(0)
res = self.tmpfile.read()
self.tmpfile.seek(0)
self.tmpfile.truncate()
return res


Expand Down
72 changes: 2 additions & 70 deletions testing/test_capture.py
Original file line number Diff line number Diff line change
@@ -1,16 +1,12 @@
import contextlib
import io
import os
import pickle
import subprocess
import sys
import textwrap
from io import StringIO
from io import UnsupportedOperation
from typing import BinaryIO
from typing import Generator
from typing import List
from typing import TextIO

import pytest
from _pytest import capture
Expand Down Expand Up @@ -865,49 +861,6 @@ def tmpfile(testdir) -> Generator[BinaryIO, None, None]:
f.close()


@needsosdup
def test_dupfile(tmpfile) -> None:
flist = [] # type: List[TextIO]
for i in range(5):
nf = capture.safe_text_dupfile(tmpfile, "wb")
assert nf != tmpfile
assert nf.fileno() != tmpfile.fileno()
assert nf not in flist
print(i, end="", file=nf)
flist.append(nf)

fname_open = flist[0].name
assert fname_open == repr(flist[0].buffer)

for i in range(5):
f = flist[i]
f.close()
fname_closed = flist[0].name
assert fname_closed == repr(flist[0].buffer)
assert fname_closed != fname_open
tmpfile.seek(0)
s = tmpfile.read()
assert "01234" in repr(s)
tmpfile.close()
assert fname_closed == repr(flist[0].buffer)


def test_dupfile_on_bytesio():
bio = io.BytesIO()
f = capture.safe_text_dupfile(bio, "wb")
f.write("hello")
assert bio.getvalue() == b"hello"
assert "BytesIO object" in f.name


def test_dupfile_on_textio():
sio = StringIO()
f = capture.safe_text_dupfile(sio, "wb")
f.write("hello")
assert sio.getvalue() == "hello"
assert not hasattr(f, "name")


@contextlib.contextmanager
def lsof_check():
pid = os.getpid()
Expand Down Expand Up @@ -1355,8 +1308,8 @@ def test_error_attribute_issue555(testdir):
"""
import sys
def test_capattr():
assert sys.stdout.errors == "strict"
assert sys.stderr.errors == "strict"
assert sys.stdout.errors == "replace"
assert sys.stderr.errors == "replace"
"""
)
reprec = testdir.inline_run()
Expand Down Expand Up @@ -1431,15 +1384,6 @@ def test_spam_in_thread():
result.stdout.no_fnmatch_line("*IOError*")


def test_pickling_and_unpickling_encoded_file():
# See https://bitbucket.org/pytest-dev/pytest/pull-request/194
# pickle.loads() raises infinite recursion if
# EncodedFile.__getattr__ is not implemented properly
ef = capture.EncodedFile(None, None)
ef_as_str = pickle.dumps(ef)
pickle.loads(ef_as_str)


def test_global_capture_with_live_logging(testdir):
# Issue 3819
# capture should work with live cli logging
Expand Down Expand Up @@ -1555,18 +1499,6 @@ def test_stderr_write_returns_len(capsys):
assert sys.stderr.write("Foo") == 3


def test_encodedfile_writelines(tmpfile: BinaryIO) -> None:
ef = capture.EncodedFile(tmpfile, "utf-8")
with pytest.raises(AttributeError):
ef.writelines([b"line1", b"line2"]) # type: ignore[list-item] # noqa: F821
assert ef.writelines(["line1", "line2"]) is None # type: ignore[func-returns-value] # noqa: F821
tmpfile.seek(0)
assert tmpfile.read() == b"line1line2"
tmpfile.close()
with pytest.raises(ValueError):
ef.read()


def test__get_multicapture() -> None:
assert isinstance(_get_multicapture("fd"), MultiCapture)
pytest.raises(ValueError, _get_multicapture, "unknown").match(
Expand Down

0 comments on commit 60f5531

Please sign in to comment.