Re-integrate original StdioManager

Default to a inject implementation if possible.
(not a good idea)

Add stdio_mgr fixture to enumerate available implementations
on the current runtime.

conftest.py

@@ -29,12 +29,12 @@
 import os
 import sys
 import warnings
+from io import TextIOBase
 
 import _pytest.warnings
 import pytest
 
-from stdio_mgr import stdio_mgr
-from stdio_mgr.io import is_stdio_unbufferedio
+from stdio_mgr.stdio_mgr import _choose_inject_impl, BufferReplaceStdioManager
 
 
 @pytest.fixture(scope="session")
@@ -74,16 +74,23 @@ def enable_warnings_plugin(request):
         yield
 
 
-@pytest.fixture(scope="session")
-def unbufferedio():
-    """Provide concise access to PYTHONUNBUFFERED envvar."""
-    return is_stdio_unbufferedio()
+def pytest_generate_tests(metafunc):
+    """Parametrize fixture stdio_mgr."""
+    if "stdio_mgr" not in metafunc.fixturenames:
+        return
+
+    if isinstance(sys.stdin, TextIOBase):
+        impls = [BufferReplaceStdioManager, _choose_inject_impl()]
+    else:
+        impls = [BufferReplaceStdioManager]
+
+    metafunc.parametrize("stdio_mgr", impls)
 
 
-@pytest.fixture(autouse=True)
+@pytest.fixture(autouse=True, scope="session")
 def add_stdio_mgr(doctest_namespace):
     """Add stdio_mgr to doctest namespace."""
-    doctest_namespace["stdio_mgr"] = stdio_mgr
+    doctest_namespace["stdio_mgr"] = BufferReplaceStdioManager
     doctest_namespace["os"] = os
 
 

src/stdio_mgr/stdio_mgr.py

@@ -26,6 +26,7 @@
 
 """
 
+import sys
 from contextlib import suppress
 from io import (
     BufferedRandom,
@@ -37,15 +38,20 @@
     TextIOBase,
     TextIOWrapper,
 )
+from os import environ
 from tempfile import TemporaryFile
 
-from stdio_mgr.io import is_stdio_unbufferedio
 from stdio_mgr.types import (
+    _MultiCloseContextManager,
     _OutStreamsCloseContextManager,
     InjectSysIoContextManager,
+    ReplaceSysIoContextManager,
     StdioTuple,
 )
 
+_RUNTIME_SYS_STREAMS = StdioTuple([sys.__stdin__, sys.__stdout__, sys.__stderr__])
+_IMPORT_SYS_STREAMS = StdioTuple([sys.stdin, sys.stdout, sys.stderr])
+
 
 class _PersistedBytesIO(BytesIO):
     """Class to persist the value after close.
@@ -375,8 +381,8 @@ class StdioManagerBase(StdioTuple):
 
     def __new__(cls, in_str="", close=True):
         """Instantiate new context manager that emulates namedtuple."""
-        if close or _unbufferedio:
-            if _unbufferedio:
+        if close or not cls._RAW:
+            if not cls._RAW:
                 out_cls = SafeCloseRandomFileIO
             else:
                 out_cls = SafeCloseRandomTextIO
@@ -397,33 +403,83 @@ def __new__(cls, in_str="", close=True):
         return self
 
 
-_unbufferedio = is_stdio_unbufferedio()
+class BufferReplaceStdioManager(  # noqa: D101
+    ReplaceSysIoContextManager, StdioManagerBase, _MultiCloseContextManager
+):
+
+    _RAW = True
+
+    def close(self):
+        """Close files only if requested."""
+        if self._close:
+            return super().close()
+
+
+class FileInjectStdioManager(InjectSysIoContextManager, StdioManagerBase):  # noqa: D101
+
+    _RAW = False
+
+    def close(self):
+        """Dont close any streams."""
+
+    def __del__(self):
+        """Delete temporary files."""
+        try:
+            self.stdout._stream._f.close()
+        except (OSError, ValueError):
+            pass
+        try:
+            self.stderr._stream._f.close()
+        except (OSError, ValueError):
+            pass
+
+        del self.stdout._stream._f
+        del self.stderr._stream._f
+
+
+class BufferInjectStdioManager(  # noqa: D101
+    InjectSysIoContextManager, _OutStreamsCloseContextManager, StdioManagerBase
+):
+    def close(self):
+        """Close files only if requested."""
+        if self._close:
+            return super().close()
+
+
+def _current_streams():
+    return StdioTuple([sys.stdin, sys.stdout, sys.stderr])
+
+
+def _choose_inject_impl(currentio=None):
+    if not currentio:
+        currentio = _current_streams()
 
+    if environ.get("PYTHONUNBUFFERED"):
+        return FileInjectStdioManager
 
-if _unbufferedio:
+    try:
+        currentio.stdout.buffer.raw
+    except AttributeError:
+        pass
+    else:
+        return BufferInjectStdioManager
 
-    class StdioManager(InjectSysIoContextManager, StdioManagerBase):  # noqa: D101
+    return FileInjectStdioManager
 
-        _RAW = False
 
-        def close(self):
-            """Dont close any streams."""
+def _choose_impl(currentio=None):
+    if not currentio:
+        currentio = _current_streams()
 
-        def __del__(self):
-            """Delete temporary files."""
-            del self.stdout._stream._f
-            del self.stderr._stream._f
+    try:
+        currentio.stdin.buffer
+    except AttributeError:
+        return BufferReplaceStdioManager
 
+    return _choose_inject_impl(currentio)
 
-else:
 
-    class StdioManager(  # noqa: D101
-        InjectSysIoContextManager, _OutStreamsCloseContextManager, StdioManagerBase
-    ):
-        def close(self):
-            """Close files only if requested."""
-            if self._close:
-                return super().close()
+StdioManager = _choose_impl(_IMPORT_SYS_STREAMS)
 
 
 stdio_mgr = StdioManager