(self)
| 30 | self.assertEqual(decompressed, TEXT * NTHREADS) |
| 31 | |
| 32 | def test_decompressor(self): |
| 33 | chunk_size = 128 |
| 34 | chunks = [bytes([ord("a") + i]) * chunk_size for i in range(NTHREADS)] |
| 35 | input_data = b"".join(chunks) |
| 36 | compressed = bz2.compress(input_data) |
| 37 | |
| 38 | bz2d = BZ2Decompressor() |
| 39 | output = [] |
| 40 | |
| 41 | def worker(): |
| 42 | data = bz2d.decompress(compressed, chunk_size) |
| 43 | self.assertEqual(len(data), chunk_size) |
| 44 | output.append(data) |
| 45 | # Read attributes concurrently with other threads decompressing |
| 46 | self.assertIsInstance(bz2d.eof, bool) |
| 47 | self.assertIsInstance(bz2d.needs_input, bool) |
| 48 | self.assertIsInstance(bz2d.unused_data, bytes) |
| 49 | |
| 50 | run_concurrently(worker_func=worker, nthreads=NTHREADS) |
| 51 | self.assertEqual(len(output), NTHREADS) |
| 52 | # Verify the expected chunks (order doesn't matter due to append race) |
| 53 | self.assertEqual(set(output), set(chunks)) |
| 54 | self.assertTrue(bz2d.eof) |
| 55 | self.assertFalse(bz2d.needs_input) |
| 56 | # Each thread added full compressed data to the buffer, but only 1 copy |
| 57 | # is consumed to produce the output. The rest remains as unused_data. |
| 58 | self.assertEqual( |
| 59 | len(bz2d.unused_data), len(compressed) * (NTHREADS - 1) |
| 60 | ) |
| 61 | |
| 62 | |
| 63 | if __name__ == "__main__": |
nothing calls this directly
no test coverage detected