diff --git a/conftest.py b/conftest.py index 4498bcb..c3e6d00 100644 --- a/conftest.py +++ b/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) diff --git a/src/stdio_mgr/io.py b/src/stdio_mgr/io.py new file mode 100644 index 0000000..bcc1f44 --- /dev/null +++ b/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 diff --git a/src/stdio_mgr/stdio_mgr.py b/src/stdio_mgr/stdio_mgr.py index 2235d04..c28aaa7 100644 --- a/src/stdio_mgr/stdio_mgr.py +++ b/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 diff --git a/src/stdio_mgr/types.py b/src/stdio_mgr/types.py index bbb9dd5..d8bd8ec 100644 --- a/src/stdio_mgr/types.py +++ b/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) diff --git a/tests/test_stdiomgr_base.py b/tests/test_stdiomgr_base.py index 0f1dfd8..e441063 100644 --- a/tests/test_stdiomgr_base.py +++ b/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()