(self, intermediate_func)
| 1925 | self.assertStartsWith(flushed, contents[:-8]) |
| 1926 | |
| 1927 | def check_writes(self, intermediate_func): |
| 1928 | # Lots of writes, test the flushed output is as expected. |
| 1929 | contents = bytes(range(256)) * 1000 |
| 1930 | n = 0 |
| 1931 | writer = self.MockRawIO() |
| 1932 | bufio = self.tp(writer, 13) |
| 1933 | # Generator of write sizes: repeat each N 15 times then proceed to N+1 |
| 1934 | def gen_sizes(): |
| 1935 | for size in count(1): |
| 1936 | for i in range(15): |
| 1937 | yield size |
| 1938 | sizes = gen_sizes() |
| 1939 | while n < len(contents): |
| 1940 | size = min(next(sizes), len(contents) - n) |
| 1941 | self.assertEqual(bufio.write(contents[n:n+size]), size) |
| 1942 | intermediate_func(bufio) |
| 1943 | n += size |
| 1944 | bufio.flush() |
| 1945 | self.assertEqual(contents, b"".join(writer._write_stack)) |
| 1946 | |
| 1947 | def test_writes(self): |
| 1948 | self.check_writes(lambda bufio: None) |
no test coverage detected