(self)
| 834 | self.assertEqual(f.tell(), LENGTH) |
| 835 | |
| 836 | def test_flush_flushes_compressor(self): |
| 837 | # See issue GH-105808. |
| 838 | b = io.BytesIO() |
| 839 | message = b"important message here." |
| 840 | with gzip.GzipFile(fileobj=b, mode='w') as f: |
| 841 | f.write(message) |
| 842 | f.flush() |
| 843 | partial_data = b.getvalue() |
| 844 | full_data = b.getvalue() |
| 845 | self.assertEqual(gzip.decompress(full_data), message) |
| 846 | # The partial data should contain the gzip header and the complete |
| 847 | # message, but not the end-of-stream markers (so we can't just |
| 848 | # decompress it directly). |
| 849 | with self.assertRaises(EOFError): |
| 850 | gzip.decompress(partial_data) |
| 851 | d = zlib.decompressobj(wbits=-zlib.MAX_WBITS) |
| 852 | f = io.BytesIO(partial_data) |
| 853 | gzip._read_gzip_header(f) |
| 854 | read_message = d.decompress(f.read()) |
| 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 |
nothing calls this directly
no test coverage detected