mirror of
https://github.com/python/cpython.git
synced 2025-10-09 16:34:44 +00:00
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
This commit is contained in:
parent
84f7afe65c
commit
01806d5beb
3 changed files with 46 additions and 3 deletions
|
@ -3767,6 +3767,33 @@ class CTextIOWrapperTest(TextIOWrapperTest):
|
|||
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
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue