Re-enable most tests in unbufferedio mode

src/stdio_mgr/stdio_mgr.py

@@ -39,7 +39,7 @@
 )
 from tempfile import TemporaryFile
 
-from .types import InjectSysIoContextManager
+from stdio_mgr.types import InjectSysIoContextManager
 
 
 class _PersistedBytesIO(BytesIO):
@@ -90,6 +90,7 @@ def getvalue(self):
 
     def close(self):
         """Send buffer to callback and close."""
+        # This isnt being called in unbufferedio mode
         self._callback(self.getvalue())
         # Do not call super close()
 
@@ -115,6 +116,7 @@ def __init__(self):
         if not hasattr(self, "_stream"):
             self._stream = _PersistedBytesIO(self._set_closed_buf)
         self._buf = BufferedRandom(self._stream)
+        self._closed_buf = None
         super().__init__(self._buf, encoding="utf-8")
 
     def write(self, *args, **kwargs):
@@ -127,11 +129,14 @@ def _set_closed_buf(self, value):
 
     def getvalue(self):
         """Obtain buffer of text sent to the stream."""
-        if self._stream.closed:
+        if self._closed_buf is not None:
             return self._closed_buf.decode(self.encoding)
         else:
             return self._stream.getvalue().decode(self.encoding)
 
+    def _save_value(self):
+        self._closed_buf = self._stream.getvalue()
+
 
 class RandomFileIO(RandomTextIO):
     """Class to capture writes to a file even when detached."""
@@ -141,6 +146,11 @@ def __init__(self):
         self._stream = _PersistedFileIO(self._set_closed_buf)
         super().__init__()
 
+    def close(self):
+        """Detach buffer on close, to more closely emulate close."""
+        self._save_value()
+        self.detach()
+
 
 class _Tee(TextIOWrapper):
     """Class to tee contents to a side buffer on read.
@@ -378,7 +388,8 @@ def __new__(cls, in_str="", close=True):
         return self
 
     def close(self):
-        """Dont close any streams."""
+        """Dont close out streams."""
+        self.stdin.close()
 
     def __del__(self):
         """Delete temporary files."""

src/stdio_mgr/types.py

@@ -213,6 +213,9 @@ def __enter__(self):
 
     def __exit__(self, exc_type, exc_value, traceback):
         """Exit context, restoring state of sys module."""
+        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)

tests/test_stdiomgr_base.py

@@ -44,6 +44,7 @@
 from stdio_mgr.types import _MultiCloseContextManager, StdioTuple, TextIOTuple
 
 _WARNING_ARGS_ERROR = "Please use pytest -p no:warnings or pytest --W error::Warning"
+_BUFFER_DETACHED_MSG = "underlying buffer has been detached"
 
 
 def test_context_manager_instance():
@@ -368,7 +369,7 @@ def test_manual_close_detached_fails(convert_newlines, unbufferedio):
         with pytest.raises(ValueError) as err:
             i.close()
 
-        assert str(err.value) == "underlying buffer has been detached"
+        assert str(err.value) == _BUFFER_DETACHED_MSG
 
         with pytest.raises(ValueError):
             i.closed
@@ -384,7 +385,7 @@ def test_manual_close_detached_fails(convert_newlines, unbufferedio):
     with pytest.raises(ValueError) as err:
         i.close()
 
-    assert str(err.value) == "underlying buffer has been detached"
+    assert str(err.value) == _BUFFER_DETACHED_MSG
 
     with pytest.raises(ValueError):
         i.closed
@@ -400,9 +401,6 @@ def test_manual_close_detached_fails(convert_newlines, unbufferedio):
 
 def test_stdin_closed(convert_newlines, unbufferedio):
     """Confirm stdin's buffer can be closed within the context."""
-    if unbufferedio:
-        pytest.skip("Skip detach/close not handled yet")
-
     with stdio_mgr() as (i, o, e):
         print("test str")
 
@@ -428,9 +426,6 @@ def test_stdin_detached(convert_newlines, unbufferedio):
 
     Like the real sys.stdin, use after detach should fail with ValueError.
     """
-    if unbufferedio:
-        pytest.skip("Skip detach/close not handled yet")
-
     with stdio_mgr() as (i, o, e):
         print("test str")
 
@@ -439,17 +434,17 @@ def test_stdin_detached(convert_newlines, unbufferedio):
         with pytest.raises(ValueError) as err:
             i.read()
 
-        assert str(err.value) == "underlying buffer has been detached"
+        assert str(err.value) == _BUFFER_DETACHED_MSG
 
         with pytest.raises(ValueError) as err:
             i.getvalue()
 
-        assert str(err.value) == "underlying buffer has been detached"
+        assert str(err.value) == _BUFFER_DETACHED_MSG
 
         with pytest.raises(ValueError) as err:
             i.append("anything")
 
-        assert str(err.value) == "underlying buffer has been detached"
+        assert str(err.value) == _BUFFER_DETACHED_MSG
 
         assert convert_newlines("test str\n") == o.getvalue()
 
@@ -460,7 +455,7 @@ def test_stdin_detached(convert_newlines, unbufferedio):
         with pytest.raises(ValueError) as err:
             i.closed
 
-        assert str(err.value) == "underlying buffer has been detached"
+        assert str(err.value) == _BUFFER_DETACHED_MSG
 
     assert convert_newlines("test str\nsecond test str\n") == o.getvalue()
 
@@ -469,10 +464,11 @@ def test_stdin_detached(convert_newlines, unbufferedio):
     with pytest.raises(ValueError) as err:
         i.closed
 
-    assert str(err.value) == "underlying buffer has been detached"
+    assert str(err.value) == _BUFFER_DETACHED_MSG
 
-    assert o.closed
-    assert e.closed
+    if not unbufferedio:
+        assert o.closed
+        assert e.closed
 
 
 def test_stdout_detached(convert_newlines, unbufferedio):
@@ -481,9 +477,6 @@ def test_stdout_detached(convert_newlines, unbufferedio):
     Like the real sys.stdout, writes after detach should fail, however
     writes to the detached stream should be captured.
     """
-    if unbufferedio:
-        pytest.skip("Skip detach/close not handled yet")
-
     with stdio_mgr() as (i, o, e):
         print("test str")
 
@@ -498,7 +491,7 @@ def test_stdout_detached(convert_newlines, unbufferedio):
         with pytest.raises(ValueError) as err:
             o.write("second test str\n")
 
-        assert str(err.value) == "underlying buffer has been detached"
+        assert str(err.value) == _BUFFER_DETACHED_MSG
 
         assert convert_newlines("test str\n") == o.getvalue()
 
@@ -510,7 +503,7 @@ def test_stdout_detached(convert_newlines, unbufferedio):
         with pytest.raises(ValueError) as err:
             o.closed
 
-        assert str(err.value) == "underlying buffer has been detached"
+        assert str(err.value) == _BUFFER_DETACHED_MSG
 
     assert convert_newlines("test str\nsecond test str\n") == o.getvalue()
 
@@ -519,17 +512,15 @@ def test_stdout_detached(convert_newlines, unbufferedio):
     with pytest.raises(ValueError) as err:
         o.closed
 
-    assert str(err.value) == "underlying buffer has been detached"
+    assert str(err.value) == _BUFFER_DETACHED_MSG
 
-    assert i.closed
-    assert e.closed
+    if not unbufferedio:
+        assert i.closed
+        assert e.closed
 
 
 def test_stdout_access_buffer_after_close(convert_newlines, unbufferedio):
     """Confirm stdout's buffer is captured after close."""
-    if unbufferedio:
-        pytest.skip("Skip detach/close not handled yet")
-
     with stdio_mgr() as (i, o, e):
         print("test str")
 
@@ -541,14 +532,21 @@ def test_stdout_access_buffer_after_close(convert_newlines, unbufferedio):
         with pytest.raises(ValueError) as err:
             o.read()
 
-        assert str(err.value) == "I/O operation on closed file."
+        if unbufferedio:
+            assert str(err.value) == _BUFFER_DETACHED_MSG
+        else:
+            assert str(err.value) == "I/O operation on closed file."
 
         assert convert_newlines("test str\nsecond test str\n") == o.getvalue()
 
-        with pytest.raises(ValueError) as err:
-            print("anything")
+        # because the real sys handle isnt closed, print still works
+        # to assist, we can override `print`, leaving sys.stdout.write
+        # still unpatched.
+        if not unbufferedio:
+            with pytest.raises(ValueError) as err:
+                print("anything")
 
-        assert str(err.value) == "write to closed file"
+            assert str(err.value) == "write to closed file"
 
         assert convert_newlines("test str\nsecond test str\n") == o.getvalue()