diff options
Diffstat (limited to 'Lib')
| -rw-r--r-- | Lib/_pyio.py | 51 | ||||
| -rw-r--r-- | Lib/test/test_io.py | 31 | ||||
| -rw-r--r-- | Lib/test/test_memoryio.py | 10 |
3 files changed, 86 insertions, 6 deletions
diff --git a/Lib/_pyio.py b/Lib/_pyio.py index e580366..e3e7c3d 100644 --- a/Lib/_pyio.py +++ b/Lib/_pyio.py @@ -642,6 +642,15 @@ class BufferedIOBase(IOBase): """ self._unsupported("write") + def detach(self) -> None: + """ + Separate the underlying raw stream from the buffer and return it. + + After the raw stream has been detached, the buffer is in an unusable + state. + """ + self._unsupported("detach") + io.BufferedIOBase.register(BufferedIOBase) @@ -689,13 +698,21 @@ class _BufferedIOMixin(BufferedIOBase): self.raw.flush() def close(self): - if not self.closed: + if not self.closed and self.raw is not None: try: self.flush() except IOError: pass # If flush() fails, just give up self.raw.close() + def detach(self): + if self.raw is None: + raise ValueError("raw stream already detached") + self.flush() + raw = self.raw + self.raw = None + return raw + ### Inquiries ### def seekable(self): @@ -1236,6 +1253,15 @@ class TextIOBase(IOBase): """ self._unsupported("readline") + def detach(self) -> None: + """ + Separate the underlying buffer from the TextIOBase and return it. + + After the underlying buffer has been detached, the TextIO is in an + unusable state. + """ + self._unsupported("detach") + @property def encoding(self): """Subclasses should override.""" @@ -1448,11 +1474,12 @@ class TextIOWrapper(TextIOBase): self._telling = self._seekable def close(self): - try: - self.flush() - except IOError: - pass # If flush() fails, just give up - self.buffer.close() + if self.buffer is not None: + try: + self.flush() + except IOError: + pass # If flush() fails, just give up + self.buffer.close() @property def closed(self): @@ -1647,6 +1674,14 @@ class TextIOWrapper(TextIOBase): self.seek(pos) return self.buffer.truncate() + def detach(self): + if self.buffer is None: + raise ValueError("buffer is already detached") + self.flush() + buffer = self.buffer + self.buffer = None + return buffer + def seek(self, cookie, whence=0): if self.closed: raise ValueError("tell on closed file") @@ -1865,3 +1900,7 @@ class StringIO(TextIOWrapper): @property def encoding(self): return None + + def detach(self): + # This doesn't make sense on StringIO. + self._unsupported("detach") diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py index a8c878e..1a525dc 100644 --- a/Lib/test/test_io.py +++ b/Lib/test/test_io.py @@ -526,6 +526,12 @@ class PyIOTest(IOTest): class CommonBufferedTests: # Tests common to BufferedReader, BufferedWriter and BufferedRandom + def test_detach(self): + raw = self.MockRawIO() + buf = self.tp(raw) + self.assertIs(buf.detach(), raw) + self.assertRaises(ValueError, buf.detach) + def test_fileno(self): rawio = self.MockRawIO() bufio = self.tp(rawio) @@ -811,6 +817,14 @@ class BufferedWriterTest(unittest.TestCase, CommonBufferedTests): bufio.flush() self.assertEquals(b"".join(rawio._write_stack), b"abcghi") + def test_detach_flush(self): + raw = self.MockRawIO() + buf = self.tp(raw) + buf.write(b"howdy!") + self.assertFalse(raw._write_stack) + buf.detach() + self.assertEqual(raw._write_stack, [b"howdy!"]) + def test_write(self): # Write to the buffered IO but don't overflow the buffer. writer = self.MockRawIO() @@ -1052,6 +1066,10 @@ class BufferedRWPairTest(unittest.TestCase): pair = self.tp(self.MockRawIO(), self.MockRawIO()) self.assertFalse(pair.closed) + def test_detach(self): + pair = self.tp(self.MockRawIO(), self.MockRawIO()) + self.assertRaises(self.UnsupportedOperation, pair.detach) + def test_constructor_max_buffer_size_deprecation(self): with support.check_warnings() as w: warnings.simplefilter("always", DeprecationWarning) @@ -1480,6 +1498,19 @@ class TextIOWrapperTest(unittest.TestCase): self.assertRaises(TypeError, t.__init__, b, newline=42) self.assertRaises(ValueError, t.__init__, b, newline='xyzzy') + def test_detach(self): + r = self.BytesIO() + b = self.BufferedWriter(r) + t = self.TextIOWrapper(b) + self.assertIs(t.detach(), b) + + t = self.TextIOWrapper(b, encoding="ascii") + t.write("howdy") + self.assertFalse(r.getvalue()) + t.detach() + self.assertEqual(r.getvalue(), b"howdy") + self.assertRaises(ValueError, t.detach) + def test_repr(self): raw = self.BytesIO("hello".encode("utf-8")) b = self.BufferedReader(raw) diff --git a/Lib/test/test_memoryio.py b/Lib/test/test_memoryio.py index ad04613..d94d9dd 100644 --- a/Lib/test/test_memoryio.py +++ b/Lib/test/test_memoryio.py @@ -57,6 +57,10 @@ class MemorySeekTestMixin: class MemoryTestMixin: + def test_detach(self): + buf = self.ioclass() + self.assertRaises(self.UnsupportedOperation, buf.detach) + def write_ops(self, f, t): self.assertEqual(f.write(t("blah.")), 5) self.assertEqual(f.seek(0), 0) @@ -336,6 +340,9 @@ class MemoryTestMixin: class PyBytesIOTest(MemoryTestMixin, MemorySeekTestMixin, unittest.TestCase): + + UnsupportedOperation = pyio.UnsupportedOperation + @staticmethod def buftype(s): return s.encode("ascii") @@ -413,6 +420,7 @@ class PyBytesIOTest(MemoryTestMixin, MemorySeekTestMixin, unittest.TestCase): class PyStringIOTest(MemoryTestMixin, MemorySeekTestMixin, unittest.TestCase): buftype = str ioclass = pyio.StringIO + UnsupportedOperation = pyio.UnsupportedOperation EOF = "" # TextIO-specific behaviour. @@ -518,9 +526,11 @@ class PyStringIOTest(MemoryTestMixin, MemorySeekTestMixin, unittest.TestCase): class CBytesIOTest(PyBytesIOTest): ioclass = io.BytesIO + UnsupportedOperation = io.UnsupportedOperation class CStringIOTest(PyStringIOTest): ioclass = io.StringIO + UnsupportedOperation = io.UnsupportedOperation # XXX: For the Python version of io.StringIO, this is highly # dependent on the encoding used for the underlying buffer. |