(self, intermediate_func)
| 708 | self.assertStartsWith(flushed, contents[:-8]) |
| 709 | |
| 710 | def check_writes(self, intermediate_func): |
| 711 | # Lots of writes, test the flushed output is as expected. |
| 712 | contents = bytes(range(256)) * 1000 |
| 713 | n = 0 |
| 714 | writer = self.MockRawIO() |
| 715 | bufio = self.tp(writer, 13) |
| 716 | # Generator of write sizes: repeat each N 15 times then proceed to N+1 |
| 717 | def gen_sizes(): |
| 718 | for size in count(1): |
| 719 | for i in range(15): |
| 720 | yield size |
| 721 | sizes = gen_sizes() |
| 722 | while n < len(contents): |
| 723 | size = min(next(sizes), len(contents) - n) |
| 724 | self.assertEqual(bufio.write(contents[n:n+size]), size) |
| 725 | intermediate_func(bufio) |
| 726 | n += size |
| 727 | bufio.flush() |
| 728 | self.assertEqual(contents, b"".join(writer._write_stack)) |
| 729 | |
| 730 | def test_writes(self): |
| 731 | self.check_writes(lambda bufio: None) |
no test coverage detected