bpo-43260: io: Prevent large data remains in textio buffer. (GH-24592)
When very large data remains in TextIOWrapper, flush() may fail forever.
So prevent that data larger than chunk_size is remained in TextIOWrapper internal
buffer.
Co-Authored-By: Eryk Sun
diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py
index cc54d0e..3768b62 100644
--- a/Lib/test/test_io.py
+++ b/Lib/test/test_io.py
@@ -3767,6 +3767,33 @@ def test_del__CHUNK_SIZE_SystemError(self):
with self.assertRaises(AttributeError):
del t._CHUNK_SIZE
+ def test_internal_buffer_size(self):
+ # bpo-43260: TextIOWrapper's internal buffer should not store
+ # data larger than chunk size.
+ chunk_size = 8192 # default chunk size, updated later
+
+ class MockIO(self.MockRawIO):
+ def write(self, data):
+ if len(data) > chunk_size:
+ raise RuntimeError
+ return super().write(data)
+
+ buf = MockIO()
+ t = self.TextIOWrapper(buf, encoding="ascii")
+ chunk_size = t._CHUNK_SIZE
+ t.write("abc")
+ t.write("def")
+ # default chunk size is 8192 bytes so t don't write data to buf.
+ self.assertEqual([], buf._write_stack)
+
+ with self.assertRaises(RuntimeError):
+ t.write("x"*(chunk_size+1))
+
+ self.assertEqual([b"abcdef"], buf._write_stack)
+ t.write("ghi")
+ t.write("x"*chunk_size)
+ self.assertEqual([b"abcdef", b"ghi", b"x"*chunk_size], buf._write_stack)
+
class PyTextIOWrapperTest(TextIOWrapperTest):
io = pyio