diff contrib/python-zstandard/tests/test_data_structures.py @ 37495:b1fb341d8a61

zstandard: vendor python-zstandard 0.9.0 This was just released. It features a number of goodies. More info at https://gregoryszorc.com/blog/2018/04/09/release-of-python-zstandard-0.9/. The clang-format ignore list was updated to reflect the new source of files. The project contains a vendored copy of zstandard 1.3.4. The old version was 1.1.3. One of the changes between those versions is that zstandard is now dual licensed BSD + GPLv2 and the patent rights grant has been removed. Good riddance. The API should be backwards compatible. So no changes in core should be needed. However, there were a number of changes in the library that we'll want to adapt to. Those will be addressed in subsequent commits. Differential Revision: https://phab.mercurial-scm.org/D3198
author Gregory Szorc <gregory.szorc@gmail.com>
date Mon, 09 Apr 2018 10:13:29 -0700
parents e0dc40530c5a
children 73fef626dae3
line wrap: on
line diff
--- a/contrib/python-zstandard/tests/test_data_structures.py	Sun Apr 08 01:08:43 2018 +0200
+++ b/contrib/python-zstandard/tests/test_data_structures.py	Mon Apr 09 10:13:29 2018 -0700
@@ -1,9 +1,7 @@
-try:
-    import unittest2 as unittest
-except ImportError:
-    import unittest
+import sys
+import unittest
 
-import zstd
+import zstandard as zstd
 
 from . common import (
     make_cffi,
@@ -12,52 +10,104 @@
 
 @make_cffi
 class TestCompressionParameters(unittest.TestCase):
-    def test_init_bad_arg_type(self):
-        with self.assertRaises(TypeError):
-            zstd.CompressionParameters()
-
-        with self.assertRaises(TypeError):
-            zstd.CompressionParameters(0, 1)
+    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)
 
-    def test_bounds(self):
-        zstd.CompressionParameters(zstd.WINDOWLOG_MIN,
-                                   zstd.CHAINLOG_MIN,
-                                   zstd.HASHLOG_MIN,
-                                   zstd.SEARCHLOG_MIN,
-                                   zstd.SEARCHLENGTH_MIN + 1,
-                                   zstd.TARGETLENGTH_MIN,
-                                   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)
 
-        zstd.CompressionParameters(zstd.WINDOWLOG_MAX,
-                                   zstd.CHAINLOG_MAX,
-                                   zstd.HASHLOG_MAX,
-                                   zstd.SEARCHLOG_MAX,
-                                   zstd.SEARCHLENGTH_MAX - 1,
-                                   zstd.TARGETLENGTH_MAX,
-                                   zstd.STRATEGY_BTOPT)
-
-    def test_get_compression_parameters(self):
-        p = zstd.get_compression_parameters(1)
+    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.CompressionParameters(10, 6, 7, 4, 5, 8, 1)
+        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.search_length, 5)
+        self.assertEqual(p.min_match, 5)
         self.assertEqual(p.target_length, 8)
-        self.assertEqual(p.strategy, 1)
+        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.CompressionParameters(20, 16, 17,  1,  5, 16, zstd.STRATEGY_DFAST)
+        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(), 1287076,
-                               delta=110)
+        self.assertAlmostEqual(p.estimated_compression_context_size(), 1294072,
+                               delta=250)
 
 
 @make_cffi
@@ -66,8 +116,18 @@
         with self.assertRaises(TypeError):
             zstd.get_frame_parameters(None)
 
-        with self.assertRaises(TypeError):
-            zstd.get_frame_parameters(u'foobarbaz')
+        # 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'):
@@ -82,21 +142,21 @@
 
     def test_attributes(self):
         params = zstd.get_frame_parameters(zstd.FRAME_HEADER + b'\x00\x00')
-        self.assertEqual(params.content_size, 0)
+        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, 0)
+        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, 0)
+        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)
@@ -110,7 +170,7 @@
 
         # Window descriptor is 2nd byte after frame header.
         params = zstd.get_frame_parameters(zstd.FRAME_HEADER + b'\x00\x40')
-        self.assertEqual(params.content_size, 0)
+        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)
@@ -121,3 +181,22 @@
         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)