Skip to content

[WIP] Refactor _getcapture #6671

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 1 commit into from
Closed
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
66 changes: 43 additions & 23 deletions src/_pytest/capture.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,13 +13,18 @@
from typing import Generator
from typing import Iterable
from typing import Optional
from typing import Union

import pytest
from _pytest.compat import CaptureAndPassthroughIO
from _pytest.compat import CaptureIO
from _pytest.compat import PassthroughCaptureIO
from _pytest.compat import TYPE_CHECKING
from _pytest.config import Config
from _pytest.fixtures import FixtureRequest

if TYPE_CHECKING:
from typing import Type

patchsysdict = {0: "stdin", 1: "stdout", 2: "stderr"}


Expand Down Expand Up @@ -89,16 +94,31 @@ def __repr__(self):
self._method, self._global_capturing, self._capture_fixture
)

def _getcapture(self, method):
if method == "fd":
return MultiCapture(out=True, err=True, Capture=FDCapture)
elif method == "sys":
return MultiCapture(out=True, err=True, Capture=SysCapture)
elif method == "no":
def _getcapture(self, method: str) -> "MultiCapture":
modes = method.split("-")

if "no" in modes:
if len(modes) > 1:
raise ValueError("'no' cannot be combined with other modes")
return MultiCapture(out=False, err=False, in_=False)
elif method == "tee-sys":
return MultiCapture(out=True, err=True, in_=False, Capture=TeeSysCapture)
raise ValueError("unknown capturing method: %r" % method) # pragma: no cover

if "tee" in modes:
if "sys" not in modes:
raise ValueError("'tee' only works with 'sys'")

capture = TeeSysCapture # type: Union[Type[Capture]]
elif "fd" in modes:
if "sys" in modes:
raise ValueError("'fd' and 'sys' cannot be combined")
capture = FDCapture
elif "sys" in modes:
if "fd" in modes:
raise ValueError("'sys' and 'fd' cannot be combined")
capture = SysCapture
else:
raise ValueError("unknown capturing method: {}".format(method))

return MultiCapture(out=True, err=True, Capture=capture)

def is_capturing(self):
if self.is_globally_capturing():
Expand Down Expand Up @@ -512,7 +532,11 @@ class NoCapture:
__init__ = start = done = suspend = resume = lambda *args: None


class FDCaptureBinary:
class Capture:
pass
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This empty class is here for type hinting only? Perhaps adding a quick one line docstring mentioning that?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, or remove it.
This certainly needs another pass.
I've thought about pulling out the renaming/moving of the new classes only for now, given that combinations are not really needed currently anyway (also because of the "choices" validation).

It might make sense to pick that part into #6690, but there it appears to be unclear still how to do it (at least to me). Will try/look into that later, but might take some time.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK, thanks



class FDCaptureBinary(Capture):
"""Capture IO to/from a given os-level filedescriptor.

snap() produces `bytes`
Expand Down Expand Up @@ -613,8 +637,7 @@ def snap(self):
return res


class SysCapture:

class SysCapture(Capture):
EMPTY_BUFFER = str()
_state = None

Expand Down Expand Up @@ -664,16 +687,13 @@ def writeorg(self, data):


class TeeSysCapture(SysCapture):
def __init__(self, fd, tmpfile=None):
name = patchsysdict[fd]
self._old = getattr(sys, name)
self.name = name
if tmpfile is None:
if name == "stdin":
tmpfile = DontReadFromInput()
else:
tmpfile = CaptureAndPassthroughIO(self._old)
self.tmpfile = tmpfile
def __init__(self, fd: int) -> None:
old = getattr(sys, patchsysdict[fd])
if fd == 0:
super().__init__(fd)
else:
super().__init__(fd, PassthroughCaptureIO(old))
assert self._old == old, (self._old, old)


class SysCaptureBinary(SysCapture):
Expand Down
2 changes: 1 addition & 1 deletion src/_pytest/compat.py
Original file line number Diff line number Diff line change
Expand Up @@ -380,7 +380,7 @@ def getvalue(self) -> str:
return self.buffer.getvalue().decode("UTF-8")


class CaptureAndPassthroughIO(CaptureIO):
class PassthroughCaptureIO(CaptureIO):
def __init__(self, other: IO) -> None:
self._other = other
super().__init__()
Expand Down
6 changes: 3 additions & 3 deletions testing/test_capture.py
Original file line number Diff line number Diff line change
Expand Up @@ -822,10 +822,10 @@ def test_write_bytes_to_buffer(self):
assert f.getvalue() == "foo\r\n"


class TestCaptureAndPassthroughIO(TestCaptureIO):
class TestPassthroughCaptureIO(TestCaptureIO):
def test_text(self):
sio = io.StringIO()
f = capture.CaptureAndPassthroughIO(sio)
f = capture.PassthroughCaptureIO(sio)
f.write("hello")
s1 = f.getvalue()
assert s1 == "hello"
Expand All @@ -836,7 +836,7 @@ def test_text(self):

def test_unicode_and_str_mixture(self):
sio = io.StringIO()
f = capture.CaptureAndPassthroughIO(sio)
f = capture.PassthroughCaptureIO(sio)
f.write("\u00f6")
pytest.raises(TypeError, f.write, b"hello")

Expand Down