WIP: Base types

src/stdio_mgr/compat.py

@@ -0,0 +1,65 @@
+r"""``stdio_mgr.compat`` *code module*.
+
+``stdio_mgr.compat`` provides backports of Python standard library.
+
+**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 Python-2.0 License.
+
+**Members**
+
+"""
+import abc
+
+# AbstractContextManager was introduced in Python 3.6
+# and may be used with typing.ContextManager.
+# See https://github.com/jazzband/contextlib2/pull/21 for more complete backport
+try:
+    from contextlib import AbstractContextManager
+except ImportError:  # pragma: no cover
+    # Copied from _collections_abc
+    def _check_methods(cls, *methods):
+        mro = cls.__mro__
+        for method in methods:
+            for base in mro:
+                if method in base.__dict__:
+                    if base.__dict__[method] is None:
+                        return NotImplemented
+                    break
+            else:
+                return NotImplemented
+        return True
+
+    # Copied from contextlib
+    class AbstractContextManager(abc.ABC):
+        """An abstract base class for context managers."""
+
+        def __enter__(self):
+            """Return `self` upon entering the runtime context."""
+            return self
+
+        @abc.abstractmethod
+        def __exit__(self, exc_type, exc_value, traceback):
+            """Raise any exception triggered within the runtime context."""
+            return None
+
+        @classmethod
+        def __subclasshook__(cls, subclass):
+            """Check whether subclass is considered a subclass of this ABC."""
+            if cls is AbstractContextManager:
+                return _check_methods(subclass, "__enter__", "__exit__")
+            return NotImplemented

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,12 +38,11 @@
     TextIOWrapper,
 )
 
-# AbstractContextManager was introduced in Python 3.6
-# and may be used with typing.ContextManager.
-try:
-    from contextlib import AbstractContextManager
-except ImportError:  # pragma: no cover
-    AbstractContextManager = object
+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):
@@ -272,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.
@@ -346,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:  # pragma: no cover
+    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()