Expand Up @@ -65,6 +65,10 @@ SUPPORT_MULTITHREADING = False C_INT_MIN = -(2**31) C_INT_MAX = (2**31) - 1 def setUpModule(): global SUPPORT_MULTITHREADING SUPPORT_MULTITHREADING = CompressionParameter.nb_workers.bounds() != (0, 0) Expand Down Expand Up @@ -196,14 +200,21 @@ def test_simple_compress_bad_args(self): self.assertRaises(TypeError, ZstdCompressor, zstd_dict=b"abcd1234") self.assertRaises(TypeError, ZstdCompressor, zstd_dict={1: 2, 3: 4}) with self.assertRaises(ValueError): ZstdCompressor(2**31) with self.assertRaises(ValueError): ZstdCompressor(options={2**31: 100}) # 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(ZstdError ): with self.assertRaises(ValueError ): ZstdCompressor(options={CompressionParameter.window_log: 100}) with self.assertRaises(ZstdError ): with self.assertRaises(ValueError ): ZstdCompressor(options={3333: 100}) # Method bad arguments Expand Down Expand Up @@ -254,18 +265,32 @@ def test_compress_parameters(self): } ZstdCompressor(options=d) # larger than signed int, ValueError d1 = d.copy() d1[CompressionParameter.ldm_bucket_size_log] = 2**31 self.assertRaises(ValueError, ZstdCompressor, options=d1) # larger than signed int d1[CompressionParameter.ldm_bucket_size_log] = C_INT_MAX with self.assertRaises(ValueError): ZstdCompressor(options=d1) # smaller than signed int d1[CompressionParameter.ldm_bucket_size_log] = C_INT_MIN with self.assertRaises(ValueError): ZstdCompressor(options=d1) #clamp compressionLevel #out of bounds compression level level_min, level_max = CompressionParameter.compression_level.bounds() compress(b'', level_max+1) compress(b'', level_min-1) compress(b'', options={CompressionParameter.compression_level:level_max+1}) compress(b'', options={CompressionParameter.compression_level:level_min-1}) with self.assertRaises(ValueError): compress(b'', level_max+1) with self.assertRaises(ValueError): compress(b'', level_min-1) with self.assertRaises(ValueError): compress(b'', 2**1000) with self.assertRaises(ValueError): compress(b'', -(2**1000)) with self.assertRaises(ValueError): compress(b'', options={ CompressionParameter.compression_level: level_max+1}) with self.assertRaises(ValueError): compress(b'', options={ CompressionParameter.compression_level: level_min-1}) # zstd lib doesn't support MT compression if not SUPPORT_MULTITHREADING: Expand All @@ -278,19 +303,19 @@ def test_compress_parameters(self): # out of bounds error msg option = {CompressionParameter.window_log:100} with self.assertRaisesRegex(ZstdError, (r'Error when setting zstd compression parameter "window_log", ' r'it should \d+ <= value <= \d+, provided value is 100\. ' r'\((?:32|64)-bit build\)')): with self.assertRaisesRegex( ValueError, "compression parameter 'window_log' received an illegal value 100; " r'the valid range is \[-?\d+, -?\d+\]', ): compress(b'', options=option) def test_unknown_compression_parameter(self): KEY = 100001234 option = {CompressionParameter.compression_level: 10, KEY: 200000000} pattern = (r'Invalid zstd compression parameter.*?' fr'"unknown parameter \(key {KEY}\)"') with self.assertRaisesRegex(ZstdError, pattern): pattern = rf"invalid compression parameter 'unknown parameter \(key {KEY}\)'" with self.assertRaisesRegex(ValueError, pattern): ZstdCompressor(options=option) @unittest.skipIf(not SUPPORT_MULTITHREADING, Expand Down Expand Up @@ -385,12 +410,22 @@ def test_simple_decompress_bad_args(self): self.assertRaises(TypeError, ZstdDecompressor, options=b'abc') with self.assertRaises(ValueError): ZstdDecompressor(options={2**31 : 100}) ZstdDecompressor(options={C_INT_MAX: 100}) with self.assertRaises(ValueError): ZstdDecompressor(options={C_INT_MIN: 100}) with self.assertRaises(ValueError): ZstdDecompressor(options={0: C_INT_MAX}) with self.assertRaises(OverflowError): ZstdDecompressor(options={2**1000: 100}) with self.assertRaises(OverflowError): ZstdDecompressor(options={-(2**1000): 100}) with self.assertRaises(OverflowError): ZstdDecompressor(options={0: -(2**1000)}) with self.assertRaises(ZstdError ): ZstdDecompressor(options={DecompressionParameter.window_log_max:100}) with self.assertRaises(ZstdError ): ZstdDecompressor(options={3333 with self.assertRaises(ValueError ): ZstdDecompressor(options={DecompressionParameter.window_log_max: with self.assertRaises(ValueError ): ZstdDecompressor(options={3333: 100}) empty = compress(b'') lzd = ZstdDecompressor() Expand All @@ -403,26 +438,52 @@ def test_decompress_parameters(self): d = {DecompressionParameter.window_log_max : 15} ZstdDecompressor(options=d) # larger than signed int, ValueError d1 = d.copy() d1[DecompressionParameter.window_log_max] = 2**31 self.assertRaises(ValueError, ZstdDecompressor, None, d1) # larger than signed int d1[DecompressionParameter.window_log_max] = 2**1000 with self.assertRaises(OverflowError): ZstdDecompressor(None, d1) # smaller than signed int d1[DecompressionParameter.window_log_max] = -(2**1000) with self.assertRaises(OverflowError): ZstdDecompressor(None, d1) d1[DecompressionParameter.window_log_max] = C_INT_MAX with self.assertRaises(ValueError): ZstdDecompressor(None, d1) d1[DecompressionParameter.window_log_max] = C_INT_MIN with self.assertRaises(ValueError): ZstdDecompressor(None, d1) # out of bounds error msg options = {DecompressionParameter.window_log_max:100} with self.assertRaisesRegex(ZstdError, (r'Error when setting zstd decompression parameter "window_log_max", ' r'it should \d+ <= value <= \d+, provided value is 100\. ' r'\((?:32|64)-bit build\)')): with self.assertRaisesRegex( ValueError, "decompression parameter 'window_log_max' received an illegal value 100; " r'the valid range is \[-?\d+, -?\d+\]', ): decompress(b'', options=options) # out of bounds deecompression parameter options[DecompressionParameter.window_log_max] = C_INT_MAX with self.assertRaises(ValueError): decompress(b'', options=options) options[DecompressionParameter.window_log_max] = C_INT_MIN with self.assertRaises(ValueError): decompress(b'', options=options) options[DecompressionParameter.window_log_max] = 2**1000 with self.assertRaises(OverflowError): decompress(b'', options=options) options[DecompressionParameter.window_log_max] = -(2**1000) with self.assertRaises(OverflowError): decompress(b'', options=options) def test_unknown_decompression_parameter(self): KEY = 100001234 options = {DecompressionParameter.window_log_max: DecompressionParameter.window_log_max.bounds()[1], KEY: 200000000} pattern = (r'Invalid zstd decompression parameter.*?' fr'"unknown parameter \(key {KEY}\)"') with self.assertRaisesRegex(ZstdError, pattern): pattern = rf"invalid decompression parameter 'unknown parameter \(key {KEY}\)'" with self.assertRaisesRegex(ValueError, pattern): ZstdDecompressor(options=options) def test_decompress_epilogue_flags(self): Expand Down Expand Up @@ -1425,11 +1486,11 @@ def test_init_bad_mode(self): ZstdFile(io.BytesIO(COMPRESSED_100_PLUS_32KB), "rw") with self.assertRaisesRegex(TypeError, r"NOT be a CompressionParameter"): r"not be a CompressionParameter"): ZstdFile(io.BytesIO(), 'rb', options={CompressionParameter.compression_level:5}) with self.assertRaisesRegex(TypeError, r"NOT be a DecompressionParameter"): r"not be a DecompressionParameter"): ZstdFile(io.BytesIO(), 'wb', options={DecompressionParameter.window_log_max:21}) Expand All @@ -1440,19 +1501,19 @@ def test_init_bad_check(self): with self.assertRaises(TypeError): ZstdFile(io.BytesIO(), "w", level='asd') # CHECK_UNKNOWN and anything above CHECK_ID_MAX should be invalid. with self.assertRaises(ZstdError ): with self.assertRaises(ValueError ): ZstdFile(io.BytesIO(), "w", options={999:9999}) with self.assertRaises(ZstdError ): with self.assertRaises(ValueError ): ZstdFile(io.BytesIO(), "w", options={CompressionParameter.window_log:99}) with self.assertRaises(TypeError): ZstdFile(io.BytesIO(COMPRESSED_100_PLUS_32KB), "r", options=33) with self.assertRaises(ValueError ): with self.assertRaises(OverflowError ): ZstdFile(io.BytesIO(COMPRESSED_100_PLUS_32KB), options={DecompressionParameter.window_log_max:2**31}) with self.assertRaises(ZstdError ): with self.assertRaises(ValueError ): ZstdFile(io.BytesIO(COMPRESSED_100_PLUS_32KB), options={444:333}) Expand All @@ -1468,7 +1529,7 @@ def test_init_close_fp(self): tmp_f.write(DAT_130K_C) filename = tmp_f.name with self.assertRaises(ValueError ): with self.assertRaises(TypeError ): ZstdFile(filename, options={'a':'b'}) # for PyPy Expand Down