[issue4428] make io.BufferedWriter observe max_buffer_size limits - Code Review (original) (raw)

OLD

NEW

1 """Unit tests for io.py."""

1 """Unit tests for io.py."""

2

2

3 import os

3 import os

4 import sys

4 import sys

5 import time

5 import time

6 import array

6 import array

7 import threading

7 import threading

8 import random

8 import random

9 import unittest

9 import unittest

10 from itertools import chain, cycle

10 from itertools import chain, cycle

(...skipping 461 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading...

472

472

473 def testWriteOverflow(self):

473 def testWriteOverflow(self):

474 writer = MockRawIO()

474 writer = MockRawIO()

475 bufio = io.BufferedWriter(writer, 8)

475 bufio = io.BufferedWriter(writer, 8)

476

476

477 bufio.write(b"abc")

477 bufio.write(b"abc")

478 bufio.write(b"defghijkl")

478 bufio.write(b"defghijkl")

479

479

480 self.assertEquals(b"abcdefghijkl", writer._write_stack[0])

480 self.assertEquals(b"abcdefghijkl", writer._write_stack[0])

481

481

482 def testWriteMaxBufferSize(self):

483 writer = MockRawIO()

484 bufio = io.BufferedWriter(writer, buffer_size=8, max_buffer_size=13)

485

486 bufio.write(b"abcdefghij")

487 bufio.write(b"klmnopqrstuvwxyz")

488 bufio.write(b"0123456789"*3)

489

490 expected_writes = [b"abcdefghij", b"klmnopqrstuvw", b"xyz0123456789",

491 b"0123456789012"]

492 self.assertEquals(expected_writes, writer._write_stack)

493 bufio.close()

494 self.assertEquals(b"3456789", writer._write_stack[4])

495

496 def testWriteNoLengthIterable(self):

497 writer = MockRawIO()

498 bufio = io.BufferedWriter(writer, buffer_size=8)

499

500 def Generator(value):

501 for x in value:

502 yield x

503 iterable = Generator(b'ABCDEFGHIJKLMNOPQRSTUVWXYZ')

504 self.assertRaises(TypeError, len, iterable)

505 bufio.write(iterable)

506

507 self.assertEquals(b'ABCDEFGHIJKLMNOPQRSTUVWXYZ', writer._write_stack[0])

508

482 def testWriteNonBlocking(self):

509 def testWriteNonBlocking(self):

483 raw = MockNonBlockWriterIO((9, 2, 22, -6, 10, 12, 12))

510 raw = MockNonBlockWriterIO((9, 2, 22, -6, 10, 12, 12))

484 bufio = io.BufferedWriter(raw, 8, 16)

511 bufio = io.BufferedWriter(raw, 8, 16)

485

512

486 bufio.write(b"asdf")

513 bufio.write(b"asdf")

487 bufio.write(b"asdfa")

514 bufio.write(b"asdfa")

488 self.assertEquals(b"asdfasdfa", raw._write_stack[0])

515 self.assertEquals(b"asdfasdfa", raw._write_stack[0])

489

516

490 bufio.write(b"asdfasdfasdf")

517 bufio.write(b"asdfasdfasdf")

491 self.assertEquals(b"asdfasdfasdf", raw._write_stack[1])

518 self.assertEquals(b"asdfasdfasdf", raw._write_stack[1])

(...skipping 878 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading...

1370

1397

1371 def test_main():

1398 def test_main():

1372 support.run_unittest(IOTest, BytesIOTest, StringIOTest,

1399 support.run_unittest(IOTest, BytesIOTest, StringIOTest,

1373 BufferedReaderTest, BufferedWriterTest,

1400 BufferedReaderTest, BufferedWriterTest,

1374 BufferedRWPairTest, BufferedRandomTest,

1401 BufferedRWPairTest, BufferedRandomTest,

1375 StatefulIncrementalDecoderTest,

1402 StatefulIncrementalDecoderTest,

1376 TextIOWrapperTest, MiscIOTest)

1403 TextIOWrapperTest, MiscIOTest)

1377

1404

1378 if __name__ == "__main__":

1405 if __name__ == "__main__":

1379 unittest.main()

1406 unittest.main()

OLD

NEW