Mercurial > public > mercurial-scm > hg
diff contrib/python-zstandard/tests/test_data_structures_fuzzing.py @ 31796:e0dc40530c5a
zstd: vendor python-zstandard 0.8.0
Commit 81e1f5bbf1fc54808649562d3ed829730765c540 from
https://github.com/indygreg/python-zstandard is imported without
modifications (other than removing unwanted files).
Updates relevant to Mercurial include:
* Support for multi-threaded compression (we can use this for
bundle and wire protocol compression).
* APIs for batch compression and decompression operations using
multiple threads and optimal memory allocation mechanism. (Can
be useful for revlog perf improvements.)
* A ``BufferWithSegments`` type that models a single memory buffer
containing N discrete items of known lengths. This type can be
used for very efficient 0-copy data operations.
# no-check-commit
| author | Gregory Szorc <gregory.szorc@gmail.com> |
|---|---|
| date | Sat, 01 Apr 2017 15:24:03 -0700 |
| parents | |
| children | b1fb341d8a61 |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/contrib/python-zstandard/tests/test_data_structures_fuzzing.py Sat Apr 01 15:24:03 2017 -0700 @@ -0,0 +1,79 @@ +import io +import os + +try: + import unittest2 as unittest +except ImportError: + import unittest + +try: + import hypothesis + import hypothesis.strategies as strategies +except ImportError: + raise unittest.SkipTest('hypothesis not available') + +import zstd + +from .common import ( + make_cffi, +) + + +s_windowlog = strategies.integers(min_value=zstd.WINDOWLOG_MIN, + max_value=zstd.WINDOWLOG_MAX) +s_chainlog = strategies.integers(min_value=zstd.CHAINLOG_MIN, + max_value=zstd.CHAINLOG_MAX) +s_hashlog = strategies.integers(min_value=zstd.HASHLOG_MIN, + max_value=zstd.HASHLOG_MAX) +s_searchlog = strategies.integers(min_value=zstd.SEARCHLOG_MIN, + max_value=zstd.SEARCHLOG_MAX) +s_searchlength = strategies.integers(min_value=zstd.SEARCHLENGTH_MIN, + max_value=zstd.SEARCHLENGTH_MAX) +s_targetlength = strategies.integers(min_value=zstd.TARGETLENGTH_MIN, + max_value=zstd.TARGETLENGTH_MAX) +s_strategy = strategies.sampled_from((zstd.STRATEGY_FAST, + zstd.STRATEGY_DFAST, + zstd.STRATEGY_GREEDY, + zstd.STRATEGY_LAZY, + zstd.STRATEGY_LAZY2, + zstd.STRATEGY_BTLAZY2, + zstd.STRATEGY_BTOPT)) + + +@make_cffi +@unittest.skipUnless('ZSTD_SLOW_TESTS' in os.environ, 'ZSTD_SLOW_TESTS not set') +class TestCompressionParametersHypothesis(unittest.TestCase): + @hypothesis.given(s_windowlog, s_chainlog, s_hashlog, s_searchlog, + s_searchlength, s_targetlength, s_strategy) + def test_valid_init(self, windowlog, chainlog, hashlog, searchlog, + searchlength, targetlength, strategy): + # ZSTD_checkCParams moves the goal posts on us from what's advertised + # in the constants. So move along with them. + if searchlength == zstd.SEARCHLENGTH_MIN and strategy in (zstd.STRATEGY_FAST, zstd.STRATEGY_GREEDY): + searchlength += 1 + elif searchlength == zstd.SEARCHLENGTH_MAX and strategy != zstd.STRATEGY_FAST: + searchlength -= 1 + + p = zstd.CompressionParameters(windowlog, chainlog, hashlog, + searchlog, searchlength, + targetlength, strategy) + + cctx = zstd.ZstdCompressor(compression_params=p) + with cctx.write_to(io.BytesIO()): + pass + + @hypothesis.given(s_windowlog, s_chainlog, s_hashlog, s_searchlog, + s_searchlength, s_targetlength, s_strategy) + def test_estimate_compression_context_size(self, windowlog, chainlog, + hashlog, searchlog, + searchlength, targetlength, + strategy): + if searchlength == zstd.SEARCHLENGTH_MIN and strategy in (zstd.STRATEGY_FAST, zstd.STRATEGY_GREEDY): + searchlength += 1 + elif searchlength == zstd.SEARCHLENGTH_MAX and strategy != zstd.STRATEGY_FAST: + searchlength -= 1 + + p = zstd.CompressionParameters(windowlog, chainlog, hashlog, + searchlog, searchlength, + targetlength, strategy) + size = zstd.estimate_compression_context_size(p)