(self)
| 15 | @threading_helper.requires_working_threading() |
| 16 | class TestZlib(unittest.TestCase): |
| 17 | def test_compressor(self): |
| 18 | comp = zlib.compressobj() |
| 19 | |
| 20 | # First compress() outputs zlib header |
| 21 | header = comp.compress(HAMLET_SCENE) |
| 22 | self.assertGreater(len(header), 0) |
| 23 | |
| 24 | def worker(): |
| 25 | # it should return empty bytes as it buffers data internally |
| 26 | data = comp.compress(HAMLET_SCENE) |
| 27 | self.assertEqual(data, b"") |
| 28 | |
| 29 | run_concurrently(worker_func=worker, nthreads=NTHREADS - 1) |
| 30 | full_compressed = header + comp.flush() |
| 31 | decompressed = zlib.decompress(full_compressed) |
| 32 | # The decompressed data should be HAMLET_SCENE repeated NTHREADS times |
| 33 | self.assertEqual(decompressed, HAMLET_SCENE * NTHREADS) |
| 34 | |
| 35 | def test_decompressor_concurrent_attribute_reads(self): |
| 36 | input_data = HAMLET_SCENE * NTHREADS |
nothing calls this directly
no test coverage detected