files.new_contents: import lz4.stream
  import sys
  import pytest
  import gc
  import os
  if sys.version_info <= (3, 2):
      import struct
  
  
  def run_gc(func):
      if os.environ.get('TRAVIS') is not None or os.environ.get('APPVEYOR') is not None:
          def wrapper(*args, **kwargs):
              return func(*args, **kwargs)
      else:
          def wrapper(*args, **kwargs):
              gc.collect()
              try:
                  result = func(*args, **kwargs)
              finally:
                  gc.collect()
              return result
  
      wrapper.__name__ = func.__name__
      return wrapper
  
  
  def get_stored_size(buff, block_length_size):
      if sys.version_info > (2, 7):
          if isinstance(buff, memoryview):
              b = buff.tobytes()
          else:
              b = bytes(buff)
      else:
          b = bytes(buff)
  
      if len(b) < block_length_size:
          return None
  
      if sys.version_info > (3, 2):
          return int.from_bytes(b[:block_length_size], 'little')
      else:
          # This would not work on a memoryview object, hence buff.tobytes call
          # above
          fmt = {1: 'B', 2: 'H', 4: 'I', }
          return struct.unpack('<' + fmt[block_length_size], b[:block_length_size])[0]
  
  
  @run_gc
  def roundtrip(x, c_kwargs, d_kwargs, dictionary):
      if dictionary:
          if isinstance(dictionary, tuple):
              dict_ = x[dictionary[0]:dictionary[1]]
          else:
              dict_ = dictionary
          c_kwargs['dictionary'] = dict_
          d_kwargs['dictionary'] = dict_
  
      c = bytes()
      with lz4.stream.LZ4StreamCompressor(**c_kwargs) as proc:
          for start in range(0, len(x), c_kwargs['buffer_size']):
              chunk = x[start:start + c_kwargs['buffer_size']]
              assert len(chunk) <= c_kwargs['buffer_size']
              block = proc.compress(chunk)
              if c_kwargs.get('return_bytearray'):
                  assert isinstance(block, bytearray)
              if start == 0:
                  c = block
              else:
                  c += block
              assert get_stored_size(block, c_kwargs['store_comp_size']) == \
                  (len(block) - c_kwargs['store_comp_size'])
  
      d = bytes()
      with lz4.stream.LZ4StreamDecompressor(**d_kwargs) as proc:
          start = 0
          while start < len(c):
              block = proc.get_block(c[start:])
              chunk = proc.decompress(block)
              if d_kwargs.get('return_bytearray'):
                  assert isinstance(chunk, bytearray)
              if start == 0:
                  d = chunk
              else:
                  d += chunk
              start += d_kwargs['store_comp_size'] + len(block)
  
      return d
  
  
  def setup_kwargs(strategy, mode, buffer_size, store_comp_size,
                   c_return_bytearray=None, d_return_bytearray=None):
      c_kwargs = {}
  
      if mode[0] is not None:
          c_kwargs['mode'] = mode[0]
      if mode[1] is not None:
          c_kwargs.update(mode[1])
  
      c_kwargs['strategy'] = strategy
      c_kwargs['buffer_size'] = buffer_size
      c_kwargs.update(store_comp_size)
  
      if(c_return_bytearray):
          c_kwargs.update(c_return_bytearray)
  
      d_kwargs = {}
  
      if(d_return_bytearray):
          d_kwargs.update(d_return_bytearray)
  
      d_kwargs['strategy'] = strategy
      d_kwargs['buffer_size'] = buffer_size
      d_kwargs.update(store_comp_size)
  
      return (c_kwargs, d_kwargs)
  
  
  # Test single threaded usage with all valid variations of input
  def test_1(data, strategy, mode, buffer_size, store_comp_size,
             c_return_bytearray, d_return_bytearray, dictionary):
      if buffer_size >= (1 << (8 * store_comp_size['store_comp_size'])):
          pytest.skip("Invalid case: buffer_size too large for the block length area")
  
      (c_kwargs, d_kwargs) = setup_kwargs(
          strategy, mode, buffer_size, store_comp_size, c_return_bytearray, d_return_bytearray)
  
      d = roundtrip(data, c_kwargs, d_kwargs, dictionary)
  
      assert d == data
  
  
  # Test multi threaded:
  #   Not relevant in the lz4.stream case (the process is highly sequential,
  #   and re-use/share the same context from one input chunk to the next one).
  def test_2(data, strategy, mode, buffer_size, store_comp_size, dictionary): # noqa
      pass
  
files.new_path: tests/stream/test_stream_0.py
files.old_contents: import lz4.stream
  import sys
  import pytest
  if sys.version_info <= (3, 2):
      import struct
  
  
  def get_stored_size(buff, block_length_size):
      if sys.version_info > (2, 7):
          if isinstance(buff, memoryview):
              b = buff.tobytes()
          else:
              b = bytes(buff)
      else:
          b = bytes(buff)
  
      if len(b) < block_length_size:
          return None
  
      if sys.version_info > (3, 2):
          return int.from_bytes(b[:block_length_size], 'little')
      else:
          # This would not work on a memoryview object, hence buff.tobytes call
          # above
          fmt = {1: 'B', 2: 'H', 4: 'I', }
          return struct.unpack('<' + fmt[block_length_size], b[:block_length_size])[0]
  
  
  def roundtrip(x, c_kwargs, d_kwargs, dictionary):
      if dictionary:
          if isinstance(dictionary, tuple):
              dict_ = x[dictionary[0]:dictionary[1]]
          else:
              dict_ = dictionary
          c_kwargs['dictionary'] = dict_
          d_kwargs['dictionary'] = dict_
  
      c = bytes()
      with lz4.stream.LZ4StreamCompressor(**c_kwargs) as proc:
          for start in range(0, len(x), c_kwargs['buffer_size']):
              chunk = x[start:start + c_kwargs['buffer_size']]
              assert len(chunk) <= c_kwargs['buffer_size']
              block = proc.compress(chunk)
              if c_kwargs.get('return_bytearray'):
                  assert isinstance(block, bytearray)
              if start == 0:
                  c = block
              else:
                  c += block
              assert get_stored_size(block, c_kwargs['store_comp_size']) == \
                  (len(block) - c_kwargs['store_comp_size'])
  
      d = bytes()
      with lz4.stream.LZ4StreamDecompressor(**d_kwargs) as proc:
          start = 0
          while start < len(c):
              block = proc.get_block(c[start:])
              chunk = proc.decompress(block)
              if d_kwargs.get('return_bytearray'):
                  assert isinstance(chunk, bytearray)
              if start == 0:
                  d = chunk
              else:
                  d += chunk
              start += d_kwargs['store_comp_size'] + len(block)
  
      return d
  
  
  def setup_kwargs(strategy, mode, buffer_size, store_comp_size,
                   c_return_bytearray=None, d_return_bytearray=None):
      c_kwargs = {}
  
      if mode[0] is not None:
          c_kwargs['mode'] = mode[0]
      if mode[1] is not None:
          c_kwargs.update(mode[1])
  
      c_kwargs['strategy'] = strategy
      c_kwargs['buffer_size'] = buffer_size
      c_kwargs.update(store_comp_size)
  
      if(c_return_bytearray):
          c_kwargs.update(c_return_bytearray)
  
      d_kwargs = {}
  
      if(d_return_bytearray):
          d_kwargs.update(d_return_bytearray)
  
      d_kwargs['strategy'] = strategy
      d_kwargs['buffer_size'] = buffer_size
      d_kwargs.update(store_comp_size)
  
      return (c_kwargs, d_kwargs)
  
  
  # Test single threaded usage with all valid variations of input
  def test_1(data, strategy, mode, buffer_size, store_comp_size,
             c_return_bytearray, d_return_bytearray, dictionary):
      if buffer_size >= (1 << (8 * store_comp_size['store_comp_size'])):
          pytest.skip("Invalid case: buffer_size too large for the block length area")
  
      (c_kwargs, d_kwargs) = setup_kwargs(
          strategy, mode, buffer_size, store_comp_size, c_return_bytearray, d_return_bytearray)
  
      d = roundtrip(data, c_kwargs, d_kwargs, dictionary)
  
      assert d == data
  
  
  # Test multi threaded:
  #   Not relevant in the lz4.stream case (the process is highly sequential,
  #   and re-use/share the same context from one input chunk to the next one).
  def test_2(data, strategy, mode, buffer_size, store_comp_size, dictionary): # noqa
      pass
  
files.old_path: tests/stream/test_stream_0.py
files.new_contents: import lz4.stream
  import pytest
  import sys
  import os
  import psutil
  import gc
  
  
  if sys.version_info < (3, ):
      from struct import pack, unpack
  
      def _get_format(length, byteorder, signed):
          _order = {'l': '<', 'b': '>'}
          _fmt = {1: 'b', 2: 'h', 4: 'i', 8: 'q'}
          _sign = {True: lambda x: x.lower(), False: lambda x: x.upper()}
          return _sign[signed](_order[byteorder[0].lower()] + _fmt[length])
  
      def int_to_bytes(value, length=4, byteorder='little', signed=False):
          return bytearray(pack(_get_format(length, byteorder, signed), value))
  
      def int_from_bytes(bytes, byteorder='little', signed=False):
          return unpack(_get_format(len(bytes), byteorder, signed), bytes)[0]
  
  else:
      def int_to_bytes(value, length=4, byteorder='little', signed=False):
          return value.to_bytes(length, byteorder, signed=signed)
  
      def int_from_bytes(bytes, byteorder='little', signed=False):
          return int.from_bytes(bytes, byteorder, signed=signed)
  
  
  # This test requires allocating a big lump of memory. In order to
  # avoid a massive memory allocation during byte compilation, we have
  # to declare a variable for the size of the buffer we're going to
  # create outside the scope of the function below. See:
  # https://bugs.python.org/issue21074
  _4GB = 0x100000000  # 4GB
  
  # This test will be killed on Travis due to the 3GB memory limit
  # there. Unfortunately psutil reports the host memory, not the memory
  # available to the container, and so can't be used to detect available
  # memory, so instead, as an ugly hack for detecting we're on Travis we
  # check for the TRAVIS environment variable being set. This is quite
  # fragile.
  
  
  def run_gc(func):
      if os.environ.get('TRAVIS') is not None or os.environ.get('APPVEYOR') is not None:
          def wrapper(*args, **kwargs):
              return func(*args, **kwargs)
      else:
          def wrapper(*args, **kwargs):
              gc.collect()
              try:
                  result = func(*args, **kwargs)
              finally:
                  gc.collect()
              return result
  
      wrapper.__name__ = func.__name__
      return wrapper
  
  
  def run_gc_param_store_comp_size(func):
      if os.environ.get('TRAVIS') is not None:
          def wrapper(store_comp_size, *args, **kwargs):
              return func(store_comp_size, *args, **kwargs)
      else:
          def wrapper(store_comp_size, *args, **kwargs):
              gc.collect()
              try:
                  result = func(store_comp_size, *args, **kwargs)
              finally:
                  gc.collect()
              return result
  
      wrapper.__name__ = func.__name__
      return wrapper
  
  
  def compress(x, c_kwargs, return_block_offset=False, check_block_type=False):
      o = [0, ]
      if c_kwargs.get('return_bytearray', False):
          c = bytearray()
      else:
          c = bytes()
      with lz4.stream.LZ4StreamCompressor(**c_kwargs) as proc:
          for start in range(0, len(x), c_kwargs['buffer_size']):
              chunk = x[start:start + c_kwargs['buffer_size']]
              block = proc.compress(chunk)
              c += block
              if return_block_offset:
                  o.append(len(c))
              if check_block_type:
                  assert isinstance(block, c.__class__)
      if return_block_offset:
          return c, o
      else:
          return c
  
  
  def decompress(x, d_kwargs, check_chunk_type=False):
      if d_kwargs.get('return_bytearray', False):
          d = bytearray()
      else:
          d = bytes()
      with lz4.stream.LZ4StreamDecompressor(**d_kwargs) as proc:
          start = 0
          while start < len(x):
              block = proc.get_block(x[start:])
              chunk = proc.decompress(block)
              d += chunk
              start += d_kwargs['store_comp_size'] + len(block)
              if check_chunk_type:
                  assert isinstance(chunk, d.__class__)
      return d
  
  
  @run_gc
  def test_invalid_config_c_1():
      c_kwargs = {}
      c_kwargs['strategy'] = "ring_buffer"
      c_kwargs['buffer_size'] = 1024
  
      with pytest.raises(NotImplementedError):
          lz4.stream.LZ4StreamCompressor(**c_kwargs)
  
  
  @run_gc
  def test_invalid_config_d_1():
      d_kwargs = {}
      d_kwargs['strategy'] = "ring_buffer"
      d_kwargs['buffer_size'] = 1024
  
      with pytest.raises(NotImplementedError):
          lz4.stream.LZ4StreamDecompressor(**d_kwargs)
  
  
  @run_gc
  def test_invalid_config_c_2():
      c_kwargs = {}
      c_kwargs['strategy'] = "foo"
      c_kwargs['buffer_size'] = 1024
  
      with pytest.raises(ValueError):
          lz4.stream.LZ4StreamCompressor(**c_kwargs)
  
  
  @run_gc
  def test_invalid_config_d_2():
      d_kwargs = {}
      d_kwargs['strategy'] = "foo"
      d_kwargs['buffer_size'] = 1024
  
      with pytest.raises(ValueError):
          lz4.stream.LZ4StreamDecompressor(**d_kwargs)
  
  
  @run_gc_param_store_comp_size
  def test_invalid_config_c_3(store_comp_size):
      c_kwargs = {}
      c_kwargs['strategy'] = "double_buffer"
      c_kwargs['buffer_size'] = 1024
      c_kwargs['store_comp_size'] = store_comp_size['store_comp_size'] + 5
  
      with pytest.raises(ValueError):
          lz4.stream.LZ4StreamCompressor(**c_kwargs)
  
  
  @run_gc_param_store_comp_size
  def test_invalid_config_d_3(store_comp_size):
      d_kwargs = {}
      d_kwargs['strategy'] = "double_buffer"
      d_kwargs['buffer_size'] = 1024
      d_kwargs['store_comp_size'] = store_comp_size['store_comp_size'] + 5
  
      with pytest.raises(ValueError):
          lz4.stream.LZ4StreamDecompressor(**d_kwargs)
  
  
  @run_gc_param_store_comp_size
  def test_invalid_config_c_4(store_comp_size):
      c_kwargs = {}
      c_kwargs['strategy'] = "double_buffer"
      c_kwargs['buffer_size'] = 1 << (8 * store_comp_size['store_comp_size'])
      c_kwargs.update(store_comp_size)
  
      if store_comp_size['store_comp_size'] >= 4:
          # No need for skiping this test case, since arguments check is
          # expecting to raise an error.
  
          # Make sure the page size is larger than what the input bound will be,
          # but still fit in 4 bytes
          c_kwargs['buffer_size'] -= 1
  
      if c_kwargs['buffer_size'] > lz4.stream.LZ4_MAX_INPUT_SIZE:
          message = r"^Invalid buffer_size argument: \d+. Cannot define output buffer size. Must be lesser or equal to 2113929216$"  # noqa
          err_class = ValueError
      else:
          message = r"^Inconsistent buffer_size/store_comp_size values. Maximal compressed length \(\d+\) cannot fit in a \d+ byte-long integer$"  # noqa
          err_class = lz4.stream.LZ4StreamError
  
      with pytest.raises(err_class, match=message):
          lz4.stream.LZ4StreamCompressor(**c_kwargs)
  
  
  @run_gc_param_store_comp_size
  def test_invalid_config_d_4(store_comp_size):
      d_kwargs = {}
      d_kwargs['strategy'] = "double_buffer"
      d_kwargs['buffer_size'] = 1 << (8 * store_comp_size['store_comp_size'])
      d_kwargs.update(store_comp_size)
  
      if store_comp_size['store_comp_size'] >= 4:
  
          if os.environ.get('TRAVIS') is not None:
              pytest.skip('Skipping test on Travis due to insufficient memory')
  
          if sys.maxsize < 0xffffffff:
              pytest.skip('Py_ssize_t too small for this test')
  
          if psutil.virtual_memory().available < 4 * d_kwargs['buffer_size']:
              # The internal LZ4 context will request at least 3 times buffer_size
              # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
              # for the output buffer), so round up to 4 buffer_size.
              pytest.skip('Insufficient system memory for this test')
  
          # Make sure the page size is larger than what the input bound will be,
          # but still fit in 4 bytes
          d_kwargs['buffer_size'] -= 1
  
      # No failure expected during instanciation/initialization
      lz4.stream.LZ4StreamDecompressor(**d_kwargs)
  
  
  @run_gc
  def test_invalid_config_c_5():
      c_kwargs = {}
      c_kwargs['strategy'] = "double_buffer"
      c_kwargs['buffer_size'] = lz4.stream.LZ4_MAX_INPUT_SIZE
  
      if os.environ.get('TRAVIS') is not None:
          pytest.skip('Skipping test on Travis due to insufficient memory')
  
      if sys.maxsize < 0xffffffff:
          pytest.skip('Py_ssize_t too small for this test')
  
      if psutil.virtual_memory().available < 3 * c_kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer)
          pytest.skip('Insufficient system memory for this test')
  
      # No failure expected
      lz4.stream.LZ4StreamCompressor(**c_kwargs)
  
      c_kwargs['buffer_size'] = lz4.stream.LZ4_MAX_INPUT_SIZE + 1
      with pytest.raises(ValueError):
          lz4.stream.LZ4StreamCompressor(**c_kwargs)
  
      # Make sure the page size is larger than what the input bound will be,
      # but still fit in 4 bytes
      c_kwargs['buffer_size'] = _4GB - 1  # 4GB - 1 (to fit in 4 bytes)
      with pytest.raises(ValueError):
          lz4.stream.LZ4StreamCompressor(**c_kwargs)
  
  
  @run_gc
  def test_invalid_config_d_5():
      d_kwargs = {}
      d_kwargs['strategy'] = "double_buffer"
  
      # No failure expected during instanciation/initialization
      d_kwargs['buffer_size'] = lz4.stream.LZ4_MAX_INPUT_SIZE
  
      if os.environ.get('TRAVIS') is not None:
          pytest.skip('Skipping test on Travis due to insufficient memory')
  
      if sys.maxsize < 0xffffffff:
          pytest.skip('Py_ssize_t too small for this test')
  
      if psutil.virtual_memory().available < 3 * d_kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer)
          pytest.skip('Insufficient system memory for this test')
  
      lz4.stream.LZ4StreamDecompressor(**d_kwargs)
  
      # No failure expected during instanciation/initialization
      d_kwargs['buffer_size'] = lz4.stream.LZ4_MAX_INPUT_SIZE + 1
  
      if os.environ.get('TRAVIS') is not None:
          pytest.skip('Skipping test on Travis due to insufficient memory')
  
      if sys.maxsize < 0xffffffff:
          pytest.skip('Py_ssize_t too small for this test')
  
      if psutil.virtual_memory().available < 3 * d_kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer)
          pytest.skip('Insufficient system memory for this test')
  
      lz4.stream.LZ4StreamDecompressor(**d_kwargs)
  
      # No failure expected during instanciation/initialization
      d_kwargs['buffer_size'] = _4GB - 1  # 4GB - 1 (to fit in 4 bytes)
  
      if os.environ.get('TRAVIS') is not None:
          pytest.skip('Skipping test on Travis due to insufficient memory')
  
      if sys.maxsize < 0xffffffff:
          pytest.skip('Py_ssize_t too small for this test')
  
      if psutil.virtual_memory().available < 4 * d_kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer), so round up to 4 buffer_size.
          pytest.skip('Insufficient system memory for this test')
  
      lz4.stream.LZ4StreamDecompressor(**d_kwargs)
  
  
  @run_gc
  def test_decompress_corrupted_input_1():
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      data = compress(b'A' * 512, c_kwargs)
      decompress(data, d_kwargs)
  
      message = r"^Requested input size \(\d+\) larger than source size \(\d+\)$"
  
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(data[4:], d_kwargs)
  
  
  @run_gc
  def test_decompress_corrupted_input_2():
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      data = compress(b'A' * 512, c_kwargs)
      decompress(data, d_kwargs)
  
      message = r"^Decompression failed. error: \d+$"
  
      # Block size corruption in the first block
  
      # Block size longer than actual:
      data = int_to_bytes(int_from_bytes(data[:4], 'little') + 1, 4, 'little') + data[4:]
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(data, d_kwargs)
  
      # Block size shorter than actual:
      data = int_to_bytes(int_from_bytes(data[:4], 'little') - 2, 4, 'little') + data[4:]
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(data, d_kwargs)
  
  
  @run_gc
  def test_decompress_corrupted_input_3():
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      data = compress(b'A' * 512, c_kwargs)
      decompress(data, d_kwargs)
  
      message = r"^Decompression failed. error: \d+$"
  
      # Block size corruption in a block in the middle of the stream
      offset = 4 + int_from_bytes(data[:4], 'little')
  
      # Block size longer than actual:
      block_len = int_from_bytes(data[offset:offset + 4], 'little') + 1
      data = data[:offset] + int_to_bytes(block_len, 4, 'little') + data[offset + 4:]
  
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(data, d_kwargs)
  
      # Block size shorter than actual:
      block_len = int_from_bytes(data[offset:offset + 4], 'little') - 2
      data = data[:offset] + int_to_bytes(block_len, 4, 'little') + data[offset + 4:]
  
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(data, d_kwargs)
  
  
  @run_gc
  def test_decompress_corrupted_input_4():
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      data = compress(b'A' * 256, c_kwargs)
      decompress(data, d_kwargs)
  
      # Block size corruption in the last block of the stream
      offset = 4 + int_from_bytes(data[:4], 'little')
  
      # Block size longer than actual:
      block_len = int_from_bytes(data[offset:offset + 4], 'little') + 1
      data = data[:offset] + int_to_bytes(block_len, 4, 'little') + data[offset + 4:]
  
      message = r"^Requested input size \(\d+\) larger than source size \(\d+\)$"
  
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(data, d_kwargs)
  
      # Block size shorter than actual:
      block_len = int_from_bytes(data[offset:offset + 4], 'little') - 2
      data = data[:offset] + int_to_bytes(block_len, 4, 'little') + data[offset + 4:]
  
      message = r"^Decompression failed. error: \d+$"
  
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(data, d_kwargs)
  
  
  @run_gc
  def test_decompress_truncated():
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      input_data = b"2099023098234882923049823094823094898239230982349081231290381209380981203981209381238901283098908123109238098123" * 24
      compressed, block_offsets = compress(input_data, c_kwargs, return_block_offset=True)
  
      last_block_offset = 0
      for n in range(len(compressed)):
          if n in block_offsets:
              # end of input matches end of block, so decompression must succeed
              last_block_offset = n
              decompress(compressed[:n], d_kwargs)
  
          else:
              # end of input does not match end of block, so decompression failure is expected
              if n - last_block_offset < c_kwargs['store_comp_size']:
                  message = "^Invalid source, too small for holding any block$"
              else:
                  message = r"^Requested input size \(\d+\) larger than source size \(\d+\)$"
  
              with pytest.raises(lz4.stream.LZ4StreamError, match=message):
                  decompress(compressed[:n], d_kwargs)
  
  
  # This next test is probably redundant given test_decompress_truncated above
  # since the trailing bytes will be considered as the truncated last block, but
  # we will keep them for now
  
  
  @run_gc
  def test_decompress_with_trailer():
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      data = b'A' * 64
      comp = compress(data, c_kwargs)
  
      message = "^Invalid source, too small for holding any block$"
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(comp + b'A', d_kwargs)
  
      message = r"^Requested input size \(\d+\) larger than source size \(\d+\)$"
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(comp + b'A' * 10, d_kwargs)
  
      for n in range(1, 10):
          if n < d_kwargs['store_comp_size']:
              message = "^Invalid source, too small for holding any block$"
          else:
              message = r"^Decompression failed. error: \d+$"
          with pytest.raises(lz4.stream.LZ4StreamError, match=message):
              decompress(comp + b'\x00' * n, d_kwargs)
  
  
  @run_gc
  def test_unicode():
      if sys.version_info < (3,):
          return  # skip
  
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      DATA = b'x'
      with pytest.raises(TypeError):
          compress(DATA.decode('latin1'), c_kwargs)
          decompress(compress(DATA, c_kwargs).decode('latin1'), d_kwargs)
  
  
  # These next two are probably redundant given test_1 above but we'll keep them
  # for now
  
  
  @run_gc
  def test_return_bytearray():
      if sys.version_info < (3,):
          return  # skip
  
      c_kwargs_r = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
      c_kwargs = {'return_bytearray': True}
      c_kwargs.update(c_kwargs_r)
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      data = os.urandom(128 * 1024)  # Read 128kb
      compressed = compress(data, c_kwargs_r, check_block_type=True)
      b = compress(data, c_kwargs, check_block_type=True)
      assert isinstance(b, bytearray)
      assert bytes(b) == compressed
      b = decompress(compressed, d_kwargs, check_chunk_type=True)
      assert isinstance(b, bytearray)
      assert bytes(b) == data
  
  
  @run_gc
  def test_memoryview():
      if sys.version_info < (2, 7):
          return  # skip
  
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      data = os.urandom(128 * 1024)  # Read 128kb
      compressed = compress(data, c_kwargs)
      assert compress(memoryview(data), c_kwargs) == compressed
      assert decompress(memoryview(compressed), d_kwargs) == data
  
  
  @run_gc
  def test_with_dict_none():
      kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      input_data = b"2099023098234882923049823094823094898239230982349081231290381209380981203981209381238901283098908123109238098123" * 24
      for mode in ['default', 'high_compression']:
          c_kwargs = {'mode': mode, 'dictionary': None}
          c_kwargs.update(kwargs)
          d_kwargs = {}
          d_kwargs.update(kwargs)
          assert decompress(compress(input_data, c_kwargs), d_kwargs) == input_data
  
          c_kwargs = {'mode': mode}
          c_kwargs.update(kwargs)
          d_kwargs = {'dictionary': None}
          d_kwargs.update(kwargs)
          assert decompress(compress(input_data, c_kwargs), d_kwargs) == input_data
  
          c_kwargs = {'mode': mode, 'dictionary': b''}
          c_kwargs.update(kwargs)
          d_kwargs = {}
          d_kwargs.update(kwargs)
          assert decompress(compress(input_data, c_kwargs), d_kwargs) == input_data
  
          c_kwargs = {'mode': mode}
          c_kwargs.update(kwargs)
          d_kwargs = {'dictionary': b''}
          d_kwargs.update(kwargs)
          assert decompress(compress(input_data, c_kwargs), d_kwargs) == input_data
  
          c_kwargs = {'mode': mode, 'dictionary': ''}
          c_kwargs.update(kwargs)
          d_kwargs = {}
          d_kwargs.update(kwargs)
          assert decompress(compress(input_data, c_kwargs), d_kwargs) == input_data
  
          c_kwargs = {'mode': mode}
          c_kwargs.update(kwargs)
          d_kwargs = {'dictionary': ''}
          d_kwargs.update(kwargs)
          assert decompress(compress(input_data, c_kwargs), d_kwargs) == input_data
  
  
  @run_gc
  def test_with_dict():
      kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      input_data = b"2099023098234882923049823094823094898239230982349081231290381209380981203981209381238901283098908123109238098123" * 24
      dict1 = input_data[10:30]
      dict2 = input_data[20:40]
      message = r"^Decompression failed. error: \d+$"
  
      for mode in ['default', 'high_compression']:
          c_kwargs = {'mode': mode, 'dictionary': dict1}
          c_kwargs.update(kwargs)
          compressed = compress(input_data, c_kwargs)
  
          d_kwargs = {}
          d_kwargs.update(kwargs)
          with pytest.raises(lz4.stream.LZ4StreamError, match=message):
              decompress(compressed, d_kwargs)
  
          d_kwargs = {'dictionary': dict1[:2]}
          d_kwargs.update(kwargs)
          with pytest.raises(lz4.stream.LZ4StreamError, match=message):
              decompress(compressed, d_kwargs)
  
          d_kwargs = {'dictionary': dict2}
          d_kwargs.update(kwargs)
          assert decompress(compressed, d_kwargs) != input_data
  
          d_kwargs = {'dictionary': dict1}
          d_kwargs.update(kwargs)
          assert decompress(compressed, d_kwargs) == input_data
  
      c_kwargs = {}
      c_kwargs.update(kwargs)
      d_kwargs = {'dictionary': dict1}
      d_kwargs.update(kwargs)
      assert decompress(compress(input_data, c_kwargs), d_kwargs) == input_data
  
  
  @run_gc
  def test_known_decompress_1():
      d_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      output = b''
  
      input = b'\x00\x00\x00\x00'
      message = "^Decompression failed. error: 1$"
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(input, d_kwargs)
  
      input = b'\x01\x00\x00\x00\x00'
      assert decompress(input, d_kwargs) == output
  
  
  @run_gc
  def test_known_decompress_2():
      d_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      input = b'\x02\x00\x00\x00\x10 '
      output = b' '
      assert decompress(input, d_kwargs) == output
  
  
  @run_gc
  def test_known_decompress_3():
      d_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      # uncompressed data size smaller than buffer_size
      input = b'%\x00\x00\x00\xff\x0bLorem ipsum dolor sit amet\x1a\x006P amet'
      output = b'Lorem ipsum dolor sit amet' * 4
      assert decompress(input, d_kwargs) == output
  
  
  @run_gc
  def test_known_decompress_4():
      d_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      input = b'%\x00\x00\x00\xff\x0bLorem ipsum dolor sit amet\x1a\x00NPit am\n\x00\x00\x00\x0fh\x00hP sit \x05\x00\x00\x00@amet'
      output = b'Lorem ipsum dolor sit amet' * 10
      assert decompress(input, d_kwargs) == output
  
files.new_path: tests/stream/test_stream_1.py
files.old_contents: import lz4.stream
  import pytest
  import sys
  import os
  import psutil
  
  if sys.version_info < (3, ):
      from struct import pack, unpack
  
      def _get_format(length, byteorder, signed):
          _order = {'l': '<', 'b': '>'}
          _fmt = {1: 'b', 2: 'h', 4: 'i', 8: 'q'}
          _sign = {True: lambda x: x.lower(), False: lambda x: x.upper()}
          return _sign[signed](_order[byteorder[0].lower()] + _fmt[length])
  
      def int_to_bytes(value, length=4, byteorder='little', signed=False):
          return bytearray(pack(_get_format(length, byteorder, signed), value))
  
      def int_from_bytes(bytes, byteorder='little', signed=False):
          return unpack(_get_format(len(bytes), byteorder, signed), bytes)[0]
  
  else:
      def int_to_bytes(value, length=4, byteorder='little', signed=False):
          return value.to_bytes(length, byteorder, signed=signed)
  
      def int_from_bytes(bytes, byteorder='little', signed=False):
          return int.from_bytes(bytes, byteorder, signed=signed)
  
  
  # This test requires allocating a big lump of memory. In order to
  # avoid a massive memory allocation during byte compilation, we have
  # to declare a variable for the size of the buffer we're going to
  # create outside the scope of the function below. See:
  # https://bugs.python.org/issue21074
  _4GB = 0x100000000  # 4GB
  
  # This test will be killed on Travis due to the 3GB memory limit
  # there. Unfortunately psutil reports the host memory, not the memory
  # available to the container, and so can't be used to detect available
  # memory, so instead, as an ugly hack for detecting we're on Travis we
  # check for the TRAVIS environment variable being set. This is quite
  # fragile.
  
  
  def compress(x, c_kwargs, return_block_offset=False, check_block_type=False):
      o = [0, ]
      if c_kwargs.get('return_bytearray', False):
          c = bytearray()
      else:
          c = bytes()
      with lz4.stream.LZ4StreamCompressor(**c_kwargs) as proc:
          for start in range(0, len(x), c_kwargs['buffer_size']):
              chunk = x[start:start + c_kwargs['buffer_size']]
              block = proc.compress(chunk)
              c += block
              if return_block_offset:
                  o.append(len(c))
              if check_block_type:
                  assert isinstance(block, c.__class__)
      if return_block_offset:
          return c, o
      else:
          return c
  
  
  def decompress(x, d_kwargs, check_chunk_type=False):
      if d_kwargs.get('return_bytearray', False):
          d = bytearray()
      else:
          d = bytes()
      with lz4.stream.LZ4StreamDecompressor(**d_kwargs) as proc:
          start = 0
          while start < len(x):
              block = proc.get_block(x[start:])
              chunk = proc.decompress(block)
              d += chunk
              start += d_kwargs['store_comp_size'] + len(block)
              if check_chunk_type:
                  assert isinstance(chunk, d.__class__)
      return d
  
  
  def test_invalid_config_c_1():
      c_kwargs = {}
      c_kwargs['strategy'] = "ring_buffer"
      c_kwargs['buffer_size'] = 1024
  
      with pytest.raises(NotImplementedError):
          lz4.stream.LZ4StreamCompressor(**c_kwargs)
  
  
  def test_invalid_config_d_1():
      d_kwargs = {}
      d_kwargs['strategy'] = "ring_buffer"
      d_kwargs['buffer_size'] = 1024
  
      with pytest.raises(NotImplementedError):
          lz4.stream.LZ4StreamDecompressor(**d_kwargs)
  
  
  def test_invalid_config_c_2():
      c_kwargs = {}
      c_kwargs['strategy'] = "foo"
      c_kwargs['buffer_size'] = 1024
  
      with pytest.raises(ValueError):
          lz4.stream.LZ4StreamCompressor(**c_kwargs)
  
  
  def test_invalid_config_d_2():
      d_kwargs = {}
      d_kwargs['strategy'] = "foo"
      d_kwargs['buffer_size'] = 1024
  
      with pytest.raises(ValueError):
          lz4.stream.LZ4StreamDecompressor(**d_kwargs)
  
  
  def test_invalid_config_c_3(store_comp_size):
      c_kwargs = {}
      c_kwargs['strategy'] = "double_buffer"
      c_kwargs['buffer_size'] = 1024
      c_kwargs['store_comp_size'] = store_comp_size['store_comp_size'] + 5
  
      with pytest.raises(ValueError):
          lz4.stream.LZ4StreamCompressor(**c_kwargs)
  
  
  def test_invalid_config_d_3(store_comp_size):
      d_kwargs = {}
      d_kwargs['strategy'] = "double_buffer"
      d_kwargs['buffer_size'] = 1024
      d_kwargs['store_comp_size'] = store_comp_size['store_comp_size'] + 5
  
      with pytest.raises(ValueError):
          lz4.stream.LZ4StreamDecompressor(**d_kwargs)
  
  
  def test_invalid_config_c_4(store_comp_size):
      c_kwargs = {}
      c_kwargs['strategy'] = "double_buffer"
      c_kwargs['buffer_size'] = 1 << (8 * store_comp_size['store_comp_size'])
      c_kwargs.update(store_comp_size)
  
      if store_comp_size['store_comp_size'] >= 4:
          # No need for skiping this test case, since arguments check is
          # expecting to raise an error.
  
          # Make sure the page size is larger than what the input bound will be,
          # but still fit in 4 bytes
          c_kwargs['buffer_size'] -= 1
  
      if c_kwargs['buffer_size'] > lz4.stream.LZ4_MAX_INPUT_SIZE:
          message = r"^Invalid buffer_size argument: \d+. Cannot define output buffer size. Must be lesser or equal to 2113929216$"  # noqa
          err_class = ValueError
      else:
          message = r"^Inconsistent buffer_size/store_comp_size values. Maximal compressed length \(\d+\) cannot fit in a \d+ byte-long integer$"  # noqa
          err_class = lz4.stream.LZ4StreamError
  
      with pytest.raises(err_class, match=message):
          lz4.stream.LZ4StreamCompressor(**c_kwargs)
  
  
  def test_invalid_config_d_4(store_comp_size):
      d_kwargs = {}
      d_kwargs['strategy'] = "double_buffer"
      d_kwargs['buffer_size'] = 1 << (8 * store_comp_size['store_comp_size'])
      d_kwargs.update(store_comp_size)
  
      if store_comp_size['store_comp_size'] >= 4:
  
          if os.environ.get('TRAVIS') is not None:
              pytest.skip('Skipping test on Travis due to insufficient memory')
  
          if sys.maxsize < 0xffffffff:
              pytest.skip('Py_ssize_t too small for this test')
  
          if psutil.virtual_memory().available < 4 * d_kwargs['buffer_size']:
              # The internal LZ4 context will request at least 3 times buffer_size
              # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
              # for the output buffer), so round up to 4 buffer_size.
              pytest.skip('Insufficient system memory for this test')
  
          # Make sure the page size is larger than what the input bound will be,
          # but still fit in 4 bytes
          d_kwargs['buffer_size'] -= 1
  
      # No failure expected during instanciation/initialization
      lz4.stream.LZ4StreamDecompressor(**d_kwargs)
  
  
  def test_invalid_config_c_5():
      c_kwargs = {}
      c_kwargs['strategy'] = "double_buffer"
      c_kwargs['buffer_size'] = lz4.stream.LZ4_MAX_INPUT_SIZE
  
      if os.environ.get('TRAVIS') is not None:
          pytest.skip('Skipping test on Travis due to insufficient memory')
  
      if sys.maxsize < 0xffffffff:
          pytest.skip('Py_ssize_t too small for this test')
  
      if psutil.virtual_memory().available < 3 * c_kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer)
          pytest.skip('Insufficient system memory for this test')
  
      # No failure expected
      lz4.stream.LZ4StreamCompressor(**c_kwargs)
  
      c_kwargs['buffer_size'] = lz4.stream.LZ4_MAX_INPUT_SIZE + 1
      with pytest.raises(ValueError):
          lz4.stream.LZ4StreamCompressor(**c_kwargs)
  
      # Make sure the page size is larger than what the input bound will be,
      # but still fit in 4 bytes
      c_kwargs['buffer_size'] = _4GB - 1  # 4GB - 1 (to fit in 4 bytes)
      with pytest.raises(ValueError):
          lz4.stream.LZ4StreamCompressor(**c_kwargs)
  
  
  def test_invalid_config_d_5():
      d_kwargs = {}
      d_kwargs['strategy'] = "double_buffer"
  
      # No failure expected during instanciation/initialization
      d_kwargs['buffer_size'] = lz4.stream.LZ4_MAX_INPUT_SIZE
  
      if os.environ.get('TRAVIS') is not None:
          pytest.skip('Skipping test on Travis due to insufficient memory')
  
      if sys.maxsize < 0xffffffff:
          pytest.skip('Py_ssize_t too small for this test')
  
      if psutil.virtual_memory().available < 3 * d_kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer)
          pytest.skip('Insufficient system memory for this test')
  
      lz4.stream.LZ4StreamDecompressor(**d_kwargs)
  
      # No failure expected during instanciation/initialization
      d_kwargs['buffer_size'] = lz4.stream.LZ4_MAX_INPUT_SIZE + 1
  
      if os.environ.get('TRAVIS') is not None:
          pytest.skip('Skipping test on Travis due to insufficient memory')
  
      if sys.maxsize < 0xffffffff:
          pytest.skip('Py_ssize_t too small for this test')
  
      if psutil.virtual_memory().available < 3 * d_kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer)
          pytest.skip('Insufficient system memory for this test')
  
      lz4.stream.LZ4StreamDecompressor(**d_kwargs)
  
      # No failure expected during instanciation/initialization
      d_kwargs['buffer_size'] = _4GB - 1  # 4GB - 1 (to fit in 4 bytes)
  
      if os.environ.get('TRAVIS') is not None:
          pytest.skip('Skipping test on Travis due to insufficient memory')
  
      if sys.maxsize < 0xffffffff:
          pytest.skip('Py_ssize_t too small for this test')
  
      if psutil.virtual_memory().available < 4 * d_kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer), so round up to 4 buffer_size.
          pytest.skip('Insufficient system memory for this test')
  
      lz4.stream.LZ4StreamDecompressor(**d_kwargs)
  
  
  def test_decompress_corrupted_input_1():
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      data = compress(b'A' * 512, c_kwargs)
      decompress(data, d_kwargs)
  
      message = r"^Requested input size \(\d+\) larger than source size \(\d+\)$"
  
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(data[4:], d_kwargs)
  
  
  def test_decompress_corrupted_input_2():
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      data = compress(b'A' * 512, c_kwargs)
      decompress(data, d_kwargs)
  
      message = r"^Decompression failed. error: \d+$"
  
      # Block size corruption in the first block
  
      # Block size longer than actual:
      data = int_to_bytes(int_from_bytes(data[:4], 'little') + 1, 4, 'little') + data[4:]
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(data, d_kwargs)
  
      # Block size shorter than actual:
      data = int_to_bytes(int_from_bytes(data[:4], 'little') - 2, 4, 'little') + data[4:]
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(data, d_kwargs)
  
  
  def test_decompress_corrupted_input_3():
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      data = compress(b'A' * 512, c_kwargs)
      decompress(data, d_kwargs)
  
      message = r"^Decompression failed. error: \d+$"
  
      # Block size corruption in a block in the middle of the stream
      offset = 4 + int_from_bytes(data[:4], 'little')
  
      # Block size longer than actual:
      block_len = int_from_bytes(data[offset:offset + 4], 'little') + 1
      data = data[:offset] + int_to_bytes(block_len, 4, 'little') + data[offset + 4:]
  
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(data, d_kwargs)
  
      # Block size shorter than actual:
      block_len = int_from_bytes(data[offset:offset + 4], 'little') - 2
      data = data[:offset] + int_to_bytes(block_len, 4, 'little') + data[offset + 4:]
  
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(data, d_kwargs)
  
  
  def test_decompress_corrupted_input_4():
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      data = compress(b'A' * 256, c_kwargs)
      decompress(data, d_kwargs)
  
      # Block size corruption in the last block of the stream
      offset = 4 + int_from_bytes(data[:4], 'little')
  
      # Block size longer than actual:
      block_len = int_from_bytes(data[offset:offset + 4], 'little') + 1
      data = data[:offset] + int_to_bytes(block_len, 4, 'little') + data[offset + 4:]
  
      message = r"^Requested input size \(\d+\) larger than source size \(\d+\)$"
  
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(data, d_kwargs)
  
      # Block size shorter than actual:
      block_len = int_from_bytes(data[offset:offset + 4], 'little') - 2
      data = data[:offset] + int_to_bytes(block_len, 4, 'little') + data[offset + 4:]
  
      message = r"^Decompression failed. error: \d+$"
  
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(data, d_kwargs)
  
  
  def test_decompress_truncated():
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      input_data = b"2099023098234882923049823094823094898239230982349081231290381209380981203981209381238901283098908123109238098123" * 24
      compressed, block_offsets = compress(input_data, c_kwargs, return_block_offset=True)
  
      last_block_offset = 0
      for n in range(len(compressed)):
          if n in block_offsets:
              # end of input matches end of block, so decompression must succeed
              last_block_offset = n
              decompress(compressed[:n], d_kwargs)
  
          else:
              # end of input does not match end of block, so decompression failure is expected
              if n - last_block_offset < c_kwargs['store_comp_size']:
                  message = "^Invalid source, too small for holding any block$"
              else:
                  message = r"^Requested input size \(\d+\) larger than source size \(\d+\)$"
  
              with pytest.raises(lz4.stream.LZ4StreamError, match=message):
                  decompress(compressed[:n], d_kwargs)
  
  
  # This next test is probably redundant given test_decompress_truncated above
  # since the trailing bytes will be considered as the truncated last block, but
  # we will keep them for now
  
  
  def test_decompress_with_trailer():
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      data = b'A' * 64
      comp = compress(data, c_kwargs)
  
      message = "^Invalid source, too small for holding any block$"
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(comp + b'A', d_kwargs)
  
      message = r"^Requested input size \(\d+\) larger than source size \(\d+\)$"
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(comp + b'A' * 10, d_kwargs)
  
      for n in range(1, 10):
          if n < d_kwargs['store_comp_size']:
              message = "^Invalid source, too small for holding any block$"
          else:
              message = r"^Decompression failed. error: \d+$"
          with pytest.raises(lz4.stream.LZ4StreamError, match=message):
              decompress(comp + b'\x00' * n, d_kwargs)
  
  
  def test_unicode():
      if sys.version_info < (3,):
          return  # skip
  
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      DATA = b'x'
      with pytest.raises(TypeError):
          compress(DATA.decode('latin1'), c_kwargs)
          decompress(compress(DATA, c_kwargs).decode('latin1'), d_kwargs)
  
  
  # These next two are probably redundant given test_1 above but we'll keep them
  # for now
  
  
  def test_return_bytearray():
      if sys.version_info < (3,):
          return  # skip
  
      c_kwargs_r = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
      c_kwargs = {'return_bytearray': True}
      c_kwargs.update(c_kwargs_r)
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      data = os.urandom(128 * 1024)  # Read 128kb
      compressed = compress(data, c_kwargs_r, check_block_type=True)
      b = compress(data, c_kwargs, check_block_type=True)
      assert isinstance(b, bytearray)
      assert bytes(b) == compressed
      b = decompress(compressed, d_kwargs, check_chunk_type=True)
      assert isinstance(b, bytearray)
      assert bytes(b) == data
  
  
  def test_memoryview():
      if sys.version_info < (2, 7):
          return  # skip
  
      c_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      d_kwargs = {}
      d_kwargs.update(c_kwargs)
  
      data = os.urandom(128 * 1024)  # Read 128kb
      compressed = compress(data, c_kwargs)
      assert compress(memoryview(data), c_kwargs) == compressed
      assert decompress(memoryview(compressed), d_kwargs) == data
  
  
  def test_with_dict_none():
      kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      input_data = b"2099023098234882923049823094823094898239230982349081231290381209380981203981209381238901283098908123109238098123" * 24
      for mode in ['default', 'high_compression']:
          c_kwargs = {'mode': mode, 'dictionary': None}
          c_kwargs.update(kwargs)
          d_kwargs = {}
          d_kwargs.update(kwargs)
          assert decompress(compress(input_data, c_kwargs), d_kwargs) == input_data
  
          c_kwargs = {'mode': mode}
          c_kwargs.update(kwargs)
          d_kwargs = {'dictionary': None}
          d_kwargs.update(kwargs)
          assert decompress(compress(input_data, c_kwargs), d_kwargs) == input_data
  
          c_kwargs = {'mode': mode, 'dictionary': b''}
          c_kwargs.update(kwargs)
          d_kwargs = {}
          d_kwargs.update(kwargs)
          assert decompress(compress(input_data, c_kwargs), d_kwargs) == input_data
  
          c_kwargs = {'mode': mode}
          c_kwargs.update(kwargs)
          d_kwargs = {'dictionary': b''}
          d_kwargs.update(kwargs)
          assert decompress(compress(input_data, c_kwargs), d_kwargs) == input_data
  
          c_kwargs = {'mode': mode, 'dictionary': ''}
          c_kwargs.update(kwargs)
          d_kwargs = {}
          d_kwargs.update(kwargs)
          assert decompress(compress(input_data, c_kwargs), d_kwargs) == input_data
  
          c_kwargs = {'mode': mode}
          c_kwargs.update(kwargs)
          d_kwargs = {'dictionary': ''}
          d_kwargs.update(kwargs)
          assert decompress(compress(input_data, c_kwargs), d_kwargs) == input_data
  
  
  def test_with_dict():
      kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      input_data = b"2099023098234882923049823094823094898239230982349081231290381209380981203981209381238901283098908123109238098123" * 24
      dict1 = input_data[10:30]
      dict2 = input_data[20:40]
      message = r"^Decompression failed. error: \d+$"
  
      for mode in ['default', 'high_compression']:
          c_kwargs = {'mode': mode, 'dictionary': dict1}
          c_kwargs.update(kwargs)
          compressed = compress(input_data, c_kwargs)
  
          d_kwargs = {}
          d_kwargs.update(kwargs)
          with pytest.raises(lz4.stream.LZ4StreamError, match=message):
              decompress(compressed, d_kwargs)
  
          d_kwargs = {'dictionary': dict1[:2]}
          d_kwargs.update(kwargs)
          with pytest.raises(lz4.stream.LZ4StreamError, match=message):
              decompress(compressed, d_kwargs)
  
          d_kwargs = {'dictionary': dict2}
          d_kwargs.update(kwargs)
          assert decompress(compressed, d_kwargs) != input_data
  
          d_kwargs = {'dictionary': dict1}
          d_kwargs.update(kwargs)
          assert decompress(compressed, d_kwargs) == input_data
  
      c_kwargs = {}
      c_kwargs.update(kwargs)
      d_kwargs = {'dictionary': dict1}
      d_kwargs.update(kwargs)
      assert decompress(compress(input_data, c_kwargs), d_kwargs) == input_data
  
  
  def test_known_decompress_1():
      d_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      output = b''
  
      input = b'\x00\x00\x00\x00'
      message = "^Decompression failed. error: 1$"
      with pytest.raises(lz4.stream.LZ4StreamError, match=message):
          decompress(input, d_kwargs)
  
      input = b'\x01\x00\x00\x00\x00'
      assert decompress(input, d_kwargs) == output
  
  
  def test_known_decompress_2():
      d_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      input = b'\x02\x00\x00\x00\x10 '
      output = b' '
      assert decompress(input, d_kwargs) == output
  
  
  def test_known_decompress_3():
      d_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      # uncompressed data size smaller than buffer_size
      input = b'%\x00\x00\x00\xff\x0bLorem ipsum dolor sit amet\x1a\x006P amet'
      output = b'Lorem ipsum dolor sit amet' * 4
      assert decompress(input, d_kwargs) == output
  
  
  def test_known_decompress_4():
      d_kwargs = {'strategy': "double_buffer", 'buffer_size': 128, 'store_comp_size': 4}
  
      input = b'%\x00\x00\x00\xff\x0bLorem ipsum dolor sit amet\x1a\x00NPit am\n\x00\x00\x00\x0fh\x00hP sit \x05\x00\x00\x00@amet'
      output = b'Lorem ipsum dolor sit amet' * 10
      assert decompress(input, d_kwargs) == output
  
files.old_path: tests/stream/test_stream_1.py
files.new_contents: import pytest
  import sys
  import lz4.stream
  import psutil
  import os
  import gc
  
  
  def run_gc(func):
      if os.environ.get('TRAVIS') is not None or os.environ.get('APPVEYOR') is not None:
          def wrapper(*args, **kwargs):
              return func(*args, **kwargs)
      else:
          def wrapper(*args, **kwargs):
              gc.collect()
              try:
                  result = func(*args, **kwargs)
              finally:
                  gc.collect()
              return result
  
      wrapper.__name__ = func.__name__
      return wrapper
  
  
  # This test requires allocating a big lump of memory. In order to
  # avoid a massive memory allocation during byte compilation, we have
  # to declare a variable for the size of the buffer we're going to
  # create outside the scope of the function below. See:
  # https://bugs.python.org/issue21074
  
  _4GB = 0xffffffff  # actually 4GB - 1B, the maximum size on 4 bytes.
  
  # This test will be killed on Travis due to the 3GB memory limit
  # there. Unfortunately psutil reports the host memory, not the memory
  # available to the container, and so can't be used to detect available
  # memory, so instead, as an ugly hack for detecting we're on Travis we
  # check for the TRAVIS environment variable being set. This is quite
  # fragile.
  
  if os.environ.get('TRAVIS') is not None or sys.maxsize < _4GB or \
     psutil.virtual_memory().available < _4GB:
      huge = None
  else:
      try:
          huge = b'\0' * _4GB
      except (MemoryError, OverflowError):
          huge = None
  
  
  @pytest.mark.skipif(
      os.environ.get('TRAVIS') is not None,
      reason='Skipping test on Travis due to insufficient memory'
  )
  @pytest.mark.skipif(
      sys.maxsize < _4GB,
      reason='Py_ssize_t too small for this test'
  )
  @pytest.mark.skipif(
      psutil.virtual_memory().available < _4GB or huge is None,
      reason='Insufficient system memory for this test'
  )
  @run_gc
  def test_huge_1():
      data = b''
      kwargs = {
          'strategy': "double_buffer",
          'buffer_size': lz4.stream.LZ4_MAX_INPUT_SIZE,
          'store_comp_size': 4,
          'dictionary': huge,
      }
  
      if psutil.virtual_memory().available < 3 * kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer)
          pytest.skip('Insufficient system memory for this test')
  
      # Triggering overflow error
      message = r'^Dictionary too large for LZ4 API$'
  
      with pytest.raises(OverflowError, match=message):
          with lz4.stream.LZ4StreamCompressor(**kwargs) as proc:
              proc.compress(data)
  
      with pytest.raises(OverflowError, match=message):
          with lz4.stream.LZ4StreamDecompressor(**kwargs) as proc:
              proc.decompress(data)
  
  
  @pytest.mark.skipif(
      os.environ.get('TRAVIS') is not None,
      reason='Skipping test on Travis due to insufficient memory'
  )
  @pytest.mark.skipif(
      sys.maxsize < 0xffffffff,
      reason='Py_ssize_t too small for this test'
  )
  @pytest.mark.skipif(
      psutil.virtual_memory().available < _4GB or huge is None,
      reason='Insufficient system memory for this test'
  )
  @run_gc
  def test_huge_2():
      data = huge
      kwargs = {
          'strategy': "double_buffer",
          'buffer_size': lz4.stream.LZ4_MAX_INPUT_SIZE,
          'store_comp_size': 4,
          'dictionary': b'',
      }
  
      if psutil.virtual_memory().available < 3 * kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer)
          pytest.skip('Insufficient system memory for this test')
  
      # Raising overflow error
      message = r'^Input too large for LZ4 API$'
  
      with pytest.raises(OverflowError, match=message):
          with lz4.stream.LZ4StreamCompressor(**kwargs) as proc:
              proc.compress(data)
  
      # On decompression, too large input will raise LZ4StreamError
      with pytest.raises(lz4.stream.LZ4StreamError):
          with lz4.stream.LZ4StreamDecompressor(**kwargs) as proc:
              proc.decompress(data)
  
  
  @pytest.mark.skipif(
      os.environ.get('TRAVIS') is not None,
      reason='Skipping test on Travis due to insufficient memory'
  )
  @pytest.mark.skipif(
      sys.maxsize < 0xffffffff,
      reason='Py_ssize_t too small for this test'
  )
  @pytest.mark.skipif(
      psutil.virtual_memory().available < _4GB or huge is None,
      reason='Insufficient system memory for this test'
  )
  @run_gc
  def test_huge_3():
      data = huge
      kwargs = {
          'strategy': "double_buffer",
          'buffer_size': lz4.stream.LZ4_MAX_INPUT_SIZE,
          'store_comp_size': 4,
          'dictionary': huge,
      }
  
      if psutil.virtual_memory().available < 3 * kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer)
          pytest.skip('Insufficient system memory for this test')
  
      # Raising overflow error (during initialization because of the dictionary parameter)
      message = r'^Dictionary too large for LZ4 API$'
  
      with pytest.raises(OverflowError, match=message):
          with lz4.stream.LZ4StreamCompressor(**kwargs) as proc:
              proc.compress(data)
  
      with pytest.raises(OverflowError, match=message):
          with lz4.stream.LZ4StreamDecompressor(**kwargs) as proc:
              proc.decompress(data)
  
  
  def test_dummy():
      pass
  
files.new_path: tests/stream/test_stream_2.py
files.old_contents: import pytest
  import sys
  import lz4.stream
  import psutil
  import os
  
  
  # This test requires allocating a big lump of memory. In order to
  # avoid a massive memory allocation during byte compilation, we have
  # to declare a variable for the size of the buffer we're going to
  # create outside the scope of the function below. See:
  # https://bugs.python.org/issue21074
  
  _4GB = 0xffffffff  # actually 4GB - 1B, the maximum size on 4 bytes.
  
  # This test will be killed on Travis due to the 3GB memory limit
  # there. Unfortunately psutil reports the host memory, not the memory
  # available to the container, and so can't be used to detect available
  # memory, so instead, as an ugly hack for detecting we're on Travis we
  # check for the TRAVIS environment variable being set. This is quite
  # fragile.
  
  if os.environ.get('TRAVIS') is not None or sys.maxsize < _4GB or \
     psutil.virtual_memory().available < _4GB:
      huge = None
  else:
      try:
          huge = b'\0' * _4GB
      except (MemoryError, OverflowError):
          huge = None
  
  
  @pytest.mark.skipif(
      os.environ.get('TRAVIS') is not None,
      reason='Skipping test on Travis due to insufficient memory'
  )
  @pytest.mark.skipif(
      sys.maxsize < _4GB,
      reason='Py_ssize_t too small for this test'
  )
  @pytest.mark.skipif(
      psutil.virtual_memory().available < _4GB or huge is None,
      reason='Insufficient system memory for this test'
  )
  def test_huge_1():
      data = b''
      kwargs = {
          'strategy': "double_buffer",
          'buffer_size': lz4.stream.LZ4_MAX_INPUT_SIZE,
          'store_comp_size': 4,
          'dictionary': huge,
      }
  
      if psutil.virtual_memory().available < 3 * kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer)
          pytest.skip('Insufficient system memory for this test')
  
      # Triggering overflow error
      message = r'^Dictionary too large for LZ4 API$'
  
      with pytest.raises(OverflowError, match=message):
          with lz4.stream.LZ4StreamCompressor(**kwargs) as proc:
              proc.compress(data)
  
      with pytest.raises(OverflowError, match=message):
          with lz4.stream.LZ4StreamDecompressor(**kwargs) as proc:
              proc.decompress(data)
  
  
  @pytest.mark.skipif(
      os.environ.get('TRAVIS') is not None,
      reason='Skipping test on Travis due to insufficient memory'
  )
  @pytest.mark.skipif(
      sys.maxsize < 0xffffffff,
      reason='Py_ssize_t too small for this test'
  )
  @pytest.mark.skipif(
      psutil.virtual_memory().available < _4GB or huge is None,
      reason='Insufficient system memory for this test'
  )
  def test_huge_2():
      data = huge
      kwargs = {
          'strategy': "double_buffer",
          'buffer_size': lz4.stream.LZ4_MAX_INPUT_SIZE,
          'store_comp_size': 4,
          'dictionary': b'',
      }
  
      if psutil.virtual_memory().available < 3 * kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer)
          pytest.skip('Insufficient system memory for this test')
  
      # Raising overflow error
      message = r'^Input too large for LZ4 API$'
  
      with pytest.raises(OverflowError, match=message):
          with lz4.stream.LZ4StreamCompressor(**kwargs) as proc:
              proc.compress(data)
  
      # On decompression, too large input will raise LZ4StreamError
      with pytest.raises(lz4.stream.LZ4StreamError):
          with lz4.stream.LZ4StreamDecompressor(**kwargs) as proc:
              proc.decompress(data)
  
  
  @pytest.mark.skipif(
      os.environ.get('TRAVIS') is not None,
      reason='Skipping test on Travis due to insufficient memory'
  )
  @pytest.mark.skipif(
      sys.maxsize < 0xffffffff,
      reason='Py_ssize_t too small for this test'
  )
  @pytest.mark.skipif(
      psutil.virtual_memory().available < _4GB or huge is None,
      reason='Insufficient system memory for this test'
  )
  def test_huge_3():
      data = huge
      kwargs = {
          'strategy': "double_buffer",
          'buffer_size': lz4.stream.LZ4_MAX_INPUT_SIZE,
          'store_comp_size': 4,
          'dictionary': huge,
      }
  
      if psutil.virtual_memory().available < 3 * kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer)
          pytest.skip('Insufficient system memory for this test')
  
      # Raising overflow error (during initialization because of the dictionary parameter)
      message = r'^Dictionary too large for LZ4 API$'
  
      with pytest.raises(OverflowError, match=message):
          with lz4.stream.LZ4StreamCompressor(**kwargs) as proc:
              proc.compress(data)
  
      with pytest.raises(OverflowError, match=message):
          with lz4.stream.LZ4StreamDecompressor(**kwargs) as proc:
              proc.decompress(data)
  
  
  def test_dummy():
      pass
  
files.old_path: tests/stream/test_stream_2.py
files.new_contents: import lz4.stream
  import pytest
  import sys
  import os
  import psutil
  import gc
  
  
  _1KB = 1024
  _1MB = _1KB * 1024
  _1GB = _1MB * 1024
  
  
  def run_gc_param_data_buffer_size(func):
      if os.environ.get('TRAVIS') is not None or os.environ.get('APPVEYOR') is not None:
          def wrapper(data, buffer_size, *args, **kwargs):
              return func(data, buffer_size, *args, **kwargs)
      else:
          def wrapper(data, buffer_size, *args, **kwargs):
              gc.collect()
              try:
                  result = func(data, buffer_size, *args, **kwargs)
              finally:
                  gc.collect()
              return result
  
      wrapper.__name__ = func.__name__
      return wrapper
  
  
  def compress(x, c_kwargs):
      if c_kwargs.get('return_bytearray', False):
          c = bytearray()
      else:
          c = bytes()
      with lz4.stream.LZ4StreamCompressor(**c_kwargs) as proc:
          for start in range(0, len(x), c_kwargs['buffer_size']):
              chunk = x[start:start + c_kwargs['buffer_size']]
              block = proc.compress(chunk)
              c += block
      return c
  
  
  def decompress(x, d_kwargs):
      if d_kwargs.get('return_bytearray', False):
          d = bytearray()
      else:
          d = bytes()
      with lz4.stream.LZ4StreamDecompressor(**d_kwargs) as proc:
          start = 0
          while start < len(x):
              block = proc.get_block(x[start:])
              chunk = proc.decompress(block)
              d += chunk
              start += d_kwargs['store_comp_size'] + len(block)
      return d
  
  
  test_buffer_size = sorted(
      [256,
       1 * _1KB,
       64 * _1KB,
       1 * _1MB,
       1 * _1GB,
       lz4.stream.LZ4_MAX_INPUT_SIZE]
  )
  
  
  @pytest.fixture(
      params=test_buffer_size,
      ids=[
          'buffer_size' + str(i) for i in range(len(test_buffer_size))
      ]
  )
  def buffer_size(request):
      return request.param
  
  
  test_data = [
      (b'a' * _1MB),
  ]
  
  
  @pytest.fixture(
      params=test_data,
      ids=[
          'data' + str(i) for i in range(len(test_data))
      ]
  )
  def data(request):
      return request.param
  
  
  @run_gc_param_data_buffer_size
  def test_block_decompress_mem_usage(data, buffer_size):
      kwargs = {
          'strategy': "double_buffer",
          'buffer_size': buffer_size,
          'store_comp_size': 4,
      }
  
      if os.environ.get('TRAVIS') is not None:
          pytest.skip('Skipping test on Travis due to insufficient memory')
  
      if os.environ.get('APPVEYOR') is not None:
          pytest.skip('Skipping test on AppVeyor due to insufficient resources')
  
      if sys.maxsize < 0xffffffff:
          pytest.skip('Py_ssize_t too small for this test')
  
      if psutil.virtual_memory().available < 3 * kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer)
          pytest.skip('Insufficient system memory for this test')
  
      tracemalloc = pytest.importorskip('tracemalloc')
  
      # Trace memory usage on compression
      tracemalloc.start()
      prev_snapshot = None
  
      for i in range(1000):
          compressed = compress(data, kwargs)
  
          if i % 100 == 0:
              snapshot = tracemalloc.take_snapshot()
  
              if prev_snapshot:
                  # Filter on lz4.stream module'a allocations
                  stats = [x for x in snapshot.compare_to(prev_snapshot, 'lineno')
                           if lz4.stream.__file__ in x.traceback._frames[0][0]]
                  assert sum(map(lambda x: x.size_diff, stats)) < (1024 * 4)
  
              prev_snapshot = snapshot
  
      tracemalloc.stop()
  
      tracemalloc.start()
      prev_snapshot = None
  
      for i in range(1000):
          decompressed = decompress(compressed, kwargs)  # noqa: F841
  
          if i % 100 == 0:
              snapshot = tracemalloc.take_snapshot()
  
              if prev_snapshot:
                  # Filter on lz4.stream module'a allocations
                  stats = [x for x in snapshot.compare_to(prev_snapshot, 'lineno')
                           if lz4.stream.__file__ in x.traceback._frames[0][0]]
                  assert sum(map(lambda x: x.size_diff, stats)) < (1024 * 4)
  
              prev_snapshot = snapshot
  
      tracemalloc.stop()
  
files.new_path: tests/stream/test_stream_3.py
files.old_contents: import lz4.stream
  import pytest
  import sys
  import os
  import psutil
  
  
  _1KB = 1024
  _1MB = _1KB * 1024
  _1GB = _1MB * 1024
  
  
  def compress(x, c_kwargs):
      if c_kwargs.get('return_bytearray', False):
          c = bytearray()
      else:
          c = bytes()
      with lz4.stream.LZ4StreamCompressor(**c_kwargs) as proc:
          for start in range(0, len(x), c_kwargs['buffer_size']):
              chunk = x[start:start + c_kwargs['buffer_size']]
              block = proc.compress(chunk)
              c += block
      return c
  
  
  def decompress(x, d_kwargs):
      if d_kwargs.get('return_bytearray', False):
          d = bytearray()
      else:
          d = bytes()
      with lz4.stream.LZ4StreamDecompressor(**d_kwargs) as proc:
          start = 0
          while start < len(x):
              block = proc.get_block(x[start:])
              chunk = proc.decompress(block)
              d += chunk
              start += d_kwargs['store_comp_size'] + len(block)
      return d
  
  
  test_buffer_size = sorted(
      [256,
       1 * _1KB,
       64 * _1KB,
       1 * _1MB,
       1 * _1GB,
       lz4.stream.LZ4_MAX_INPUT_SIZE]
  )
  
  
  @pytest.fixture(
      params=test_buffer_size,
      ids=[
          'buffer_size' + str(i) for i in range(len(test_buffer_size))
      ]
  )
  def buffer_size(request):
      return request.param
  
  
  test_data = [
      (b'a' * _1MB),
  ]
  
  
  @pytest.fixture(
      params=test_data,
      ids=[
          'data' + str(i) for i in range(len(test_data))
      ]
  )
  def data(request):
      return request.param
  
  
  def test_block_decompress_mem_usage(data, buffer_size):
      kwargs = {
          'strategy': "double_buffer",
          'buffer_size': buffer_size,
          'store_comp_size': 4,
      }
  
      if os.environ.get('TRAVIS') is not None:
          pytest.skip('Skipping test on Travis due to insufficient memory')
  
      if os.environ.get('APPVEYOR') is not None:
          pytest.skip('Skipping test on AppVeyor due to insufficient resources')
  
      if sys.maxsize < 0xffffffff:
          pytest.skip('Py_ssize_t too small for this test')
  
      if psutil.virtual_memory().available < 3 * kwargs['buffer_size']:
          # The internal LZ4 context will request at least 3 times buffer_size
          # as memory (2 buffer_size for the double-buffer, and 1.x buffer_size
          # for the output buffer)
          pytest.skip('Insufficient system memory for this test')
  
      tracemalloc = pytest.importorskip('tracemalloc')
  
      # Trace memory usage on compression
      tracemalloc.start()
      prev_snapshot = None
  
      for i in range(1000):
          compressed = compress(data, kwargs)
  
          if i % 100 == 0:
              snapshot = tracemalloc.take_snapshot()
  
              if prev_snapshot:
                  # Filter on lz4.stream module'a allocations
                  stats = [x for x in snapshot.compare_to(prev_snapshot, 'lineno')
                           if lz4.stream.__file__ in x.traceback._frames[0][0]]
                  assert sum(map(lambda x: x.size_diff, stats)) < (1024 * 4)
  
              prev_snapshot = snapshot
  
      tracemalloc.stop()
  
      tracemalloc.start()
      prev_snapshot = None
  
      for i in range(1000):
          decompressed = decompress(compressed, kwargs)  # noqa: F841
  
          if i % 100 == 0:
              snapshot = tracemalloc.take_snapshot()
  
              if prev_snapshot:
                  # Filter on lz4.stream module'a allocations
                  stats = [x for x in snapshot.compare_to(prev_snapshot, 'lineno')
                           if lz4.stream.__file__ in x.traceback._frames[0][0]]
                  assert sum(map(lambda x: x.size_diff, stats)) < (1024 * 4)
  
              prev_snapshot = snapshot
  
      tracemalloc.stop()
  
files.old_path: tests/stream/test_stream_3.py