From e29aaff0edd81370aab01193a7f58dbb01f0e8dc Mon Sep 17 00:00:00 2001 From: John Vandenberg Date: Mon, 26 Aug 2019 10:35:25 +0700 Subject: [PATCH 1/2] Flexible instantiation --- .travis.yml | 2 +- src/stdio_mgr/stdio_mgr.py | 73 ++++++++++++++++++++++++++++++-------- tox.ini | 2 +- 3 files changed, 60 insertions(+), 17 deletions(-) diff --git a/.travis.yml b/.travis.yml index d514790..abf45a4 100644 --- a/.travis.yml +++ b/.travis.yml @@ -23,6 +23,6 @@ jobs: script: - python --version - pip list - - pytest --cov=src -p no:warnings + - pytest --cov=src -p no:warnings -s - if (( $PYTHON_MAJOR == 3 && $PYTHON_MINOR == 7 )); then tox -e flake8; else echo "No flake8."; fi - if (( $PYTHON_MAJOR == 3 && $PYTHON_MINOR == 7 )); then codecov; else echo "No codecov."; fi diff --git a/src/stdio_mgr/stdio_mgr.py b/src/stdio_mgr/stdio_mgr.py index bf6e4ce..280f34a 100644 --- a/src/stdio_mgr/stdio_mgr.py +++ b/src/stdio_mgr/stdio_mgr.py @@ -272,7 +272,28 @@ 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): @@ -289,17 +310,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): + """Tuple of stdin, stdout and stderr streams. Parameters ---------- @@ -328,7 +340,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 +352,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 +405,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 +422,4 @@ def __exit__(self, exc_type, exc_value, traceback): stdio_mgr = StdioManager +_INITIAL_SYS_STREAMS = StdioTuple([sys.stdin, sys.stdout, sys.stderr]) diff --git a/tox.ini b/tox.ini index 81b5239..61984a1 100644 --- a/tox.ini +++ b/tox.ini @@ -7,7 +7,7 @@ envlist= [testenv] commands= - pytest -p no:warnings + pytest -p no:warnings -s deps= pytest From fc4952cc3e7fdbe14fdbff2624cb2606c3a2249c Mon Sep 17 00:00:00 2001 From: John Vandenberg Date: Thu, 5 Sep 2019 11:11:02 +0700 Subject: [PATCH 2/2] _MultiCloseContextManager: Allow missing __exit__ --- src/stdio_mgr/stdio_mgr.py | 7 ++++++- tests/test_stdiomgr_base.py | 18 ++++++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/src/stdio_mgr/stdio_mgr.py b/src/stdio_mgr/stdio_mgr.py index 280f34a..5cf4433 100644 --- a/src/stdio_mgr/stdio_mgr.py +++ b/src/stdio_mgr/stdio_mgr.py @@ -299,7 +299,12 @@ class _MultiCloseContextManager(TupleContextManager): 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 diff --git a/tests/test_stdiomgr_base.py b/tests/test_stdiomgr_base.py index 3453776..a9d8105 100644 --- a/tests/test_stdiomgr_base.py +++ b/tests/test_stdiomgr_base.py @@ -28,6 +28,7 @@ import collections.abc import io +import os import sys import warnings @@ -543,6 +544,23 @@ def test_tee_type(): assert str(err.value) == "tee must be a TextIOBase." +def test_non_closing_type(): + """Test that incorrect type doesnt raise exceptions.""" + # Ensure the type used has no __exit__ or close() + assert not hasattr("", "__exit__") + assert not hasattr("", "close") + + with StdioManager(streams=("", "", "")): + pass + + +def test_dev_null(): + """Test that os.devnull is a valid input and output stream.""" + with open(os.devnull, "r") as devnull_in, open(os.devnull, "w") as devnull_out: + with StdioManager(streams=(devnull_in, devnull_out, devnull_out)): + print("hello") + + @pytest.mark.xfail(reason="Want to ensure 'real' warnings aren't suppressed") def test_bare_warning(skip_warnings): """Test that a "real" warning is exposed when raised."""