summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_io.py
diff options
context:
space:
mode:
authorAntoine Pitrou <solipsis@pitrou.net>2011-07-23 19:50:21 (GMT)
committerAntoine Pitrou <solipsis@pitrou.net>2011-07-23 19:50:21 (GMT)
commitd42c1d09e99d74f6b411bdb1d3c8b9a648dd6a79 (patch)
treead9943d40fcf0ab23ae4d5ddeaacb71afab0d433 /Lib/test/test_io.py
parentf23339a7bb9d9fe977de6d90c64e3c9a27e6b00b (diff)
parente96ec6810184f5daacb2d47ab8801365c99bb206 (diff)
downloadcpython-d42c1d09e99d74f6b411bdb1d3c8b9a648dd6a79.zip
cpython-d42c1d09e99d74f6b411bdb1d3c8b9a648dd6a79.tar.gz
cpython-d42c1d09e99d74f6b411bdb1d3c8b9a648dd6a79.tar.bz2
Issue #12591: Allow io.TextIOWrapper to work with raw IO objects (without
a read1() method), and add a *write_through* parameter to mandate unbuffered writes.
Diffstat (limited to 'Lib/test/test_io.py')
-rw-r--r--Lib/test/test_io.py21
1 files changed, 21 insertions, 0 deletions
diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py
index 419921f..0d68ca1 100644
--- a/Lib/test/test_io.py
+++ b/Lib/test/test_io.py
@@ -2314,6 +2314,27 @@ class TextIOWrapperTest(unittest.TestCase):
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):