import sys import unittest import zstandard as zstd from . common import ( make_cffi, ) @make_cffi class TestCompressionParameters(unittest.TestCase): def test_bounds(self): zstd.ZstdCompressionParameters(window_log=zstd.WINDOWLOG_MIN, chain_log=zstd.CHAINLOG_MIN, hash_log=zstd.HASHLOG_MIN, search_log=zstd.SEARCHLOG_MIN, min_match=zstd.SEARCHLENGTH_MIN + 1, target_length=zstd.TARGETLENGTH_MIN, compression_strategy=zstd.STRATEGY_FAST) zstd.ZstdCompressionParameters(window_log=zstd.WINDOWLOG_MAX, chain_log=zstd.CHAINLOG_MAX, hash_log=zstd.HASHLOG_MAX, search_log=zstd.SEARCHLOG_MAX, min_match=zstd.SEARCHLENGTH_MAX - 1, compression_strategy=zstd.STRATEGY_BTULTRA) def test_from_level(self): p = zstd.ZstdCompressionParameters.from_level(1) self.assertIsInstance(p, zstd.CompressionParameters) self.assertEqual(p.window_log, 19) p = zstd.ZstdCompressionParameters.from_level(-4) self.assertEqual(p.window_log, 19) self.assertEqual(p.compress_literals, 0) def test_members(self): p = zstd.ZstdCompressionParameters(window_log=10, chain_log=6, hash_log=7, search_log=4, min_match=5, target_length=8, compression_strategy=1) self.assertEqual(p.window_log, 10) self.assertEqual(p.chain_log, 6) self.assertEqual(p.hash_log, 7) self.assertEqual(p.search_log, 4) self.assertEqual(p.min_match, 5) self.assertEqual(p.target_length, 8) self.assertEqual(p.compression_strategy, 1) p = zstd.ZstdCompressionParameters(compression_level=2) self.assertEqual(p.compression_level, 2) p = zstd.ZstdCompressionParameters(threads=4) self.assertEqual(p.threads, 4) p = zstd.ZstdCompressionParameters(threads=2, job_size=1048576, overlap_size_log=6) self.assertEqual(p.threads, 2) self.assertEqual(p.job_size, 1048576) self.assertEqual(p.overlap_size_log, 6) p = zstd.ZstdCompressionParameters(compression_level=2) self.assertEqual(p.compress_literals, 1) p = zstd.ZstdCompressionParameters(compress_literals=False) self.assertEqual(p.compress_literals, 0) p = zstd.ZstdCompressionParameters(compression_level=-1) self.assertEqual(p.compression_level, -1) self.assertEqual(p.compress_literals, 0) p = zstd.ZstdCompressionParameters(compression_level=-2, compress_literals=True) self.assertEqual(p.compression_level, -2) self.assertEqual(p.compress_literals, 1) p = zstd.ZstdCompressionParameters(force_max_window=True) self.assertEqual(p.force_max_window, 1) p = zstd.ZstdCompressionParameters(enable_ldm=True) self.assertEqual(p.enable_ldm, 1) p = zstd.ZstdCompressionParameters(ldm_hash_log=7) self.assertEqual(p.ldm_hash_log, 7) p = zstd.ZstdCompressionParameters(ldm_min_match=6) self.assertEqual(p.ldm_min_match, 6) p = zstd.ZstdCompressionParameters(ldm_bucket_size_log=7) self.assertEqual(p.ldm_bucket_size_log, 7) p = zstd.ZstdCompressionParameters(ldm_hash_every_log=8) self.assertEqual(p.ldm_hash_every_log, 8) def test_estimated_compression_context_size(self): p = zstd.ZstdCompressionParameters(window_log=20, chain_log=16, hash_log=17, search_log=1, min_match=5, target_length=16, compression_strategy=zstd.STRATEGY_DFAST) # 32-bit has slightly different values from 64-bit. self.assertAlmostEqual(p.estimated_compression_context_size(), 1294072, delta=250) @make_cffi class TestFrameParameters(unittest.TestCase): def test_invalid_type(self): with self.assertRaises(TypeError): zstd.get_frame_parameters(None) # Python 3 doesn't appear to convert unicode to Py_buffer. if sys.version_info[0] >= 3: with self.assertRaises(TypeError): zstd.get_frame_parameters(u'foobarbaz') else: # CPython will convert unicode to Py_buffer. But CFFI won't. if zstd.backend == 'cffi': with self.assertRaises(TypeError): zstd.get_frame_parameters(u'foobarbaz') else: with self.assertRaises(zstd.ZstdError): zstd.get_frame_parameters(u'foobarbaz') def test_invalid_input_sizes(self): with self.assertRaisesRegexp(zstd.ZstdError, 'not enough data for frame'): zstd.get_frame_parameters(b'') with self.assertRaisesRegexp(zstd.ZstdError, 'not enough data for frame'): zstd.get_frame_parameters(zstd.FRAME_HEADER) def test_invalid_frame(self): with self.assertRaisesRegexp(zstd.ZstdError, 'Unknown frame descriptor'): zstd.get_frame_parameters(b'foobarbaz') def test_attributes(self): params = zstd.get_frame_parameters(zstd.FRAME_HEADER + b'\x00\x00') self.assertEqual(params.content_size, zstd.CONTENTSIZE_UNKNOWN) self.assertEqual(params.window_size, 1024) self.assertEqual(params.dict_id, 0) self.assertFalse(params.has_checksum) # Lowest 2 bits indicate a dictionary and length. Here, the dict id is 1 byte. params = zstd.get_frame_parameters(zstd.FRAME_HEADER + b'\x01\x00\xff') self.assertEqual(params.content_size, zstd.CONTENTSIZE_UNKNOWN) self.assertEqual(params.window_size, 1024) self.assertEqual(params.dict_id, 255) self.assertFalse(params.has_checksum) # Lowest 3rd bit indicates if checksum is present. params = zstd.get_frame_parameters(zstd.FRAME_HEADER + b'\x04\x00') self.assertEqual(params.content_size, zstd.CONTENTSIZE_UNKNOWN) self.assertEqual(params.window_size, 1024) self.assertEqual(params.dict_id, 0) self.assertTrue(params.has_checksum) # Upper 2 bits indicate content size. params = zstd.get_frame_parameters(zstd.FRAME_HEADER + b'\x40\x00\xff\x00') self.assertEqual(params.content_size, 511) self.assertEqual(params.window_size, 1024) self.assertEqual(params.dict_id, 0) self.assertFalse(params.has_checksum) # Window descriptor is 2nd byte after frame header. params = zstd.get_frame_parameters(zstd.FRAME_HEADER + b'\x00\x40') self.assertEqual(params.content_size, zstd.CONTENTSIZE_UNKNOWN) self.assertEqual(params.window_size, 262144) self.assertEqual(params.dict_id, 0) self.assertFalse(params.has_checksum) # Set multiple things. params = zstd.get_frame_parameters(zstd.FRAME_HEADER + b'\x45\x40\x0f\x10\x00') self.assertEqual(params.content_size, 272) self.assertEqual(params.window_size, 262144) self.assertEqual(params.dict_id, 15) self.assertTrue(params.has_checksum) def test_input_types(self): v = zstd.FRAME_HEADER + b'\x00\x00' mutable_array = bytearray(len(v)) mutable_array[:] = v sources = [ memoryview(v), bytearray(v), mutable_array, ] for source in sources: params = zstd.get_frame_parameters(source) self.assertEqual(params.content_size, zstd.CONTENTSIZE_UNKNOWN) self.assertEqual(params.window_size, 1024) self.assertEqual(params.dict_id, 0) self.assertFalse(params.has_checksum)