Merge two injection implementations

Local usage uses unbuffered.
CI usage uses buffered.
Local usage of PYTHONUNBUFFERED=1 uses buffered

Detection is not working

conftest.py

@@ -34,26 +34,7 @@
 import pytest
 
 from stdio_mgr import stdio_mgr
-
-
-def is_stdout_buffered():
-    """Check if stdout is buffered.
-
-    Copied from https://stackoverflow.com/a/49736559
-    Licensed CC-BY-SA 4.0
-    Author https://stackoverflow.com/users/528711/sparrowt
-    """
-    # Print a single space + carriage return but no new-line
-    # (should have no visible effect)
-    print(" \r")
-    # If the file position is a positive integer then stdout is buffered
-    try:
-        pos = sys.stdout.tell()
-        if pos > 0:
-            return True
-    except IOError:  # In some terminals tell() throws IOError if stdout is unbuffered
-        pass
-    return False
+from stdio_mgr.io import is_stdio_unbufferedio
 
 
 @pytest.fixture(scope="session")
@@ -96,7 +77,7 @@ def enable_warnings_plugin(request):
 @pytest.fixture(scope="session")
 def unbufferedio():
     """Provide concise access to PYTHONUNBUFFERED envvar."""
-    return os.environ.get("PYTHONUNBUFFERED") or not is_stdout_buffered()
+    return is_stdio_unbufferedio()
 
 
 @pytest.fixture(autouse=True)

src/stdio_mgr/io.py

@@ -0,0 +1,49 @@
+"""IO related functions and classes."""
+import os
+import sys
+
+# Any use of is_stdout_buffered is temporary, because -u is not reflected
+# anywhere in the python sys module, such as sys.flags.
+# It is undesirable as written, as it has side-effects, especially bad
+# side-effects as using a stream usually causes parts of the stream
+# state to get baked-in, preventing reconfiguration on Python 3.7.
+# Also temporary because of incompatible licensing.
+
+
+def is_stdout_buffered():
+    """Check if stdout is buffered.
+
+    Copied from https://stackoverflow.com/a/49736559
+    Licensed CC-BY-SA 4.0
+    Author https://stackoverflow.com/users/528711/sparrowt
+    with a fix of the print for Python 3 compatibility
+    """
+    # Print a single space + carriage return but no new-line
+    # (should have no visible effect)
+    print(" \r")
+    # If the file position is a positive integer then stdout is buffered
+    try:
+        pos = sys.stdout.tell()
+        if pos > 0:
+            return True
+    except IOError:  # In some terminals tell() throws IOError if stdout is unbuffered
+        pass
+    return False
+
+
+def is_stdio_unbufferedio():
+    """Detect if PYTHONUNBUFFERED was set or CI was set as proxy for -u."""
+    # It should be possible to detect unbuffered by the *initial* state of
+    # sys.stdout and its buffers.
+    # Also need to take into account PYTHONLEGACYWINDOWSSTDIO
+    # And all this is quite unhelpful because the state of sys.stdio
+    # may be very different by the time that a StdioManager is instantiated
+    if os.environ.get("PYTHONUNBUFFERED"):
+        return True
+
+    try:
+        sys.stdout.buffer.raw
+    except AttributeError:
+        return True
+
+    return False

src/stdio_mgr/stdio_mgr.py

@@ -39,7 +39,12 @@
 )
 from tempfile import TemporaryFile
 
-from stdio_mgr.types import InjectSysIoContextManager
+from stdio_mgr.io import is_stdio_unbufferedio
+from stdio_mgr.types import (
+    _OutStreamsCloseContextManager,
+    InjectSysIoContextManager,
+    StdioTuple,
+)
 
 
 class _PersistedBytesIO(BytesIO):
@@ -330,7 +335,7 @@ class SafeCloseTeeStdin(_SafeCloseIOBase, TeeStdin):
     """
 
 
-class StdioManager(InjectSysIoContextManager):
+class StdioManagerBase(StdioTuple):
     r"""Substitute temporary text buffers for `stdio` in a managed context.
 
     Context manager.
@@ -370,31 +375,55 @@ class StdioManager(InjectSysIoContextManager):
 
     def __new__(cls, in_str="", close=True):
         """Instantiate new context manager that emulates namedtuple."""
-        if close:
-            out_cls = SafeCloseRandomFileIO
+        if close or _unbufferedio:
+            if _unbufferedio:
+                out_cls = SafeCloseRandomFileIO
+            else:
+                out_cls = SafeCloseRandomTextIO
             in_cls = SafeCloseTeeStdin
         else:
-            out_cls = RandomFileIO
+            # no unbufferedio equivalent exists yet
+            out_cls = RandomTextIO
             in_cls = TeeStdin
 
         stdout = out_cls()
         stderr = out_cls()
         stdin = in_cls(stdout, in_str)
 
-        self = super(StdioManager, cls).__new__(cls, [stdin, stdout, stderr])
+        self = super(StdioManagerBase, cls).__new__(cls, [stdin, stdout, stderr])
 
         self._close = close
 
         return self
 
-    def close(self):
-        """Dont close out streams."""
-        self.stdin.close()
 
-    def __del__(self):
-        """Delete temporary files."""
-        del self.stdout._stream._f
-        del self.stderr._stream._f
+_unbufferedio = is_stdio_unbufferedio()
+
+
+if _unbufferedio:
+
+    class StdioManager(InjectSysIoContextManager, StdioManagerBase):  # noqa: D101
+
+        _RAW = False
+
+        def close(self):
+            """Dont close any streams."""
+
+        def __del__(self):
+            """Delete temporary files."""
+            del self.stdout._stream._f
+            del self.stderr._stream._f
+
+
+else:
+
+    class StdioManager(  # noqa: D101
+        InjectSysIoContextManager, _OutStreamsCloseContextManager, StdioManagerBase
+    ):
+        def close(self):
+            """Close files only if requested."""
+            if self._close:
+                return super().close()
 
 
 stdio_mgr = StdioManager

src/stdio_mgr/types.py

@@ -198,26 +198,40 @@ def __enter__(self):
 class InjectSysIoContextManager(StdioTuple):
     """Replace sys stdio with members of the tuple."""
 
+    _RAW = True
+
     def __enter__(self):
         """Enter context, replacing sys stdio objects with capturing streams."""
         self._prior_stdin = sys.stdin
-        self._prior_filenos = (sys.stdout.fileno(), sys.stderr.fileno())
+        if self._RAW:
+            self._prior_out = sys.stdout.buffer.raw, sys.stderr.buffer.raw
+            new_stdout = self.stdout.buffer.raw
+            new_stderr = self.stderr.buffer.raw
+        else:
+            self._prior_out = (sys.stdout.fileno(), sys.stderr.fileno())
+            new_stdout = self.stdout.fileno()
+            new_stderr = self.stderr.fileno()
 
         super().__enter__()
 
         sys.stdin = self.stdin
-        sys.stdout.buffer.__init__(self.stdout.fileno())
-        sys.stderr.buffer.__init__(self.stderr.fileno())
+        sys.stdout.buffer.__init__(new_stdout)
+        sys.stderr.buffer.__init__(new_stderr)
 
         return self
 
     def __exit__(self, exc_type, exc_value, traceback):
         """Exit context, restoring state of sys module."""
-        self.stdout._save_value()
-        self.stderr._save_value()
+        if not self._RAW:
+            self.stdout._save_value()
+            self.stderr._save_value()
 
         sys.stdin = self._prior_stdin
-        sys.stdout.buffer.__init__(self._prior_filenos[0], mode="wb", closefd=False)
-        sys.stderr.buffer.__init__(self._prior_filenos[1], mode="wb", closefd=False)
+        if self._RAW:
+            sys.stdout.buffer.__init__(self._prior_out[0])
+            sys.stderr.buffer.__init__(self._prior_out[1])
+        else:
+            sys.stdout.buffer.__init__(self._prior_out[0], mode="wb", closefd=False)
+            sys.stderr.buffer.__init__(self._prior_out[1], mode="wb", closefd=False)
 
         return super().__exit__(exc_type, exc_value, traceback)

tests/test_stdiomgr_base.py

@@ -47,7 +47,7 @@
 _BUFFER_DETACHED_MSG = "underlying buffer has been detached"
 
 
-def test_context_manager_instance():
+def test_context_manager_instantiation():
     """Confirm StdioManager instance is a tuple and registered context manager."""
     cm = StdioManager()