Method test_compressor

Lib/test/test_free_threading/test_bz2.py:18–30

(self)

@threading_helper.requires_working_threading()
class TestBZ2(unittest.TestCase):
    def test_compressor(self):
        bz2c = BZ2Compressor()

        def worker():
            # it should return empty bytes as it buffers data internally
            data = bz2c.compress(TEXT)
            self.assertEqual(data, b"")

        run_concurrently(worker_func=worker, nthreads=NTHREADS)
        data = bz2c.flush()
        # The decompressed data should be TEXT repeated NTHREADS times
        decompressed = ext_decompress(data)
        self.assertEqual(decompressed, TEXT * NTHREADS)

    def test_decompressor(self):
        chunk_size = 128