files: import pytest
  import sys
  import lz4.block
  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 = 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.
  
  
  @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,
      reason='Insufficient system memory for this test'
  )
  def test_huge():
      try:
          huge = b'\0' * _4GB
      except MemoryError:
          pytest.skip('Insufficient system memory for this test')
  
      with pytest.raises(
              OverflowError, match='Input too large for LZ4 API'
      ):
          lz4.block.compress(huge)
  
      with pytest.raises(
              OverflowError, match='Dictionary too large for LZ4 API'
      ):
          lz4.block.compress(b'', dict=huge)
  
      with pytest.raises(
              OverflowError, match='Input too large for LZ4 API'
      ):
          lz4.block.decompress(huge)
  
      with pytest.raises(
              OverflowError, match='Dictionary too large for LZ4 API'
      ):
          lz4.block.decompress(b'', dict=huge)
  
  
  def test_dummy():
      pass
  
  
  [... the rest is omitted ...]
files: tests/block/test_block_2.py
files: import pytest
  import sys
  import lz4.block
  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 = 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.
  
  
  @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().total < _4GB,
      reason='Insufficient system memory for this test'
  )
  def test_huge():
      try:
          huge = b'\0' * _4GB
      except MemoryError:
          pytest.skip('Insufficient system memory for this test')
  
      with pytest.raises(
              OverflowError, match='Input too large for LZ4 API'
      ):
          lz4.block.compress(huge)
  
      with pytest.raises(
              OverflowError, match='Dictionary too large for LZ4 API'
      ):
          lz4.block.compress(b'', dict=huge)
  
      with pytest.raises(
              OverflowError, match='Input too large for LZ4 API'
      ):
          lz4.block.decompress(huge)
  
      with pytest.raises(
              OverflowError, match='Dictionary too large for LZ4 API'
      ):
          lz4.block.decompress(b'', dict=huge)
  
  
  def test_dummy():
      pass
  
  
  [... the rest is omitted ...]
files: tests/block/test_block_2.py
files: 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(store_comp_size):
      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(store_comp_size):
      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:
  
          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')
  
          # 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
  
  [... the rest is omitted ...]
files: tests/stream/test_stream_1.py
files: 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(store_comp_size):
      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(store_comp_size):
      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:
  
          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().total < 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')
  
          # 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
  
  [... the rest is omitted ...]
files: tests/stream/test_stream_1.py
files: 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.
  
  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
  
  
  [... the rest is omitted ...]
files: tests/stream/test_stream_2.py
files: 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.
  
  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().total < _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().total < 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().total < _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().total < 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().total < _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().total < 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
  
  
  [... the rest is omitted ...]
files: tests/stream/test_stream_2.py
files: 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()
  
  
  [... the rest is omitted ...]
files: tests/stream/test_stream_3.py
files: 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().total < 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()
  
  
  [... the rest is omitted ...]
files: tests/stream/test_stream_3.py