Issue #12591: Allow io.TextIOWrapper to work with raw IO objects (without
a read1() method), and add an undocumented *write_through* parameter to
mandate unbuffered writes.
diff --git a/Lib/_pyio.py b/Lib/_pyio.py
index b79d5fc..78c6d95 100644
--- a/Lib/_pyio.py
+++ b/Lib/_pyio.py
@@ -1472,7 +1472,7 @@
_CHUNK_SIZE = 2048
def __init__(self, buffer, encoding=None, errors=None, newline=None,
- line_buffering=False):
+ line_buffering=False, write_through=False):
if newline is not None and not isinstance(newline, str):
raise TypeError("illegal newline type: %r" % (type(newline),))
if newline not in (None, "", "\n", "\r", "\r\n"):
@@ -1515,6 +1515,7 @@
self._decoded_chars_used = 0 # offset into _decoded_chars for read()
self._snapshot = None # info for reconstructing decoder state
self._seekable = self._telling = self.buffer.seekable()
+ self._has_read1 = hasattr(self.buffer, 'read1')
if self._seekable and self.writable():
position = self.buffer.tell()
@@ -1680,7 +1681,10 @@
# len(dec_buffer) bytes ago with decoder state (b'', dec_flags).
# Read a chunk, decode it, and put the result in self._decoded_chars.
- input_chunk = self.buffer.read1(self._CHUNK_SIZE)
+ if self._has_read1:
+ input_chunk = self.buffer.read1(self._CHUNK_SIZE)
+ else:
+ input_chunk = self.buffer.read(self._CHUNK_SIZE)
eof = not input_chunk
self._set_decoded_chars(self._decoder.decode(input_chunk, eof))
diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py
index 39fda2b..0ca621c 100644
--- a/Lib/test/test_io.py
+++ b/Lib/test/test_io.py
@@ -2307,6 +2307,27 @@
with self.assertRaises(AttributeError):
txt.buffer = buf
+ def test_rawio(self):
+ # Issue #12591: TextIOWrapper must work with raw I/O objects, so
+ # that subprocess.Popen() can have the required unbuffered
+ # semantics with universal_newlines=True.
+ raw = self.MockRawIO([b'abc', b'def', b'ghi\njkl\nopq\n'])
+ txt = self.TextIOWrapper(raw, encoding='ascii', newline='\n')
+ # Reads
+ self.assertEqual(txt.read(4), 'abcd')
+ self.assertEqual(txt.readline(), 'efghi\n')
+ self.assertEqual(list(txt), ['jkl\n', 'opq\n'])
+
+ def test_rawio_write_through(self):
+ # Issue #12591: with write_through=True, writes don't need a flush
+ raw = self.MockRawIO([b'abc', b'def', b'ghi\njkl\nopq\n'])
+ txt = self.TextIOWrapper(raw, encoding='ascii', newline='\n',
+ write_through=True)
+ txt.write('1')
+ txt.write('23\n4')
+ txt.write('5')
+ self.assertEqual(b''.join(raw._write_stack), b'123\n45')
+
class CTextIOWrapperTest(TextIOWrapperTest):
def test_initialization(self):