capture: revisit/fix __repr__, define _in_suspended (#6749)
This commit is contained in:
parent
2b13a9b95d
commit
fb16d3e27a
|
@ -438,6 +438,7 @@ CaptureResult = collections.namedtuple("CaptureResult", ["out", "err"])
|
||||||
class MultiCapture:
|
class MultiCapture:
|
||||||
out = err = in_ = None
|
out = err = in_ = None
|
||||||
_state = None
|
_state = None
|
||||||
|
_in_suspended = False
|
||||||
|
|
||||||
def __init__(self, out=True, err=True, in_=True, Capture=None):
|
def __init__(self, out=True, err=True, in_=True, Capture=None):
|
||||||
if in_:
|
if in_:
|
||||||
|
@ -449,11 +450,7 @@ class MultiCapture:
|
||||||
|
|
||||||
def __repr__(self):
|
def __repr__(self):
|
||||||
return "<MultiCapture out={!r} err={!r} in_={!r} _state={!r} _in_suspended={!r}>".format(
|
return "<MultiCapture out={!r} err={!r} in_={!r} _state={!r} _in_suspended={!r}>".format(
|
||||||
self.out,
|
self.out, self.err, self.in_, self._state, self._in_suspended,
|
||||||
self.err,
|
|
||||||
self.in_,
|
|
||||||
self._state,
|
|
||||||
getattr(self, "_in_suspended", "<UNSET>"),
|
|
||||||
)
|
)
|
||||||
|
|
||||||
def start_capturing(self):
|
def start_capturing(self):
|
||||||
|
@ -490,9 +487,9 @@ class MultiCapture:
|
||||||
self.out.resume()
|
self.out.resume()
|
||||||
if self.err:
|
if self.err:
|
||||||
self.err.resume()
|
self.err.resume()
|
||||||
if hasattr(self, "_in_suspended"):
|
if self._in_suspended:
|
||||||
self.in_.resume()
|
self.in_.resume()
|
||||||
del self._in_suspended
|
self._in_suspended = False
|
||||||
|
|
||||||
def stop_capturing(self):
|
def stop_capturing(self):
|
||||||
""" stop capturing and reset capturing streams """
|
""" stop capturing and reset capturing streams """
|
||||||
|
@ -555,8 +552,12 @@ class FDCaptureBinary:
|
||||||
self.tmpfile_fd = tmpfile.fileno()
|
self.tmpfile_fd = tmpfile.fileno()
|
||||||
|
|
||||||
def __repr__(self):
|
def __repr__(self):
|
||||||
return "<FDCapture {} oldfd={} _state={!r}>".format(
|
return "<{} {} oldfd={} _state={!r} tmpfile={}>".format(
|
||||||
self.targetfd, getattr(self, "targetfd_save", None), self._state
|
self.__class__.__name__,
|
||||||
|
self.targetfd,
|
||||||
|
getattr(self, "targetfd_save", "<UNSET>"),
|
||||||
|
self._state,
|
||||||
|
hasattr(self, "tmpfile") and repr(self.tmpfile) or "<UNSET>",
|
||||||
)
|
)
|
||||||
|
|
||||||
def _start(self):
|
def _start(self):
|
||||||
|
@ -637,8 +638,12 @@ class SysCapture:
|
||||||
self.tmpfile = tmpfile
|
self.tmpfile = tmpfile
|
||||||
|
|
||||||
def __repr__(self):
|
def __repr__(self):
|
||||||
return "<SysCapture {} _old={!r}, tmpfile={!r} _state={!r}>".format(
|
return "<{} {} _old={} _state={!r} tmpfile={!r}>".format(
|
||||||
self.name, self._old, self.tmpfile, self._state
|
self.__class__.__name__,
|
||||||
|
self.name,
|
||||||
|
hasattr(self, "_old") and repr(self._old) or "<UNSET>",
|
||||||
|
self._state,
|
||||||
|
self.tmpfile,
|
||||||
)
|
)
|
||||||
|
|
||||||
def start(self):
|
def start(self):
|
||||||
|
|
|
@ -1005,6 +1005,18 @@ class TestFDCapture:
|
||||||
cap.done()
|
cap.done()
|
||||||
pytest.raises(AttributeError, cap.suspend)
|
pytest.raises(AttributeError, cap.suspend)
|
||||||
|
|
||||||
|
assert repr(cap) == (
|
||||||
|
"<FDCapture 1 oldfd=<UNSET> _state='done' tmpfile={!r}>".format(
|
||||||
|
cap.tmpfile
|
||||||
|
)
|
||||||
|
)
|
||||||
|
# Should not crash with missing "_old".
|
||||||
|
assert repr(cap.syscapture) == (
|
||||||
|
"<SysCapture stdout _old=<UNSET> _state='done' tmpfile={!r}>".format(
|
||||||
|
cap.syscapture.tmpfile
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
def test_capfd_sys_stdout_mode(self, capfd):
|
def test_capfd_sys_stdout_mode(self, capfd):
|
||||||
assert "b" not in sys.stdout.mode
|
assert "b" not in sys.stdout.mode
|
||||||
|
|
||||||
|
@ -1212,19 +1224,19 @@ class TestStdCaptureFDinvalidFD:
|
||||||
def test_stdout():
|
def test_stdout():
|
||||||
os.close(1)
|
os.close(1)
|
||||||
cap = StdCaptureFD(out=True, err=False, in_=False)
|
cap = StdCaptureFD(out=True, err=False, in_=False)
|
||||||
assert repr(cap.out) == "<FDCapture 1 oldfd=None _state=None>"
|
assert repr(cap.out) == "<FDCapture 1 oldfd=<UNSET> _state=None tmpfile=<UNSET>>"
|
||||||
cap.stop_capturing()
|
cap.stop_capturing()
|
||||||
|
|
||||||
def test_stderr():
|
def test_stderr():
|
||||||
os.close(2)
|
os.close(2)
|
||||||
cap = StdCaptureFD(out=False, err=True, in_=False)
|
cap = StdCaptureFD(out=False, err=True, in_=False)
|
||||||
assert repr(cap.err) == "<FDCapture 2 oldfd=None _state=None>"
|
assert repr(cap.err) == "<FDCapture 2 oldfd=<UNSET> _state=None tmpfile=<UNSET>>"
|
||||||
cap.stop_capturing()
|
cap.stop_capturing()
|
||||||
|
|
||||||
def test_stdin():
|
def test_stdin():
|
||||||
os.close(0)
|
os.close(0)
|
||||||
cap = StdCaptureFD(out=False, err=False, in_=True)
|
cap = StdCaptureFD(out=False, err=False, in_=True)
|
||||||
assert repr(cap.in_) == "<FDCapture 0 oldfd=None _state=None>"
|
assert repr(cap.in_) == "<FDCapture 0 oldfd=<UNSET> _state=None tmpfile=<UNSET>>"
|
||||||
cap.stop_capturing()
|
cap.stop_capturing()
|
||||||
"""
|
"""
|
||||||
)
|
)
|
||||||
|
|
Loading…
Reference in New Issue