Method test_compressor

Lib/test/test_free_threading/test_lzma.py:17–33

(self)

@threading_helper.requires_working_threading()
class TestLZMA(unittest.TestCase):
    def test_compressor(self):
        lzc = LZMACompressor()

        # First compress() outputs LZMA header
        header = lzc.compress(INPUT)
        self.assertGreater(len(header), 0)

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

        run_concurrently(worker_func=worker, nthreads=NTHREADS - 1)
        full_compressed = header + lzc.flush()
        decompressed = lzma.decompress(full_compressed)
        # The decompressed data should be INPUT repeated NTHREADS times
        self.assertEqual(decompressed, INPUT * NTHREADS)

    def test_decompressor(self):
        chunk_size = 128