Method test_simple_compress_bad_args

Lib/test/test_zstd.py:194–241

(self)

class CompressorTestCase(unittest.TestCase):

    def test_simple_compress_bad_args(self):
        # ZstdCompressor
        self.assertRaises(TypeError, ZstdCompressor, [])
        self.assertRaises(TypeError, ZstdCompressor, level=3.14)
        self.assertRaises(TypeError, ZstdCompressor, level="abc")
        self.assertRaises(TypeError, ZstdCompressor, options=b"abc")

        self.assertRaises(TypeError, ZstdCompressor, zstd_dict=123)
        self.assertRaises(TypeError, ZstdCompressor, zstd_dict=b"abcd1234")
        self.assertRaises(TypeError, ZstdCompressor, zstd_dict={1: 2, 3: 4})

        # valid range for compression level is [-(1<<17), 22]
        msg = r'illegal compression level {}; the valid range is \[-?\d+, -?\d+\]'
        with self.assertRaisesRegex(ValueError, msg.format(C_INT_MAX)):
            ZstdCompressor(C_INT_MAX)
        with self.assertRaisesRegex(ValueError, msg.format(C_INT_MIN)):
            ZstdCompressor(C_INT_MIN)
        msg = r'illegal compression level; the valid range is \[-?\d+, -?\d+\]'
        with self.assertRaisesRegex(ValueError, msg):
            ZstdCompressor(level=-(2**1000))
        with self.assertRaisesRegex(ValueError, msg):
            ZstdCompressor(level=2**1000)

        with self.assertRaises(ValueError):
            ZstdCompressor(options={CompressionParameter.window_log: 100})
        with self.assertRaises(ValueError):
            ZstdCompressor(options={3333: 100})

        # Method bad arguments
        zc = ZstdCompressor()
        self.assertRaises(TypeError, zc.compress)
        self.assertRaises((TypeError, ValueError), zc.compress, b"foo", b"bar")
        self.assertRaises(TypeError, zc.compress, "str")
        self.assertRaises((TypeError, ValueError), zc.flush, b"foo")
        self.assertRaises(TypeError, zc.flush, b"blah", 1)

        self.assertRaises(ValueError, zc.compress, b'', -1)
        self.assertRaises(ValueError, zc.compress, b'', 3)
        self.assertRaises(ValueError, zc.flush, zc.CONTINUE) # 0
        self.assertRaises(ValueError, zc.flush, 3)

        zc.compress(b'')
        zc.compress(b'', zc.CONTINUE)
        zc.compress(b'', zc.FLUSH_BLOCK)
        zc.compress(b'', zc.FLUSH_FRAME)
        empty = zc.flush()
        zc.flush(zc.FLUSH_BLOCK)
        zc.flush(zc.FLUSH_FRAME)

    def test_compress_parameters(self):
        d = {CompressionParameter.compression_level : 10,