1 import io

     2 import os

     3 

     4 try:

     5     import unittest2 as unittest

     6 except ImportError:

     7     import unittest

     8 

     9 try:

    10     import hypothesis

    11     import hypothesis.strategies as strategies

    12 except ImportError:

    13     raise unittest.SkipTest('hypothesis not available')

    14 

    15 import zstd

    16 

    17 from .common import (

    18     make_cffi,

    19 )

    20 

    21 

    22 s_windowlog = strategies.integers(min_value=zstd.WINDOWLOG_MIN,

    23                                     max_value=zstd.WINDOWLOG_MAX)

    24 s_chainlog = strategies.integers(min_value=zstd.CHAINLOG_MIN,

    25                                     max_value=zstd.CHAINLOG_MAX)

    26 s_hashlog = strategies.integers(min_value=zstd.HASHLOG_MIN,

    27                                 max_value=zstd.HASHLOG_MAX)

    28 s_searchlog = strategies.integers(min_value=zstd.SEARCHLOG_MIN,

    29                                     max_value=zstd.SEARCHLOG_MAX)

    30 s_searchlength = strategies.integers(min_value=zstd.SEARCHLENGTH_MIN,

    31                                         max_value=zstd.SEARCHLENGTH_MAX)

    32 s_targetlength = strategies.integers(min_value=zstd.TARGETLENGTH_MIN,

    33                                         max_value=zstd.TARGETLENGTH_MAX)

    34 s_strategy = strategies.sampled_from((zstd.STRATEGY_FAST,

    35                                         zstd.STRATEGY_DFAST,

    36                                         zstd.STRATEGY_GREEDY,

    37                                         zstd.STRATEGY_LAZY,

    38                                         zstd.STRATEGY_LAZY2,

    39                                         zstd.STRATEGY_BTLAZY2,

    40                                         zstd.STRATEGY_BTOPT))

    41 

    42 

    43 @make_cffi

    44 @unittest.skipUnless('ZSTD_SLOW_TESTS' in os.environ, 'ZSTD_SLOW_TESTS not set')

    45 class TestCompressionParametersHypothesis(unittest.TestCase):

    46     @hypothesis.given(s_windowlog, s_chainlog, s_hashlog, s_searchlog,

    47                         s_searchlength, s_targetlength, s_strategy)

    48     def test_valid_init(self, windowlog, chainlog, hashlog, searchlog,

    49                         searchlength, targetlength, strategy):

    50         # ZSTD_checkCParams moves the goal posts on us from what's advertised

    51         # in the constants. So move along with them.

    52         if searchlength == zstd.SEARCHLENGTH_MIN and strategy in (zstd.STRATEGY_FAST, zstd.STRATEGY_GREEDY):

    53             searchlength += 1

    54         elif searchlength == zstd.SEARCHLENGTH_MAX and strategy != zstd.STRATEGY_FAST:

    55             searchlength -= 1

    56 

    57         p = zstd.CompressionParameters(windowlog, chainlog, hashlog,

    58                                         searchlog, searchlength,

    59                                         targetlength, strategy)

    60 

    61         cctx = zstd.ZstdCompressor(compression_params=p)

    62         with cctx.write_to(io.BytesIO()):

    63             pass

    64 

    65     @hypothesis.given(s_windowlog, s_chainlog, s_hashlog, s_searchlog,

    66                         s_searchlength, s_targetlength, s_strategy)

    67     def test_estimate_compression_context_size(self, windowlog, chainlog,

    68                                                 hashlog, searchlog,

    69                                                 searchlength, targetlength,

    70                                                 strategy):

    71         if searchlength == zstd.SEARCHLENGTH_MIN and strategy in (zstd.STRATEGY_FAST, zstd.STRATEGY_GREEDY):

    72             searchlength += 1

    73         elif searchlength == zstd.SEARCHLENGTH_MAX and strategy != zstd.STRATEGY_FAST:

    74             searchlength -= 1

    75 

    76         p = zstd.CompressionParameters(windowlog, chainlog, hashlog,

    77                             searchlog, searchlength,

    78                             targetlength, strategy)

    79         size = zstd.estimate_compression_context_size(p)