Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
raise ImportError("Install xxhash package with:\n pip install xxhash\nor\n conda install -c conda-forge python-xxhash")
try:
import lz4.block
except ImportError:
raise ImportError("Install lz4 package with:\n pip install lz4\nor\n conda install -c anaconda lz4")
if level >= 4:
after_compressed = lz4.block.compress(givenbytes, compression=level, mode="high_compression", store_size=False)
else:
after_compressed = lz4.block.compress(givenbytes, store_size=False)
compressedbytes = len(after_compressed) + 8
checksum = xxhash.xxh64(after_compressed).digest()
if (compressedbytes + 9) < uncompressedbytes:
c1 = (compressedbytes >> 0) & 0xff
c2 = (compressedbytes >> 8) & 0xff
c3 = (compressedbytes >> 16) & 0xff
method = lz4.library_version_number() // (100 * 100)
cursor.write_fields(context._sink, _header, algo, method, c1, c2, c3, u1, u2, u3)
cursor.write_data(context._sink, checksum)
cursor.write_data(context._sink, after_compressed)
key.fObjlen = uncompressedbytes
key.fNbytes = compressedbytes + key.fKeylen + 9
key.write(keycursor, context._sink)
else:
key.fObjlen = len(givenbytes)
key.fNbytes += key.fObjlen
key.write(keycursor, context._sink)
cursor.write_data(context._sink, givenbytes)
elif algorithm == uproot.const.kLZMA:
algo = b"XZ"
try:
import lzma
def test_library_version_number():
v = lz4.library_version_number()
assert isinstance(v, int)
assert v > 10000
def test_reset_decompression_context_1():
if lz4.library_version_number() >= 10800:
context = lz4frame.create_decompression_context()
r = lz4frame.reset_decompression_context(context)
assert r is None
else:
pass
def test_reset_decompression_context_2():
if lz4.library_version_number() >= 10800:
c = lz4frame.compress(b'1234', return_bytearray=False)
context = lz4frame.create_decompression_context()
try:
# Simulate an error by passing junk to decompress
d = lz4frame.decompress_chunk(context, c[1:3])
except:
pass
r = lz4frame.reset_decompression_context(context)
assert r is None
# And confirm we can use the context after reset
d, bytes_read, eof = lz4frame.decompress_chunk(context, c)
assert d == b'1234'
assert bytes_read == len(c)
assert eof is True
else:
pass
def __init__(self,
block_size=BLOCKSIZE_DEFAULT,
block_linked=True,
compression_level=COMPRESSIONLEVEL_MIN,
content_checksum=False,
block_checksum=False,
auto_flush=False,
return_bytearray=False):
self.block_size = block_size
self.block_linked = block_linked
self.compression_level = compression_level
self.content_checksum = content_checksum
if block_checksum and lz4.library_version_number() < 10800:
raise RuntimeError(
'Attempt to set block_checksum to True with LZ4 library'
'version < 10800'
)
self.block_checksum = block_checksum
self.auto_flush = auto_flush
self.return_bytearray = return_bytearray
self._context = None
self._started = False
def __init__(self,
block_size=BLOCKSIZE_DEFAULT,
block_linked=True,
compression_level=COMPRESSIONLEVEL_MIN,
content_checksum=False,
block_checksum=False,
auto_flush=False,
return_bytearray=False):
self.block_size = block_size
self.block_linked = block_linked
self.compression_level = compression_level
self.content_checksum = content_checksum
if block_checksum and lz4.library_version_number() < 10800:
raise RuntimeError(
'Attempt to set block_checksum to True with LZ4 library'
'version < 10800'
)
self.block_checksum = block_checksum
self.auto_flush = auto_flush
self.return_bytearray = return_bytearray
self._context = None
self._started = False