(self)
| 855 | self.assertEqual(read_message, message) |
| 856 | |
| 857 | def test_flush_modes(self): |
| 858 | # Make sure the argument to flush is properly passed to the |
| 859 | # zlib.compressobj; see issue GH-105808. |
| 860 | class FakeCompressor: |
| 861 | def __init__(self): |
| 862 | self.modes = [] |
| 863 | def compress(self, data): |
| 864 | return b'' |
| 865 | def flush(self, mode=-1): |
| 866 | self.modes.append(mode) |
| 867 | return b'' |
| 868 | b = io.BytesIO() |
| 869 | fc = FakeCompressor() |
| 870 | with gzip.GzipFile(fileobj=b, mode='w') as f: |
| 871 | f.compress = fc |
| 872 | f.flush() |
| 873 | f.flush(50) |
| 874 | f.flush(zlib_mode=100) |
| 875 | # The implicit close will also flush the compressor. |
| 876 | expected_modes = [ |
| 877 | zlib.Z_SYNC_FLUSH, |
| 878 | 50, |
| 879 | 100, |
| 880 | -1, |
| 881 | ] |
| 882 | self.assertEqual(fc.modes, expected_modes) |
| 883 | |
| 884 | def test_write_seek_write(self): |
| 885 | # Make sure that offset is up-to-date before seeking |
nothing calls this directly
no test coverage detected