DEV: Add types and triple modules

IOTriple combines MultiItemIterable and TupleContextManager,
and provides properties `stdin`, `stdout` and `stderr` for
literate referencing the three members of the tuple.

MultiItemIterable provides _map(), _all() and _any() for
iterables, with suppress versions able to ignore exceptions.

Base TupleContextManager creates a consistent MRO of tuple and
AbstractContextManager.

ClosingStdioTuple is a sample context manager which performs close()
on each item in the tuple.

TextIOTriple requires items to be a TextIOBase.

AutoIOTriple can be used to create either a TextIOTriple or a
FakeIOTriple, depending on whether the items are all a TextIOBase.
This will allow attaching different implementations to each.

AutoIOTuple is used to capture `sys.__std*__` and `sys.std*` as
they existed at module import.

src/stdio_mgr/stdio_mgr.py

@@ -27,7 +27,7 @@
 """
 
 import sys
-from contextlib import ExitStack, suppress
+from contextlib import suppress
 from io import (
     BufferedRandom,
     BufferedReader,
@@ -38,7 +38,11 @@
     TextIOWrapper,
 )
 
-from stdio_mgr.compat import AbstractContextManager
+from stdio_mgr.triple import AutoIOTriple, IOTriple
+from stdio_mgr.types import _MultiCloseContextManager
+
+_RUNTIME_SYS_STREAMS = AutoIOTriple([sys.__stdin__, sys.__stdout__, sys.__stderr__])
+_IMPORT_SYS_STREAMS = AutoIOTriple([sys.stdin, sys.stdout, sys.stderr])
 
 
 class _PersistedBytesIO(BytesIO):
@@ -267,24 +271,7 @@ class SafeCloseTeeStdin(_SafeCloseIOBase, TeeStdin):
     """
 
 
-class _MultiCloseContextManager(tuple, AbstractContextManager):
-    """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))
-
-            self._close_files = stack.pop_all().close
-
-        return self
-
-    def __exit__(self, exc_type, exc_value, traceback):
-        """Exit context, closing all members."""
-        self._close_files()
-
-
-class StdioManager(_MultiCloseContextManager):
+class StdioManager(_MultiCloseContextManager, IOTriple):
     r"""Substitute temporary text buffers for `stdio` in a managed context.
 
     Context manager.
@@ -341,21 +328,6 @@ def __new__(cls, in_str="", close=True):
 
         return self
 
-    @property
-    def stdin(self):
-        """Return capturing stdin stream."""
-        return self[0]
-
-    @property
-    def stdout(self):
-        """Return capturing stdout stream."""
-        return self[1]
-
-    @property
-    def stderr(self):
-        """Return capturing stderr stream."""
-        return self[2]
-
     def __enter__(self):
         """Enter context, replacing sys stdio objects with capturing streams."""
         self._prior_streams = (sys.stdin, sys.stdout, sys.stderr)

src/stdio_mgr/triple.py

@@ -0,0 +1,100 @@
+r"""``stdio_mgr.types`` *code module*.
+
+``stdio_mgr.types`` provides context managers for convenient
+interaction with ``stdin``/``stdout``/``stderr`` as a tuple.
+
+**Author**
+    John Vandenberg ([email protected])
+
+**File Created**
+    6 Sep 2019
+
+**Copyright**
+    \(c) Brian Skinn 2018-2019
+
+**Source Repository**
+    http://www.github.com/bskinn/stdio-mgr
+
+**Documentation**
+    See README.rst at the GitHub repository
+
+**License**
+    The MIT License; see |license_txt|_ for full license terms
+
+**Members**
+
+"""
+from io import TextIOBase
+
+from stdio_mgr.types import MultiItemIterable, TupleContextManager
+
+
+class IOTriple(TupleContextManager, MultiItemIterable):
+    """Base type for a type of stdin, stdout and stderr stream-like objects.
+
+    While it is a context manager, no action is taken on entering or exiting
+    the context.
+
+    Used as a context manager, it will close all of the streams on exit.
+    however it does not open the streams on entering the context manager.
+
+    No exception handling is performed while closing the streams, so any
+    exception occurring the close of any stream renders them all in an
+    unpredictable state.
+    """
+
+    def __new__(cls, iterable):
+        """Instantiate new tuple from iterable containing three streams."""
+        items = list(iterable)
+        assert len(items) == 3, "iterable must be three items"  # noqa: S101
+
+        return super(IOTriple, cls).__new__(cls, items)
+
+    @property
+    def stdin(self):
+        """Return stdin stream."""
+        return self[0]
+
+    @property
+    def stdout(self):
+        """Return stdout stream."""
+        return self[1]
+
+    @property
+    def stderr(self):
+        """Return stderr stream."""
+        return self[2]
+
+
+class TextIOTriple(IOTriple):
+    """Tuple context manager of stdin, stdout and stderr TextIOBase objects."""
+
+    _ITEM_BASE = TextIOBase
+
+    # pytest and colorama inject objects into sys.std* that are not real TextIOBase
+    # and fail the assertion of this class
+
+    def __new__(cls, iterable):
+        """Instantiate new tuple from iterable containing three TextIOBase streams."""
+        self = super(TextIOTriple, cls).__new__(cls, iterable)
+        if not self.all_(lambda item: isinstance(item, cls._ITEM_BASE)):
+            raise ValueError(
+                "iterable must contain only {}".format(cls._ITEM_BASE.__name__)
+            )
+        return self
+
+
+class FakeIOTriple(IOTriple):
+    """Tuple context manager of stdin, stdout and stderr-like objects."""
+
+
+class AutoIOTriple(IOTriple):
+    """Tuple context manager which will create FakeIOTuple or TextIOTuple."""
+
+    def __new__(cls, iterable):
+        """Instantiate new TextIOTuple or FakeIOTuple from iterable."""
+        items = list(iterable)
+        if any(not isinstance(item, TextIOTriple._ITEM_BASE) for item in items):
+            return FakeIOTriple(items)
+        else:
+            return TextIOTriple(items)

src/stdio_mgr/types.py

@@ -0,0 +1,122 @@
+r"""``stdio_mgr.types`` *code module*.
+
+``stdio_mgr.types`` provides misc. types and classes.
+
+**Author**
+    John Vandenberg ([email protected])
+
+**File Created**
+    6 Sep 2019
+
+**Copyright**
+    \(c) Brian Skinn 2018-2019
+
+**Source Repository**
+    http://www.github.com/bskinn/stdio-mgr
+
+**Documentation**
+    See README.rst at the GitHub repository
+
+**License**
+    The MIT License; see |license_txt|_ for full license terms
+
+**Members**
+
+"""
+import collections.abc
+from contextlib import ExitStack, suppress
+
+try:
+    from contextlib import AbstractContextManager
+except ImportError:
+    from stdio_mgr.compat import AbstractContextManager
+
+
+class MultiItemIterable(collections.abc.Iterable):
+    """Iterable with methods that operate on all items."""
+
+    @staticmethod
+    def _invoke_name(item, name):
+        item = getattr(item, name)
+        if callable(item):
+            return item()
+        return item
+
+    def _map_name(self, name):
+        """Perform attribute name on all items."""
+        for item in self:
+            item = self._invoke_name(item, name)
+            yield item
+
+    def map_(self, op):
+        """Return generator for performing op on all items."""
+        if isinstance(op, str):
+            return self._map_name(op)
+
+        return map(op, self)
+
+    def suppress_map(self, ex, op):
+        """Return generator for performing op on all item, suppressing ex."""
+        for item in self:
+            with suppress(ex):
+                yield self._invoke_name(item, op) if isinstance(op, str) else op(item)
+
+    def all_(self, op):
+        """Perform op on all items, returning True when all were successful."""
+        return all(self.map_(op))
+
+    def suppress_all(self, ex, op):
+        """Perform op on all items, suppressing ex."""
+        return all(self.suppress_map(ex, op))
+
+    def any_(self, op):
+        """Perform op on all items, returning True when all were successful."""
+        return any(self.map_(op))
+
+
+class TupleContextManager(tuple, AbstractContextManager):
+    """Base for context managers that are also a tuple."""
+
+    # This is needed to establish a workable MRO.
+
+
+class ClosingStdioTuple(TupleContextManager, MultiItemIterable):
+    """Context manager of streams objects to close them on exit.
+
+    Used as a context manager, it will close all of the streams on exit.
+    however it does not open the streams on entering the context manager.
+
+    No exception handling is performed while closing the streams, so any
+    exception occurring the close of any stream renders them all in an
+    unpredictable state.
+    """
+
+    def close(self):
+        """Close all streams."""
+        return list(self.map_("close"))
+
+    def safe_close(self):
+        """Close all streams, ignoring any ValueError."""
+        return list(self.suppress_map(ValueError, "close"))
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        """Exit context, closing all members."""
+        self.close()
+        return super().__exit__(exc_type, exc_value, traceback)
+
+
+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))
+
+            self._close_files = stack.pop_all().close
+
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        """Exit context, closing all members."""
+        self._close_files()

tests/test_stdiomgr_base.py

@@ -35,8 +35,14 @@
 import pytest
 
 from stdio_mgr import stdio_mgr, StdioManager
-from stdio_mgr.compat import AbstractContextManager
-from stdio_mgr.stdio_mgr import _MultiCloseContextManager, _Tee
+from stdio_mgr.stdio_mgr import _Tee
+from stdio_mgr.triple import IOTriple
+from stdio_mgr.types import (
+    _MultiCloseContextManager,
+    AbstractContextManager,
+    MultiItemIterable,
+    TupleContextManager,
+)
 
 _WARNING_ARGS_ERROR = "Please use pytest -p no:warnings or pytest --W error::Warning"
 _SKIP_WARNING_TESTS = "Skip tests using warnings when warnings are errors"
@@ -76,9 +82,13 @@ def test_context_manager_mro():
     assert mro == (
         StdioManager,
         _MultiCloseContextManager,
+        IOTriple,
+        TupleContextManager,
         tuple,
         AbstractContextManager,
         abc.ABC,
+        MultiItemIterable,
+        collections.abc.Iterable,
         object,
     )