-
Notifications
You must be signed in to change notification settings - Fork 4
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
WIP: Flexible instantiation #64
base: master
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -272,13 +272,39 @@ class SafeCloseTeeStdin(_SafeCloseIOBase, TeeStdin): | |
""" | ||
|
||
|
||
class _MultiCloseContextManager(tuple, AbstractContextManager): | ||
class TupleContextManager(tuple, AbstractContextManager): | ||
"""Base for context managers that are also a tuple.""" | ||
|
||
# This is needed to establish a workable MRO. | ||
|
||
|
||
class StdioTuple(TupleContextManager): | ||
"""Tuple context manager of stdin, stdout and stderr streams.""" | ||
|
||
def __new__(cls, iterable): | ||
"""Instantiate new tuple from iterable constaining three TextIOBase.""" | ||
items = list(iterable) | ||
assert len(items) == 3 # noqa: S101 | ||
# pytest and colorama break this assertion when applied to the sys.foo | ||
# when they replace them with custom objects, and probably many other | ||
# similar tools do the same | ||
# assert all(isinstance(item, TextIOBase) for item in items) # noqa: E800 | ||
|
||
return super(StdioTuple, cls).__new__(cls, items) | ||
|
||
|
||
class _MultiCloseContextManager(TupleContextManager): | ||
"""Manage multiple closable members of a tuple.""" | ||
|
||
def __enter__(self): | ||
"""Enter context of all members.""" | ||
with ExitStack() as stack: | ||
all(map(stack.enter_context, self)) | ||
# If not all items are TextIOBase, they may not have __exit__ | ||
for item in self: | ||
try: | ||
stack.enter_context(item) | ||
except AttributeError: | ||
pass | ||
|
||
self._close_files = stack.pop_all().close | ||
|
||
|
@@ -289,17 +315,8 @@ def __exit__(self, exc_type, exc_value, traceback): | |
self._close_files() | ||
|
||
|
||
class StdioManager(_MultiCloseContextManager): | ||
r"""Substitute temporary text buffers for `stdio` in a managed context. | ||
|
||
Context manager. | ||
|
||
Substitutes empty :class:`RandomTextIO`\ s for | ||
:obj:`sys.stdout` and :obj:`sys.stderr`, | ||
and a :class:`TeeStdin` for :obj:`sys.stdin` within the managed context. | ||
|
||
Upon exiting the context, the original stream objects are restored | ||
within :mod:`sys`, and the temporary streams are closed. | ||
class FakeStdioTuple(StdioTuple, _MultiCloseContextManager): | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. the There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Ah, the important part is that dual-inheritance is possible, so others could do it if they wanted to use |
||
"""Tuple of stdin, stdout and stderr streams. | ||
|
||
Parameters | ||
---------- | ||
|
@@ -328,7 +345,7 @@ class StdioManager(_MultiCloseContextManager): | |
""" | ||
|
||
def __new__(cls, in_str="", close=True): | ||
"""Instantiate new context manager that emulates namedtuple.""" | ||
"""Instantiate new tuple of fake stdin, stdout and stderr.""" | ||
if close: | ||
out_cls = SafeCloseRandomTextIO | ||
in_cls = SafeCloseTeeStdin | ||
|
@@ -340,7 +357,37 @@ def __new__(cls, in_str="", close=True): | |
stderr = out_cls() | ||
stdin = in_cls(stdout, in_str) | ||
|
||
self = super(StdioManager, cls).__new__(cls, [stdin, stdout, stderr]) | ||
return super(FakeStdioTuple, cls).__new__(cls, [stdin, stdout, stderr]) | ||
|
||
|
||
class CurrentSysIoStreams(StdioTuple): | ||
"""Tuple of current stdin, stdout and stderr streams.""" | ||
|
||
def __new__(cls): | ||
"""Instantiate new tuple of current sys.stdin, sys.stdout and sys.stderr.""" | ||
items = [sys.stdin, sys.stdout, sys.stderr] | ||
return super(CurrentSysIoStreams, cls).__new__(cls, items) | ||
|
||
|
||
class StdioManager(_MultiCloseContextManager): | ||
r"""Substitute temporary text buffers for `stdio` in a managed context. | ||
|
||
Context manager. | ||
|
||
Substitutes empty :class:`RandomTextIO`\ s for | ||
:obj:`sys.stdout` and :obj:`sys.stderr`, | ||
and a :class:`TeeStdin` for :obj:`sys.stdin` within the managed context. | ||
|
||
Upon exiting the context, the original stream objects are restored | ||
within :mod:`sys`, and the temporary streams are closed. | ||
""" | ||
|
||
def __new__(cls, in_str="", close=True, streams=None): | ||
"""Instantiate new context manager for streams.""" | ||
if not streams: | ||
streams = FakeStdioTuple(in_str, close) | ||
|
||
self = super(StdioManager, cls).__new__(cls, streams) | ||
|
||
self._close = close | ||
|
||
|
@@ -363,7 +410,7 @@ def stderr(self): | |
|
||
def __enter__(self): | ||
"""Enter context, replacing sys stdio objects with capturing streams.""" | ||
self._prior_streams = (sys.stdin, sys.stdout, sys.stderr) | ||
self._prior_streams = CurrentSysIoStreams() | ||
|
||
super().__enter__() | ||
|
||
|
@@ -380,3 +427,4 @@ def __exit__(self, exc_type, exc_value, traceback): | |
|
||
|
||
stdio_mgr = StdioManager | ||
_INITIAL_SYS_STREAMS = StdioTuple([sys.stdin, sys.stdout, sys.stderr]) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. These should be There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Ah, I hadn't thought about using those, but yeah -- definitely helpful. It's not a carte-blanche thing, though. We can reference back to those in some cases -- but in general, there may be situations where we might want to restore to the prior I think testing That actually may be an argument against trying to directly reach within There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Yes, this is the dangerous territory that #78 is going in -- IMO the 'dont touch' is cause for concern, lots of it, especially in multi-threading as the tinkering being done isnt atomic, and locking is going to be hard. #78 may need to be opt-in for a while. |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -7,7 +7,7 @@ envlist= | |
|
||
[testenv] | ||
commands= | ||
pytest -p no:warnings | ||
pytest -p no:warnings -s | ||
deps= | ||
pytest | ||
|
||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
this
-s
can be removed, because of the commented out assertion inStdioTuple.__new__
.the alternatives include:
_INITIAL_SYS_STREAMS
and re-add the assertion, which requires the caller to disable any incompatible wrappers before using stdio-mgr__new__
but only emit a warning, and let the user encounter real breakages somewhere else, possibly quite confusing. (this appears to be theio
approach, as we saw withbuffer=StringIO()
not breaking during instantiation)There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Disabling pytest capturing from within the test suite is given a small note at the bottom of https://docs.pytest.org/en/latest/capture.html .
There was also talk of it being possible using markers pytest-dev/pytest#1599 (comment)
And it seems class scope disabling can be achieved using the pytest plugin manager pytest-dev/pytest#1599 (comment)
(If we are going to expect users to uncapture pytest, https://docs.pytest.org/en/latest/capture.html needs a rewrite to assist them.)