--- a/contrib/python-zstandard/tests/test_compressor.py Fri Dec 27 18:54:57 2019 -0500
+++ b/contrib/python-zstandard/tests/test_compressor.py Sat Dec 28 09:55:45 2019 -0800
@@ -13,6 +13,7 @@
make_cffi,
NonClosingBytesIO,
OpCountingBytesIO,
+ TestCase,
)
@@ -23,14 +24,13 @@
def multithreaded_chunk_size(level, source_size=0):
- params = zstd.ZstdCompressionParameters.from_level(level,
- source_size=source_size)
+ params = zstd.ZstdCompressionParameters.from_level(level, source_size=source_size)
return 1 << (params.window_log + 2)
@make_cffi
-class TestCompressor(unittest.TestCase):
+class TestCompressor(TestCase):
def test_level_bounds(self):
with self.assertRaises(ValueError):
zstd.ZstdCompressor(level=23)
@@ -41,11 +41,11 @@
@make_cffi
-class TestCompressor_compress(unittest.TestCase):
+class TestCompressor_compress(TestCase):
def test_compress_empty(self):
cctx = zstd.ZstdCompressor(level=1, write_content_size=False)
- result = cctx.compress(b'')
- self.assertEqual(result, b'\x28\xb5\x2f\xfd\x00\x48\x01\x00\x00')
+ result = cctx.compress(b"")
+ self.assertEqual(result, b"\x28\xb5\x2f\xfd\x00\x48\x01\x00\x00")
params = zstd.get_frame_parameters(result)
self.assertEqual(params.content_size, zstd.CONTENTSIZE_UNKNOWN)
self.assertEqual(params.window_size, 524288)
@@ -53,21 +53,21 @@
self.assertFalse(params.has_checksum, 0)
cctx = zstd.ZstdCompressor()
- result = cctx.compress(b'')
- self.assertEqual(result, b'\x28\xb5\x2f\xfd\x20\x00\x01\x00\x00')
+ result = cctx.compress(b"")
+ self.assertEqual(result, b"\x28\xb5\x2f\xfd\x20\x00\x01\x00\x00")
params = zstd.get_frame_parameters(result)
self.assertEqual(params.content_size, 0)
def test_input_types(self):
cctx = zstd.ZstdCompressor(level=1, write_content_size=False)
- expected = b'\x28\xb5\x2f\xfd\x00\x00\x19\x00\x00\x66\x6f\x6f'
+ expected = b"\x28\xb5\x2f\xfd\x00\x00\x19\x00\x00\x66\x6f\x6f"
mutable_array = bytearray(3)
- mutable_array[:] = b'foo'
+ mutable_array[:] = b"foo"
sources = [
- memoryview(b'foo'),
- bytearray(b'foo'),
+ memoryview(b"foo"),
+ bytearray(b"foo"),
mutable_array,
]
@@ -77,43 +77,46 @@
def test_compress_large(self):
chunks = []
for i in range(255):
- chunks.append(struct.Struct('>B').pack(i) * 16384)
+ chunks.append(struct.Struct(">B").pack(i) * 16384)
cctx = zstd.ZstdCompressor(level=3, write_content_size=False)
- result = cctx.compress(b''.join(chunks))
+ result = cctx.compress(b"".join(chunks))
self.assertEqual(len(result), 999)
- self.assertEqual(result[0:4], b'\x28\xb5\x2f\xfd')
+ self.assertEqual(result[0:4], b"\x28\xb5\x2f\xfd")
# This matches the test for read_to_iter() below.
cctx = zstd.ZstdCompressor(level=1, write_content_size=False)
- result = cctx.compress(b'f' * zstd.COMPRESSION_RECOMMENDED_INPUT_SIZE + b'o')
- self.assertEqual(result, b'\x28\xb5\x2f\xfd\x00\x40\x54\x00\x00'
- b'\x10\x66\x66\x01\x00\xfb\xff\x39\xc0'
- b'\x02\x09\x00\x00\x6f')
+ result = cctx.compress(b"f" * zstd.COMPRESSION_RECOMMENDED_INPUT_SIZE + b"o")
+ self.assertEqual(
+ result,
+ b"\x28\xb5\x2f\xfd\x00\x40\x54\x00\x00"
+ b"\x10\x66\x66\x01\x00\xfb\xff\x39\xc0"
+ b"\x02\x09\x00\x00\x6f",
+ )
def test_negative_level(self):
cctx = zstd.ZstdCompressor(level=-4)
- result = cctx.compress(b'foo' * 256)
+ result = cctx.compress(b"foo" * 256)
def test_no_magic(self):
- params = zstd.ZstdCompressionParameters.from_level(
- 1, format=zstd.FORMAT_ZSTD1)
+ params = zstd.ZstdCompressionParameters.from_level(1, format=zstd.FORMAT_ZSTD1)
cctx = zstd.ZstdCompressor(compression_params=params)
- magic = cctx.compress(b'foobar')
+ magic = cctx.compress(b"foobar")
params = zstd.ZstdCompressionParameters.from_level(
- 1, format=zstd.FORMAT_ZSTD1_MAGICLESS)
+ 1, format=zstd.FORMAT_ZSTD1_MAGICLESS
+ )
cctx = zstd.ZstdCompressor(compression_params=params)
- no_magic = cctx.compress(b'foobar')
+ no_magic = cctx.compress(b"foobar")
- self.assertEqual(magic[0:4], b'\x28\xb5\x2f\xfd')
+ self.assertEqual(magic[0:4], b"\x28\xb5\x2f\xfd")
self.assertEqual(magic[4:], no_magic)
def test_write_checksum(self):
cctx = zstd.ZstdCompressor(level=1)
- no_checksum = cctx.compress(b'foobar')
+ no_checksum = cctx.compress(b"foobar")
cctx = zstd.ZstdCompressor(level=1, write_checksum=True)
- with_checksum = cctx.compress(b'foobar')
+ with_checksum = cctx.compress(b"foobar")
self.assertEqual(len(with_checksum), len(no_checksum) + 4)
@@ -125,9 +128,9 @@
def test_write_content_size(self):
cctx = zstd.ZstdCompressor(level=1)
- with_size = cctx.compress(b'foobar' * 256)
+ with_size = cctx.compress(b"foobar" * 256)
cctx = zstd.ZstdCompressor(level=1, write_content_size=False)
- no_size = cctx.compress(b'foobar' * 256)
+ no_size = cctx.compress(b"foobar" * 256)
self.assertEqual(len(with_size), len(no_size) + 1)
@@ -139,17 +142,17 @@
def test_no_dict_id(self):
samples = []
for i in range(128):
- samples.append(b'foo' * 64)
- samples.append(b'bar' * 64)
- samples.append(b'foobar' * 64)
+ samples.append(b"foo" * 64)
+ samples.append(b"bar" * 64)
+ samples.append(b"foobar" * 64)
d = zstd.train_dictionary(1024, samples)
cctx = zstd.ZstdCompressor(level=1, dict_data=d)
- with_dict_id = cctx.compress(b'foobarfoobar')
+ with_dict_id = cctx.compress(b"foobarfoobar")
cctx = zstd.ZstdCompressor(level=1, dict_data=d, write_dict_id=False)
- no_dict_id = cctx.compress(b'foobarfoobar')
+ no_dict_id = cctx.compress(b"foobarfoobar")
self.assertEqual(len(with_dict_id), len(no_dict_id) + 4)
@@ -161,23 +164,23 @@
def test_compress_dict_multiple(self):
samples = []
for i in range(128):
- samples.append(b'foo' * 64)
- samples.append(b'bar' * 64)
- samples.append(b'foobar' * 64)
+ samples.append(b"foo" * 64)
+ samples.append(b"bar" * 64)
+ samples.append(b"foobar" * 64)
d = zstd.train_dictionary(8192, samples)
cctx = zstd.ZstdCompressor(level=1, dict_data=d)
for i in range(32):
- cctx.compress(b'foo bar foobar foo bar foobar')
+ cctx.compress(b"foo bar foobar foo bar foobar")
def test_dict_precompute(self):
samples = []
for i in range(128):
- samples.append(b'foo' * 64)
- samples.append(b'bar' * 64)
- samples.append(b'foobar' * 64)
+ samples.append(b"foo" * 64)
+ samples.append(b"bar" * 64)
+ samples.append(b"foobar" * 64)
d = zstd.train_dictionary(8192, samples)
d.precompute_compress(level=1)
@@ -185,11 +188,11 @@
cctx = zstd.ZstdCompressor(level=1, dict_data=d)
for i in range(32):
- cctx.compress(b'foo bar foobar foo bar foobar')
+ cctx.compress(b"foo bar foobar foo bar foobar")
def test_multithreaded(self):
chunk_size = multithreaded_chunk_size(1)
- source = b''.join([b'x' * chunk_size, b'y' * chunk_size])
+ source = b"".join([b"x" * chunk_size, b"y" * chunk_size])
cctx = zstd.ZstdCompressor(level=1, threads=2)
compressed = cctx.compress(source)
@@ -205,73 +208,72 @@
def test_multithreaded_dict(self):
samples = []
for i in range(128):
- samples.append(b'foo' * 64)
- samples.append(b'bar' * 64)
- samples.append(b'foobar' * 64)
+ samples.append(b"foo" * 64)
+ samples.append(b"bar" * 64)
+ samples.append(b"foobar" * 64)
d = zstd.train_dictionary(1024, samples)
cctx = zstd.ZstdCompressor(dict_data=d, threads=2)
- result = cctx.compress(b'foo')
- params = zstd.get_frame_parameters(result);
- self.assertEqual(params.content_size, 3);
+ result = cctx.compress(b"foo")
+ params = zstd.get_frame_parameters(result)
+ self.assertEqual(params.content_size, 3)
self.assertEqual(params.dict_id, d.dict_id())
- self.assertEqual(result,
- b'\x28\xb5\x2f\xfd\x23\x8f\x55\x0f\x70\x03\x19\x00\x00'
- b'\x66\x6f\x6f')
+ self.assertEqual(
+ result,
+ b"\x28\xb5\x2f\xfd\x23\x8f\x55\x0f\x70\x03\x19\x00\x00" b"\x66\x6f\x6f",
+ )
def test_multithreaded_compression_params(self):
params = zstd.ZstdCompressionParameters.from_level(0, threads=2)
cctx = zstd.ZstdCompressor(compression_params=params)
- result = cctx.compress(b'foo')
- params = zstd.get_frame_parameters(result);
- self.assertEqual(params.content_size, 3);
+ result = cctx.compress(b"foo")
+ params = zstd.get_frame_parameters(result)
+ self.assertEqual(params.content_size, 3)
- self.assertEqual(result,
- b'\x28\xb5\x2f\xfd\x20\x03\x19\x00\x00\x66\x6f\x6f')
+ self.assertEqual(result, b"\x28\xb5\x2f\xfd\x20\x03\x19\x00\x00\x66\x6f\x6f")
@make_cffi
-class TestCompressor_compressobj(unittest.TestCase):
+class TestCompressor_compressobj(TestCase):
def test_compressobj_empty(self):
cctx = zstd.ZstdCompressor(level=1, write_content_size=False)
cobj = cctx.compressobj()
- self.assertEqual(cobj.compress(b''), b'')
- self.assertEqual(cobj.flush(),
- b'\x28\xb5\x2f\xfd\x00\x48\x01\x00\x00')
+ self.assertEqual(cobj.compress(b""), b"")
+ self.assertEqual(cobj.flush(), b"\x28\xb5\x2f\xfd\x00\x48\x01\x00\x00")
def test_input_types(self):
- expected = b'\x28\xb5\x2f\xfd\x00\x48\x19\x00\x00\x66\x6f\x6f'
+ expected = b"\x28\xb5\x2f\xfd\x00\x48\x19\x00\x00\x66\x6f\x6f"
cctx = zstd.ZstdCompressor(level=1, write_content_size=False)
mutable_array = bytearray(3)
- mutable_array[:] = b'foo'
+ mutable_array[:] = b"foo"
sources = [
- memoryview(b'foo'),
- bytearray(b'foo'),
+ memoryview(b"foo"),
+ bytearray(b"foo"),
mutable_array,
]
for source in sources:
cobj = cctx.compressobj()
- self.assertEqual(cobj.compress(source), b'')
+ self.assertEqual(cobj.compress(source), b"")
self.assertEqual(cobj.flush(), expected)
def test_compressobj_large(self):
chunks = []
for i in range(255):
- chunks.append(struct.Struct('>B').pack(i) * 16384)
+ chunks.append(struct.Struct(">B").pack(i) * 16384)
cctx = zstd.ZstdCompressor(level=3)
cobj = cctx.compressobj()
- result = cobj.compress(b''.join(chunks)) + cobj.flush()
+ result = cobj.compress(b"".join(chunks)) + cobj.flush()
self.assertEqual(len(result), 999)
- self.assertEqual(result[0:4], b'\x28\xb5\x2f\xfd')
+ self.assertEqual(result[0:4], b"\x28\xb5\x2f\xfd")
params = zstd.get_frame_parameters(result)
self.assertEqual(params.content_size, zstd.CONTENTSIZE_UNKNOWN)
@@ -282,10 +284,10 @@
def test_write_checksum(self):
cctx = zstd.ZstdCompressor(level=1)
cobj = cctx.compressobj()
- no_checksum = cobj.compress(b'foobar') + cobj.flush()
+ no_checksum = cobj.compress(b"foobar") + cobj.flush()
cctx = zstd.ZstdCompressor(level=1, write_checksum=True)
cobj = cctx.compressobj()
- with_checksum = cobj.compress(b'foobar') + cobj.flush()
+ with_checksum = cobj.compress(b"foobar") + cobj.flush()
no_params = zstd.get_frame_parameters(no_checksum)
with_params = zstd.get_frame_parameters(with_checksum)
@@ -300,11 +302,11 @@
def test_write_content_size(self):
cctx = zstd.ZstdCompressor(level=1)
- cobj = cctx.compressobj(size=len(b'foobar' * 256))
- with_size = cobj.compress(b'foobar' * 256) + cobj.flush()
+ cobj = cctx.compressobj(size=len(b"foobar" * 256))
+ with_size = cobj.compress(b"foobar" * 256) + cobj.flush()
cctx = zstd.ZstdCompressor(level=1, write_content_size=False)
- cobj = cctx.compressobj(size=len(b'foobar' * 256))
- no_size = cobj.compress(b'foobar' * 256) + cobj.flush()
+ cobj = cctx.compressobj(size=len(b"foobar" * 256))
+ no_size = cobj.compress(b"foobar" * 256) + cobj.flush()
no_params = zstd.get_frame_parameters(no_size)
with_params = zstd.get_frame_parameters(with_size)
@@ -321,48 +323,53 @@
cctx = zstd.ZstdCompressor()
cobj = cctx.compressobj()
- cobj.compress(b'foo')
+ cobj.compress(b"foo")
cobj.flush()
- with self.assertRaisesRegexp(zstd.ZstdError, r'cannot call compress\(\) after compressor'):
- cobj.compress(b'foo')
+ with self.assertRaisesRegex(
+ zstd.ZstdError, r"cannot call compress\(\) after compressor"
+ ):
+ cobj.compress(b"foo")
- with self.assertRaisesRegexp(zstd.ZstdError, 'compressor object already finished'):
+ with self.assertRaisesRegex(
+ zstd.ZstdError, "compressor object already finished"
+ ):
cobj.flush()
def test_flush_block_repeated(self):
cctx = zstd.ZstdCompressor(level=1)
cobj = cctx.compressobj()
- self.assertEqual(cobj.compress(b'foo'), b'')
- self.assertEqual(cobj.flush(zstd.COMPRESSOBJ_FLUSH_BLOCK),
- b'\x28\xb5\x2f\xfd\x00\x48\x18\x00\x00foo')
- self.assertEqual(cobj.compress(b'bar'), b'')
+ self.assertEqual(cobj.compress(b"foo"), b"")
+ self.assertEqual(
+ cobj.flush(zstd.COMPRESSOBJ_FLUSH_BLOCK),
+ b"\x28\xb5\x2f\xfd\x00\x48\x18\x00\x00foo",
+ )
+ self.assertEqual(cobj.compress(b"bar"), b"")
# 3 byte header plus content.
- self.assertEqual(cobj.flush(zstd.COMPRESSOBJ_FLUSH_BLOCK),
- b'\x18\x00\x00bar')
- self.assertEqual(cobj.flush(), b'\x01\x00\x00')
+ self.assertEqual(cobj.flush(zstd.COMPRESSOBJ_FLUSH_BLOCK), b"\x18\x00\x00bar")
+ self.assertEqual(cobj.flush(), b"\x01\x00\x00")
def test_flush_empty_block(self):
cctx = zstd.ZstdCompressor(write_checksum=True)
cobj = cctx.compressobj()
- cobj.compress(b'foobar')
+ cobj.compress(b"foobar")
cobj.flush(zstd.COMPRESSOBJ_FLUSH_BLOCK)
# No-op if no block is active (this is internal to zstd).
- self.assertEqual(cobj.flush(zstd.COMPRESSOBJ_FLUSH_BLOCK), b'')
+ self.assertEqual(cobj.flush(zstd.COMPRESSOBJ_FLUSH_BLOCK), b"")
trailing = cobj.flush()
# 3 bytes block header + 4 bytes frame checksum
self.assertEqual(len(trailing), 7)
header = trailing[0:3]
- self.assertEqual(header, b'\x01\x00\x00')
+ self.assertEqual(header, b"\x01\x00\x00")
def test_multithreaded(self):
source = io.BytesIO()
- source.write(b'a' * 1048576)
- source.write(b'b' * 1048576)
- source.write(b'c' * 1048576)
+ source.write(b"a" * 1048576)
+ source.write(b"b" * 1048576)
+ source.write(b"c" * 1048576)
source.seek(0)
cctx = zstd.ZstdCompressor(level=1, threads=2)
@@ -378,9 +385,9 @@
chunks.append(cobj.flush())
- compressed = b''.join(chunks)
+ compressed = b"".join(chunks)
- self.assertEqual(len(compressed), 295)
+ self.assertEqual(len(compressed), 119)
def test_frame_progression(self):
cctx = zstd.ZstdCompressor()
@@ -389,7 +396,7 @@
cobj = cctx.compressobj()
- cobj.compress(b'foobar')
+ cobj.compress(b"foobar")
self.assertEqual(cctx.frame_progression(), (6, 0, 0))
cobj.flush()
@@ -399,20 +406,20 @@
cctx = zstd.ZstdCompressor()
cobj = cctx.compressobj(size=2)
- with self.assertRaisesRegexp(zstd.ZstdError, 'Src size is incorrect'):
- cobj.compress(b'foo')
+ with self.assertRaisesRegex(zstd.ZstdError, "Src size is incorrect"):
+ cobj.compress(b"foo")
# Try another operation on this instance.
- with self.assertRaisesRegexp(zstd.ZstdError, 'Src size is incorrect'):
- cobj.compress(b'aa')
+ with self.assertRaisesRegex(zstd.ZstdError, "Src size is incorrect"):
+ cobj.compress(b"aa")
# Try another operation on the compressor.
cctx.compressobj(size=4)
- cctx.compress(b'foobar')
+ cctx.compress(b"foobar")
@make_cffi
-class TestCompressor_copy_stream(unittest.TestCase):
+class TestCompressor_copy_stream(TestCase):
def test_no_read(self):
source = object()
dest = io.BytesIO()
@@ -438,13 +445,12 @@
self.assertEqual(int(r), 0)
self.assertEqual(w, 9)
- self.assertEqual(dest.getvalue(),
- b'\x28\xb5\x2f\xfd\x00\x48\x01\x00\x00')
+ self.assertEqual(dest.getvalue(), b"\x28\xb5\x2f\xfd\x00\x48\x01\x00\x00")
def test_large_data(self):
source = io.BytesIO()
for i in range(255):
- source.write(struct.Struct('>B').pack(i) * 16384)
+ source.write(struct.Struct(">B").pack(i) * 16384)
source.seek(0)
dest = io.BytesIO()
@@ -461,7 +467,7 @@
self.assertFalse(params.has_checksum)
def test_write_checksum(self):
- source = io.BytesIO(b'foobar')
+ source = io.BytesIO(b"foobar")
no_checksum = io.BytesIO()
cctx = zstd.ZstdCompressor(level=1)
@@ -472,8 +478,7 @@
cctx = zstd.ZstdCompressor(level=1, write_checksum=True)
cctx.copy_stream(source, with_checksum)
- self.assertEqual(len(with_checksum.getvalue()),
- len(no_checksum.getvalue()) + 4)
+ self.assertEqual(len(with_checksum.getvalue()), len(no_checksum.getvalue()) + 4)
no_params = zstd.get_frame_parameters(no_checksum.getvalue())
with_params = zstd.get_frame_parameters(with_checksum.getvalue())
@@ -485,7 +490,7 @@
self.assertTrue(with_params.has_checksum)
def test_write_content_size(self):
- source = io.BytesIO(b'foobar' * 256)
+ source = io.BytesIO(b"foobar" * 256)
no_size = io.BytesIO()
cctx = zstd.ZstdCompressor(level=1, write_content_size=False)
@@ -497,16 +502,14 @@
cctx.copy_stream(source, with_size)
# Source content size is unknown, so no content size written.
- self.assertEqual(len(with_size.getvalue()),
- len(no_size.getvalue()))
+ self.assertEqual(len(with_size.getvalue()), len(no_size.getvalue()))
source.seek(0)
with_size = io.BytesIO()
cctx.copy_stream(source, with_size, size=len(source.getvalue()))
# We specified source size, so content size header is present.
- self.assertEqual(len(with_size.getvalue()),
- len(no_size.getvalue()) + 1)
+ self.assertEqual(len(with_size.getvalue()), len(no_size.getvalue()) + 1)
no_params = zstd.get_frame_parameters(no_size.getvalue())
with_params = zstd.get_frame_parameters(with_size.getvalue())
@@ -518,7 +521,7 @@
self.assertFalse(with_params.has_checksum)
def test_read_write_size(self):
- source = OpCountingBytesIO(b'foobarfoobar')
+ source = OpCountingBytesIO(b"foobarfoobar")
dest = OpCountingBytesIO()
cctx = zstd.ZstdCompressor()
r, w = cctx.copy_stream(source, dest, read_size=1, write_size=1)
@@ -530,16 +533,16 @@
def test_multithreaded(self):
source = io.BytesIO()
- source.write(b'a' * 1048576)
- source.write(b'b' * 1048576)
- source.write(b'c' * 1048576)
+ source.write(b"a" * 1048576)
+ source.write(b"b" * 1048576)
+ source.write(b"c" * 1048576)
source.seek(0)
dest = io.BytesIO()
cctx = zstd.ZstdCompressor(threads=2, write_content_size=False)
r, w = cctx.copy_stream(source, dest)
self.assertEqual(r, 3145728)
- self.assertEqual(w, 295)
+ self.assertEqual(w, 111)
params = zstd.get_frame_parameters(dest.getvalue())
self.assertEqual(params.content_size, zstd.CONTENTSIZE_UNKNOWN)
@@ -559,15 +562,15 @@
def test_bad_size(self):
source = io.BytesIO()
- source.write(b'a' * 32768)
- source.write(b'b' * 32768)
+ source.write(b"a" * 32768)
+ source.write(b"b" * 32768)
source.seek(0)
dest = io.BytesIO()
cctx = zstd.ZstdCompressor()
- with self.assertRaisesRegexp(zstd.ZstdError, 'Src size is incorrect'):
+ with self.assertRaisesRegex(zstd.ZstdError, "Src size is incorrect"):
cctx.copy_stream(source, dest, size=42)
# Try another operation on this compressor.
@@ -577,31 +580,31 @@
@make_cffi
-class TestCompressor_stream_reader(unittest.TestCase):
+class TestCompressor_stream_reader(TestCase):
def test_context_manager(self):
cctx = zstd.ZstdCompressor()
- with cctx.stream_reader(b'foo') as reader:
- with self.assertRaisesRegexp(ValueError, 'cannot __enter__ multiple times'):
+ with cctx.stream_reader(b"foo") as reader:
+ with self.assertRaisesRegex(ValueError, "cannot __enter__ multiple times"):
with reader as reader2:
pass
def test_no_context_manager(self):
cctx = zstd.ZstdCompressor()
- reader = cctx.stream_reader(b'foo')
+ reader = cctx.stream_reader(b"foo")
reader.read(4)
self.assertFalse(reader.closed)
reader.close()
self.assertTrue(reader.closed)
- with self.assertRaisesRegexp(ValueError, 'stream is closed'):
+ with self.assertRaisesRegex(ValueError, "stream is closed"):
reader.read(1)
def test_not_implemented(self):
cctx = zstd.ZstdCompressor()
- with cctx.stream_reader(b'foo' * 60) as reader:
+ with cctx.stream_reader(b"foo" * 60) as reader:
with self.assertRaises(io.UnsupportedOperation):
reader.readline()
@@ -618,12 +621,12 @@
reader.writelines([])
with self.assertRaises(OSError):
- reader.write(b'foo')
+ reader.write(b"foo")
def test_constant_methods(self):
cctx = zstd.ZstdCompressor()
- with cctx.stream_reader(b'boo') as reader:
+ with cctx.stream_reader(b"boo") as reader:
self.assertTrue(reader.readable())
self.assertFalse(reader.writable())
self.assertFalse(reader.seekable())
@@ -637,27 +640,29 @@
def test_read_closed(self):
cctx = zstd.ZstdCompressor()
- with cctx.stream_reader(b'foo' * 60) as reader:
+ with cctx.stream_reader(b"foo" * 60) as reader:
reader.close()
self.assertTrue(reader.closed)
- with self.assertRaisesRegexp(ValueError, 'stream is closed'):
+ with self.assertRaisesRegex(ValueError, "stream is closed"):
reader.read(10)
def test_read_sizes(self):
cctx = zstd.ZstdCompressor()
- foo = cctx.compress(b'foo')
+ foo = cctx.compress(b"foo")
- with cctx.stream_reader(b'foo') as reader:
- with self.assertRaisesRegexp(ValueError, 'cannot read negative amounts less than -1'):
+ with cctx.stream_reader(b"foo") as reader:
+ with self.assertRaisesRegex(
+ ValueError, "cannot read negative amounts less than -1"
+ ):
reader.read(-2)
- self.assertEqual(reader.read(0), b'')
+ self.assertEqual(reader.read(0), b"")
self.assertEqual(reader.read(), foo)
def test_read_buffer(self):
cctx = zstd.ZstdCompressor()
- source = b''.join([b'foo' * 60, b'bar' * 60, b'baz' * 60])
+ source = b"".join([b"foo" * 60, b"bar" * 60, b"baz" * 60])
frame = cctx.compress(source)
with cctx.stream_reader(source) as reader:
@@ -667,13 +672,13 @@
result = reader.read(8192)
self.assertEqual(result, frame)
self.assertEqual(reader.tell(), len(result))
- self.assertEqual(reader.read(), b'')
+ self.assertEqual(reader.read(), b"")
self.assertEqual(reader.tell(), len(result))
def test_read_buffer_small_chunks(self):
cctx = zstd.ZstdCompressor()
- source = b'foo' * 60
+ source = b"foo" * 60
chunks = []
with cctx.stream_reader(source) as reader:
@@ -687,12 +692,12 @@
chunks.append(chunk)
self.assertEqual(reader.tell(), sum(map(len, chunks)))
- self.assertEqual(b''.join(chunks), cctx.compress(source))
+ self.assertEqual(b"".join(chunks), cctx.compress(source))
def test_read_stream(self):
cctx = zstd.ZstdCompressor()
- source = b''.join([b'foo' * 60, b'bar' * 60, b'baz' * 60])
+ source = b"".join([b"foo" * 60, b"bar" * 60, b"baz" * 60])
frame = cctx.compress(source)
with cctx.stream_reader(io.BytesIO(source), size=len(source)) as reader:
@@ -701,13 +706,13 @@
chunk = reader.read(8192)
self.assertEqual(chunk, frame)
self.assertEqual(reader.tell(), len(chunk))
- self.assertEqual(reader.read(), b'')
+ self.assertEqual(reader.read(), b"")
self.assertEqual(reader.tell(), len(chunk))
def test_read_stream_small_chunks(self):
cctx = zstd.ZstdCompressor()
- source = b'foo' * 60
+ source = b"foo" * 60
chunks = []
with cctx.stream_reader(io.BytesIO(source), size=len(source)) as reader:
@@ -721,25 +726,25 @@
chunks.append(chunk)
self.assertEqual(reader.tell(), sum(map(len, chunks)))
- self.assertEqual(b''.join(chunks), cctx.compress(source))
+ self.assertEqual(b"".join(chunks), cctx.compress(source))
def test_read_after_exit(self):
cctx = zstd.ZstdCompressor()
- with cctx.stream_reader(b'foo' * 60) as reader:
+ with cctx.stream_reader(b"foo" * 60) as reader:
while reader.read(8192):
pass
- with self.assertRaisesRegexp(ValueError, 'stream is closed'):
+ with self.assertRaisesRegex(ValueError, "stream is closed"):
reader.read(10)
def test_bad_size(self):
cctx = zstd.ZstdCompressor()
- source = io.BytesIO(b'foobar')
+ source = io.BytesIO(b"foobar")
with cctx.stream_reader(source, size=2) as reader:
- with self.assertRaisesRegexp(zstd.ZstdError, 'Src size is incorrect'):
+ with self.assertRaisesRegex(zstd.ZstdError, "Src size is incorrect"):
reader.read(10)
# Try another compression operation.
@@ -748,36 +753,36 @@
def test_readall(self):
cctx = zstd.ZstdCompressor()
- frame = cctx.compress(b'foo' * 1024)
+ frame = cctx.compress(b"foo" * 1024)
- reader = cctx.stream_reader(b'foo' * 1024)
+ reader = cctx.stream_reader(b"foo" * 1024)
self.assertEqual(reader.readall(), frame)
def test_readinto(self):
cctx = zstd.ZstdCompressor()
- foo = cctx.compress(b'foo')
+ foo = cctx.compress(b"foo")
- reader = cctx.stream_reader(b'foo')
+ reader = cctx.stream_reader(b"foo")
with self.assertRaises(Exception):
- reader.readinto(b'foobar')
+ reader.readinto(b"foobar")
# readinto() with sufficiently large destination.
b = bytearray(1024)
- reader = cctx.stream_reader(b'foo')
+ reader = cctx.stream_reader(b"foo")
self.assertEqual(reader.readinto(b), len(foo))
- self.assertEqual(b[0:len(foo)], foo)
+ self.assertEqual(b[0 : len(foo)], foo)
self.assertEqual(reader.readinto(b), 0)
- self.assertEqual(b[0:len(foo)], foo)
+ self.assertEqual(b[0 : len(foo)], foo)
# readinto() with small reads.
b = bytearray(1024)
- reader = cctx.stream_reader(b'foo', read_size=1)
+ reader = cctx.stream_reader(b"foo", read_size=1)
self.assertEqual(reader.readinto(b), len(foo))
- self.assertEqual(b[0:len(foo)], foo)
+ self.assertEqual(b[0 : len(foo)], foo)
# Too small destination buffer.
b = bytearray(2)
- reader = cctx.stream_reader(b'foo')
+ reader = cctx.stream_reader(b"foo")
self.assertEqual(reader.readinto(b), 2)
self.assertEqual(b[:], foo[0:2])
self.assertEqual(reader.readinto(b), 2)
@@ -787,41 +792,41 @@
def test_readinto1(self):
cctx = zstd.ZstdCompressor()
- foo = b''.join(cctx.read_to_iter(io.BytesIO(b'foo')))
+ foo = b"".join(cctx.read_to_iter(io.BytesIO(b"foo")))
- reader = cctx.stream_reader(b'foo')
+ reader = cctx.stream_reader(b"foo")
with self.assertRaises(Exception):
- reader.readinto1(b'foobar')
+ reader.readinto1(b"foobar")
b = bytearray(1024)
- source = OpCountingBytesIO(b'foo')
+ source = OpCountingBytesIO(b"foo")
reader = cctx.stream_reader(source)
self.assertEqual(reader.readinto1(b), len(foo))
- self.assertEqual(b[0:len(foo)], foo)
+ self.assertEqual(b[0 : len(foo)], foo)
self.assertEqual(source._read_count, 2)
# readinto1() with small reads.
b = bytearray(1024)
- source = OpCountingBytesIO(b'foo')
+ source = OpCountingBytesIO(b"foo")
reader = cctx.stream_reader(source, read_size=1)
self.assertEqual(reader.readinto1(b), len(foo))
- self.assertEqual(b[0:len(foo)], foo)
+ self.assertEqual(b[0 : len(foo)], foo)
self.assertEqual(source._read_count, 4)
def test_read1(self):
cctx = zstd.ZstdCompressor()
- foo = b''.join(cctx.read_to_iter(io.BytesIO(b'foo')))
+ foo = b"".join(cctx.read_to_iter(io.BytesIO(b"foo")))
- b = OpCountingBytesIO(b'foo')
+ b = OpCountingBytesIO(b"foo")
reader = cctx.stream_reader(b)
self.assertEqual(reader.read1(), foo)
self.assertEqual(b._read_count, 2)
- b = OpCountingBytesIO(b'foo')
+ b = OpCountingBytesIO(b"foo")
reader = cctx.stream_reader(b)
- self.assertEqual(reader.read1(0), b'')
+ self.assertEqual(reader.read1(0), b"")
self.assertEqual(reader.read1(2), foo[0:2])
self.assertEqual(b._read_count, 2)
self.assertEqual(reader.read1(2), foo[2:4])
@@ -829,7 +834,7 @@
@make_cffi
-class TestCompressor_stream_writer(unittest.TestCase):
+class TestCompressor_stream_writer(TestCase):
def test_io_api(self):
buffer = io.BytesIO()
cctx = zstd.ZstdCompressor()
@@ -899,7 +904,7 @@
self.assertFalse(writer.closed)
def test_fileno_file(self):
- with tempfile.TemporaryFile('wb') as tf:
+ with tempfile.TemporaryFile("wb") as tf:
cctx = zstd.ZstdCompressor()
writer = cctx.stream_writer(tf)
@@ -910,33 +915,35 @@
cctx = zstd.ZstdCompressor(level=1)
writer = cctx.stream_writer(buffer)
- writer.write(b'foo' * 1024)
+ writer.write(b"foo" * 1024)
self.assertFalse(writer.closed)
self.assertFalse(buffer.closed)
writer.close()
self.assertTrue(writer.closed)
self.assertTrue(buffer.closed)
- with self.assertRaisesRegexp(ValueError, 'stream is closed'):
- writer.write(b'foo')
+ with self.assertRaisesRegex(ValueError, "stream is closed"):
+ writer.write(b"foo")
- with self.assertRaisesRegexp(ValueError, 'stream is closed'):
+ with self.assertRaisesRegex(ValueError, "stream is closed"):
writer.flush()
- with self.assertRaisesRegexp(ValueError, 'stream is closed'):
+ with self.assertRaisesRegex(ValueError, "stream is closed"):
with writer:
pass
- self.assertEqual(buffer.getvalue(),
- b'\x28\xb5\x2f\xfd\x00\x48\x55\x00\x00\x18\x66\x6f'
- b'\x6f\x01\x00\xfa\xd3\x77\x43')
+ self.assertEqual(
+ buffer.getvalue(),
+ b"\x28\xb5\x2f\xfd\x00\x48\x55\x00\x00\x18\x66\x6f"
+ b"\x6f\x01\x00\xfa\xd3\x77\x43",
+ )
# Context manager exit should close stream.
buffer = io.BytesIO()
writer = cctx.stream_writer(buffer)
with writer:
- writer.write(b'foo')
+ writer.write(b"foo")
self.assertTrue(writer.closed)
@@ -944,10 +951,10 @@
buffer = NonClosingBytesIO()
cctx = zstd.ZstdCompressor(level=1, write_content_size=False)
with cctx.stream_writer(buffer) as compressor:
- compressor.write(b'')
+ compressor.write(b"")
result = buffer.getvalue()
- self.assertEqual(result, b'\x28\xb5\x2f\xfd\x00\x48\x01\x00\x00')
+ self.assertEqual(result, b"\x28\xb5\x2f\xfd\x00\x48\x01\x00\x00")
params = zstd.get_frame_parameters(result)
self.assertEqual(params.content_size, zstd.CONTENTSIZE_UNKNOWN)
@@ -958,11 +965,11 @@
# Test without context manager.
buffer = io.BytesIO()
compressor = cctx.stream_writer(buffer)
- self.assertEqual(compressor.write(b''), 0)
- self.assertEqual(buffer.getvalue(), b'')
+ self.assertEqual(compressor.write(b""), 0)
+ self.assertEqual(buffer.getvalue(), b"")
self.assertEqual(compressor.flush(zstd.FLUSH_FRAME), 9)
result = buffer.getvalue()
- self.assertEqual(result, b'\x28\xb5\x2f\xfd\x00\x48\x01\x00\x00')
+ self.assertEqual(result, b"\x28\xb5\x2f\xfd\x00\x48\x01\x00\x00")
params = zstd.get_frame_parameters(result)
self.assertEqual(params.content_size, zstd.CONTENTSIZE_UNKNOWN)
@@ -972,18 +979,18 @@
# Test write_return_read=True
compressor = cctx.stream_writer(buffer, write_return_read=True)
- self.assertEqual(compressor.write(b''), 0)
+ self.assertEqual(compressor.write(b""), 0)
def test_input_types(self):
- expected = b'\x28\xb5\x2f\xfd\x00\x48\x19\x00\x00\x66\x6f\x6f'
+ expected = b"\x28\xb5\x2f\xfd\x00\x48\x19\x00\x00\x66\x6f\x6f"
cctx = zstd.ZstdCompressor(level=1)
mutable_array = bytearray(3)
- mutable_array[:] = b'foo'
+ mutable_array[:] = b"foo"
sources = [
- memoryview(b'foo'),
- bytearray(b'foo'),
+ memoryview(b"foo"),
+ bytearray(b"foo"),
mutable_array,
]
@@ -1001,51 +1008,55 @@
buffer = NonClosingBytesIO()
cctx = zstd.ZstdCompressor(level=5)
with cctx.stream_writer(buffer) as compressor:
- self.assertEqual(compressor.write(b'foo'), 0)
- self.assertEqual(compressor.write(b'bar'), 0)
- self.assertEqual(compressor.write(b'x' * 8192), 0)
+ self.assertEqual(compressor.write(b"foo"), 0)
+ self.assertEqual(compressor.write(b"bar"), 0)
+ self.assertEqual(compressor.write(b"x" * 8192), 0)
result = buffer.getvalue()
- self.assertEqual(result,
- b'\x28\xb5\x2f\xfd\x00\x58\x75\x00\x00\x38\x66\x6f'
- b'\x6f\x62\x61\x72\x78\x01\x00\xfc\xdf\x03\x23')
+ self.assertEqual(
+ result,
+ b"\x28\xb5\x2f\xfd\x00\x58\x75\x00\x00\x38\x66\x6f"
+ b"\x6f\x62\x61\x72\x78\x01\x00\xfc\xdf\x03\x23",
+ )
# Test without context manager.
buffer = io.BytesIO()
compressor = cctx.stream_writer(buffer)
- self.assertEqual(compressor.write(b'foo'), 0)
- self.assertEqual(compressor.write(b'bar'), 0)
- self.assertEqual(compressor.write(b'x' * 8192), 0)
+ self.assertEqual(compressor.write(b"foo"), 0)
+ self.assertEqual(compressor.write(b"bar"), 0)
+ self.assertEqual(compressor.write(b"x" * 8192), 0)
self.assertEqual(compressor.flush(zstd.FLUSH_FRAME), 23)
result = buffer.getvalue()
- self.assertEqual(result,
- b'\x28\xb5\x2f\xfd\x00\x58\x75\x00\x00\x38\x66\x6f'
- b'\x6f\x62\x61\x72\x78\x01\x00\xfc\xdf\x03\x23')
+ self.assertEqual(
+ result,
+ b"\x28\xb5\x2f\xfd\x00\x58\x75\x00\x00\x38\x66\x6f"
+ b"\x6f\x62\x61\x72\x78\x01\x00\xfc\xdf\x03\x23",
+ )
# Test with write_return_read=True.
compressor = cctx.stream_writer(buffer, write_return_read=True)
- self.assertEqual(compressor.write(b'foo'), 3)
- self.assertEqual(compressor.write(b'barbiz'), 6)
- self.assertEqual(compressor.write(b'x' * 8192), 8192)
+ self.assertEqual(compressor.write(b"foo"), 3)
+ self.assertEqual(compressor.write(b"barbiz"), 6)
+ self.assertEqual(compressor.write(b"x" * 8192), 8192)
def test_dictionary(self):
samples = []
for i in range(128):
- samples.append(b'foo' * 64)
- samples.append(b'bar' * 64)
- samples.append(b'foobar' * 64)
+ samples.append(b"foo" * 64)
+ samples.append(b"bar" * 64)
+ samples.append(b"foobar" * 64)
d = zstd.train_dictionary(8192, samples)
h = hashlib.sha1(d.as_bytes()).hexdigest()
- self.assertEqual(h, '7a2e59a876db958f74257141045af8f912e00d4e')
+ self.assertEqual(h, "7a2e59a876db958f74257141045af8f912e00d4e")
buffer = NonClosingBytesIO()
cctx = zstd.ZstdCompressor(level=9, dict_data=d)
with cctx.stream_writer(buffer) as compressor:
- self.assertEqual(compressor.write(b'foo'), 0)
- self.assertEqual(compressor.write(b'bar'), 0)
- self.assertEqual(compressor.write(b'foo' * 16384), 0)
+ self.assertEqual(compressor.write(b"foo"), 0)
+ self.assertEqual(compressor.write(b"bar"), 0)
+ self.assertEqual(compressor.write(b"foo" * 16384), 0)
compressed = buffer.getvalue()
@@ -1056,14 +1067,15 @@
self.assertFalse(params.has_checksum)
h = hashlib.sha1(compressed).hexdigest()
- self.assertEqual(h, '0a7c05635061f58039727cdbe76388c6f4cfef06')
+ self.assertEqual(h, "0a7c05635061f58039727cdbe76388c6f4cfef06")
- source = b'foo' + b'bar' + (b'foo' * 16384)
+ source = b"foo" + b"bar" + (b"foo" * 16384)
dctx = zstd.ZstdDecompressor(dict_data=d)
- self.assertEqual(dctx.decompress(compressed, max_output_size=len(source)),
- source)
+ self.assertEqual(
+ dctx.decompress(compressed, max_output_size=len(source)), source
+ )
def test_compression_params(self):
params = zstd.ZstdCompressionParameters(
@@ -1073,14 +1085,15 @@
min_match=5,
search_log=4,
target_length=10,
- strategy=zstd.STRATEGY_FAST)
+ strategy=zstd.STRATEGY_FAST,
+ )
buffer = NonClosingBytesIO()
cctx = zstd.ZstdCompressor(compression_params=params)
with cctx.stream_writer(buffer) as compressor:
- self.assertEqual(compressor.write(b'foo'), 0)
- self.assertEqual(compressor.write(b'bar'), 0)
- self.assertEqual(compressor.write(b'foobar' * 16384), 0)
+ self.assertEqual(compressor.write(b"foo"), 0)
+ self.assertEqual(compressor.write(b"bar"), 0)
+ self.assertEqual(compressor.write(b"foobar" * 16384), 0)
compressed = buffer.getvalue()
@@ -1091,18 +1104,18 @@
self.assertFalse(params.has_checksum)
h = hashlib.sha1(compressed).hexdigest()
- self.assertEqual(h, 'dd4bb7d37c1a0235b38a2f6b462814376843ef0b')
+ self.assertEqual(h, "dd4bb7d37c1a0235b38a2f6b462814376843ef0b")
def test_write_checksum(self):
no_checksum = NonClosingBytesIO()
cctx = zstd.ZstdCompressor(level=1)
with cctx.stream_writer(no_checksum) as compressor:
- self.assertEqual(compressor.write(b'foobar'), 0)
+ self.assertEqual(compressor.write(b"foobar"), 0)
with_checksum = NonClosingBytesIO()
cctx = zstd.ZstdCompressor(level=1, write_checksum=True)
with cctx.stream_writer(with_checksum) as compressor:
- self.assertEqual(compressor.write(b'foobar'), 0)
+ self.assertEqual(compressor.write(b"foobar"), 0)
no_params = zstd.get_frame_parameters(no_checksum.getvalue())
with_params = zstd.get_frame_parameters(with_checksum.getvalue())
@@ -1113,29 +1126,27 @@
self.assertFalse(no_params.has_checksum)
self.assertTrue(with_params.has_checksum)
- self.assertEqual(len(with_checksum.getvalue()),
- len(no_checksum.getvalue()) + 4)
+ self.assertEqual(len(with_checksum.getvalue()), len(no_checksum.getvalue()) + 4)
def test_write_content_size(self):
no_size = NonClosingBytesIO()
cctx = zstd.ZstdCompressor(level=1, write_content_size=False)
with cctx.stream_writer(no_size) as compressor:
- self.assertEqual(compressor.write(b'foobar' * 256), 0)
+ self.assertEqual(compressor.write(b"foobar" * 256), 0)
with_size = NonClosingBytesIO()
cctx = zstd.ZstdCompressor(level=1)
with cctx.stream_writer(with_size) as compressor:
- self.assertEqual(compressor.write(b'foobar' * 256), 0)
+ self.assertEqual(compressor.write(b"foobar" * 256), 0)
# Source size is not known in streaming mode, so header not
# written.
- self.assertEqual(len(with_size.getvalue()),
- len(no_size.getvalue()))
+ self.assertEqual(len(with_size.getvalue()), len(no_size.getvalue()))
# Declaring size will write the header.
with_size = NonClosingBytesIO()
- with cctx.stream_writer(with_size, size=len(b'foobar' * 256)) as compressor:
- self.assertEqual(compressor.write(b'foobar' * 256), 0)
+ with cctx.stream_writer(with_size, size=len(b"foobar" * 256)) as compressor:
+ self.assertEqual(compressor.write(b"foobar" * 256), 0)
no_params = zstd.get_frame_parameters(no_size.getvalue())
with_params = zstd.get_frame_parameters(with_size.getvalue())
@@ -1146,31 +1157,30 @@
self.assertFalse(no_params.has_checksum)
self.assertFalse(with_params.has_checksum)
- self.assertEqual(len(with_size.getvalue()),
- len(no_size.getvalue()) + 1)
+ self.assertEqual(len(with_size.getvalue()), len(no_size.getvalue()) + 1)
def test_no_dict_id(self):
samples = []
for i in range(128):
- samples.append(b'foo' * 64)
- samples.append(b'bar' * 64)
- samples.append(b'foobar' * 64)
+ samples.append(b"foo" * 64)
+ samples.append(b"bar" * 64)
+ samples.append(b"foobar" * 64)
d = zstd.train_dictionary(1024, samples)
with_dict_id = NonClosingBytesIO()
cctx = zstd.ZstdCompressor(level=1, dict_data=d)
with cctx.stream_writer(with_dict_id) as compressor:
- self.assertEqual(compressor.write(b'foobarfoobar'), 0)
+ self.assertEqual(compressor.write(b"foobarfoobar"), 0)
- self.assertEqual(with_dict_id.getvalue()[4:5], b'\x03')
+ self.assertEqual(with_dict_id.getvalue()[4:5], b"\x03")
cctx = zstd.ZstdCompressor(level=1, dict_data=d, write_dict_id=False)
no_dict_id = NonClosingBytesIO()
with cctx.stream_writer(no_dict_id) as compressor:
- self.assertEqual(compressor.write(b'foobarfoobar'), 0)
+ self.assertEqual(compressor.write(b"foobarfoobar"), 0)
- self.assertEqual(no_dict_id.getvalue()[4:5], b'\x00')
+ self.assertEqual(no_dict_id.getvalue()[4:5], b"\x00")
no_params = zstd.get_frame_parameters(no_dict_id.getvalue())
with_params = zstd.get_frame_parameters(with_dict_id.getvalue())
@@ -1181,14 +1191,13 @@
self.assertFalse(no_params.has_checksum)
self.assertFalse(with_params.has_checksum)
- self.assertEqual(len(with_dict_id.getvalue()),
- len(no_dict_id.getvalue()) + 4)
+ self.assertEqual(len(with_dict_id.getvalue()), len(no_dict_id.getvalue()) + 4)
def test_memory_size(self):
cctx = zstd.ZstdCompressor(level=3)
buffer = io.BytesIO()
with cctx.stream_writer(buffer) as compressor:
- compressor.write(b'foo')
+ compressor.write(b"foo")
size = compressor.memory_size()
self.assertGreater(size, 100000)
@@ -1197,9 +1206,9 @@
cctx = zstd.ZstdCompressor(level=3)
dest = OpCountingBytesIO()
with cctx.stream_writer(dest, write_size=1) as compressor:
- self.assertEqual(compressor.write(b'foo'), 0)
- self.assertEqual(compressor.write(b'bar'), 0)
- self.assertEqual(compressor.write(b'foobar'), 0)
+ self.assertEqual(compressor.write(b"foo"), 0)
+ self.assertEqual(compressor.write(b"bar"), 0)
+ self.assertEqual(compressor.write(b"foobar"), 0)
self.assertEqual(len(dest.getvalue()), dest._write_count)
@@ -1207,15 +1216,15 @@
cctx = zstd.ZstdCompressor(level=3)
dest = OpCountingBytesIO()
with cctx.stream_writer(dest) as compressor:
- self.assertEqual(compressor.write(b'foo'), 0)
+ self.assertEqual(compressor.write(b"foo"), 0)
self.assertEqual(dest._write_count, 0)
self.assertEqual(compressor.flush(), 12)
self.assertEqual(dest._write_count, 1)
- self.assertEqual(compressor.write(b'bar'), 0)
+ self.assertEqual(compressor.write(b"bar"), 0)
self.assertEqual(dest._write_count, 1)
self.assertEqual(compressor.flush(), 6)
self.assertEqual(dest._write_count, 2)
- self.assertEqual(compressor.write(b'baz'), 0)
+ self.assertEqual(compressor.write(b"baz"), 0)
self.assertEqual(dest._write_count, 3)
@@ -1223,7 +1232,7 @@
cctx = zstd.ZstdCompressor(level=3, write_checksum=True)
dest = OpCountingBytesIO()
with cctx.stream_writer(dest) as compressor:
- self.assertEqual(compressor.write(b'foobar' * 8192), 0)
+ self.assertEqual(compressor.write(b"foobar" * 8192), 0)
count = dest._write_count
offset = dest.tell()
self.assertEqual(compressor.flush(), 23)
@@ -1238,41 +1247,43 @@
self.assertEqual(len(trailing), 7)
header = trailing[0:3]
- self.assertEqual(header, b'\x01\x00\x00')
+ self.assertEqual(header, b"\x01\x00\x00")
def test_flush_frame(self):
cctx = zstd.ZstdCompressor(level=3)
dest = OpCountingBytesIO()
with cctx.stream_writer(dest) as compressor:
- self.assertEqual(compressor.write(b'foobar' * 8192), 0)
+ self.assertEqual(compressor.write(b"foobar" * 8192), 0)
self.assertEqual(compressor.flush(zstd.FLUSH_FRAME), 23)
- compressor.write(b'biz' * 16384)
+ compressor.write(b"biz" * 16384)
- self.assertEqual(dest.getvalue(),
- # Frame 1.
- b'\x28\xb5\x2f\xfd\x00\x58\x75\x00\x00\x30\x66\x6f\x6f'
- b'\x62\x61\x72\x01\x00\xf7\xbf\xe8\xa5\x08'
- # Frame 2.
- b'\x28\xb5\x2f\xfd\x00\x58\x5d\x00\x00\x18\x62\x69\x7a'
- b'\x01\x00\xfa\x3f\x75\x37\x04')
+ self.assertEqual(
+ dest.getvalue(),
+ # Frame 1.
+ b"\x28\xb5\x2f\xfd\x00\x58\x75\x00\x00\x30\x66\x6f\x6f"
+ b"\x62\x61\x72\x01\x00\xf7\xbf\xe8\xa5\x08"
+ # Frame 2.
+ b"\x28\xb5\x2f\xfd\x00\x58\x5d\x00\x00\x18\x62\x69\x7a"
+ b"\x01\x00\xfa\x3f\x75\x37\x04",
+ )
def test_bad_flush_mode(self):
cctx = zstd.ZstdCompressor()
dest = io.BytesIO()
with cctx.stream_writer(dest) as compressor:
- with self.assertRaisesRegexp(ValueError, 'unknown flush_mode: 42'):
+ with self.assertRaisesRegex(ValueError, "unknown flush_mode: 42"):
compressor.flush(flush_mode=42)
def test_multithreaded(self):
dest = NonClosingBytesIO()
cctx = zstd.ZstdCompressor(threads=2)
with cctx.stream_writer(dest) as compressor:
- compressor.write(b'a' * 1048576)
- compressor.write(b'b' * 1048576)
- compressor.write(b'c' * 1048576)
+ compressor.write(b"a" * 1048576)
+ compressor.write(b"b" * 1048576)
+ compressor.write(b"c" * 1048576)
- self.assertEqual(len(dest.getvalue()), 295)
+ self.assertEqual(len(dest.getvalue()), 111)
def test_tell(self):
dest = io.BytesIO()
@@ -1281,7 +1292,7 @@
self.assertEqual(compressor.tell(), 0)
for i in range(256):
- compressor.write(b'foo' * (i + 1))
+ compressor.write(b"foo" * (i + 1))
self.assertEqual(compressor.tell(), dest.tell())
def test_bad_size(self):
@@ -1289,9 +1300,9 @@
dest = io.BytesIO()
- with self.assertRaisesRegexp(zstd.ZstdError, 'Src size is incorrect'):
+ with self.assertRaisesRegex(zstd.ZstdError, "Src size is incorrect"):
with cctx.stream_writer(dest, size=2) as compressor:
- compressor.write(b'foo')
+ compressor.write(b"foo")
# Test another operation.
with cctx.stream_writer(dest, size=42):
@@ -1301,20 +1312,20 @@
dest = NonClosingBytesIO()
cctx = zstd.ZstdCompressor()
with cctx.stream_writer(dest) as compressor:
- with tarfile.open('tf', mode='w|', fileobj=compressor) as tf:
- tf.add(__file__, 'test_compressor.py')
+ with tarfile.open("tf", mode="w|", fileobj=compressor) as tf:
+ tf.add(__file__, "test_compressor.py")
dest = io.BytesIO(dest.getvalue())
dctx = zstd.ZstdDecompressor()
with dctx.stream_reader(dest) as reader:
- with tarfile.open(mode='r|', fileobj=reader) as tf:
+ with tarfile.open(mode="r|", fileobj=reader) as tf:
for member in tf:
- self.assertEqual(member.name, 'test_compressor.py')
+ self.assertEqual(member.name, "test_compressor.py")
@make_cffi
-class TestCompressor_read_to_iter(unittest.TestCase):
+class TestCompressor_read_to_iter(TestCase):
def test_type_validation(self):
cctx = zstd.ZstdCompressor()
@@ -1323,10 +1334,10 @@
pass
# Buffer protocol works.
- for chunk in cctx.read_to_iter(b'foobar'):
+ for chunk in cctx.read_to_iter(b"foobar"):
pass
- with self.assertRaisesRegexp(ValueError, 'must pass an object with a read'):
+ with self.assertRaisesRegex(ValueError, "must pass an object with a read"):
for chunk in cctx.read_to_iter(True):
pass
@@ -1337,22 +1348,22 @@
it = cctx.read_to_iter(source)
chunks = list(it)
self.assertEqual(len(chunks), 1)
- compressed = b''.join(chunks)
- self.assertEqual(compressed, b'\x28\xb5\x2f\xfd\x00\x48\x01\x00\x00')
+ compressed = b"".join(chunks)
+ self.assertEqual(compressed, b"\x28\xb5\x2f\xfd\x00\x48\x01\x00\x00")
# And again with the buffer protocol.
- it = cctx.read_to_iter(b'')
+ it = cctx.read_to_iter(b"")
chunks = list(it)
self.assertEqual(len(chunks), 1)
- compressed2 = b''.join(chunks)
+ compressed2 = b"".join(chunks)
self.assertEqual(compressed2, compressed)
def test_read_large(self):
cctx = zstd.ZstdCompressor(level=1, write_content_size=False)
source = io.BytesIO()
- source.write(b'f' * zstd.COMPRESSION_RECOMMENDED_INPUT_SIZE)
- source.write(b'o')
+ source.write(b"f" * zstd.COMPRESSION_RECOMMENDED_INPUT_SIZE)
+ source.write(b"o")
source.seek(0)
# Creating an iterator should not perform any compression until
@@ -1380,9 +1391,9 @@
next(it)
# We should get the same output as the one-shot compression mechanism.
- self.assertEqual(b''.join(chunks), cctx.compress(source.getvalue()))
+ self.assertEqual(b"".join(chunks), cctx.compress(source.getvalue()))
- params = zstd.get_frame_parameters(b''.join(chunks))
+ params = zstd.get_frame_parameters(b"".join(chunks))
self.assertEqual(params.content_size, zstd.CONTENTSIZE_UNKNOWN)
self.assertEqual(params.window_size, 262144)
self.assertEqual(params.dict_id, 0)
@@ -1393,16 +1404,16 @@
chunks = list(it)
self.assertEqual(len(chunks), 2)
- params = zstd.get_frame_parameters(b''.join(chunks))
+ params = zstd.get_frame_parameters(b"".join(chunks))
self.assertEqual(params.content_size, zstd.CONTENTSIZE_UNKNOWN)
- #self.assertEqual(params.window_size, 262144)
+ # self.assertEqual(params.window_size, 262144)
self.assertEqual(params.dict_id, 0)
self.assertFalse(params.has_checksum)
- self.assertEqual(b''.join(chunks), cctx.compress(source.getvalue()))
+ self.assertEqual(b"".join(chunks), cctx.compress(source.getvalue()))
def test_read_write_size(self):
- source = OpCountingBytesIO(b'foobarfoobar')
+ source = OpCountingBytesIO(b"foobarfoobar")
cctx = zstd.ZstdCompressor(level=3)
for chunk in cctx.read_to_iter(source, read_size=1, write_size=1):
self.assertEqual(len(chunk), 1)
@@ -1411,42 +1422,42 @@
def test_multithreaded(self):
source = io.BytesIO()
- source.write(b'a' * 1048576)
- source.write(b'b' * 1048576)
- source.write(b'c' * 1048576)
+ source.write(b"a" * 1048576)
+ source.write(b"b" * 1048576)
+ source.write(b"c" * 1048576)
source.seek(0)
cctx = zstd.ZstdCompressor(threads=2)
- compressed = b''.join(cctx.read_to_iter(source))
- self.assertEqual(len(compressed), 295)
+ compressed = b"".join(cctx.read_to_iter(source))
+ self.assertEqual(len(compressed), 111)
def test_bad_size(self):
cctx = zstd.ZstdCompressor()
- source = io.BytesIO(b'a' * 42)
+ source = io.BytesIO(b"a" * 42)
- with self.assertRaisesRegexp(zstd.ZstdError, 'Src size is incorrect'):
- b''.join(cctx.read_to_iter(source, size=2))
+ with self.assertRaisesRegex(zstd.ZstdError, "Src size is incorrect"):
+ b"".join(cctx.read_to_iter(source, size=2))
# Test another operation on errored compressor.
- b''.join(cctx.read_to_iter(source))
+ b"".join(cctx.read_to_iter(source))
@make_cffi
-class TestCompressor_chunker(unittest.TestCase):
+class TestCompressor_chunker(TestCase):
def test_empty(self):
cctx = zstd.ZstdCompressor(write_content_size=False)
chunker = cctx.chunker()
- it = chunker.compress(b'')
+ it = chunker.compress(b"")
with self.assertRaises(StopIteration):
next(it)
it = chunker.finish()
- self.assertEqual(next(it), b'\x28\xb5\x2f\xfd\x00\x58\x01\x00\x00')
+ self.assertEqual(next(it), b"\x28\xb5\x2f\xfd\x00\x58\x01\x00\x00")
with self.assertRaises(StopIteration):
next(it)
@@ -1455,21 +1466,23 @@
cctx = zstd.ZstdCompressor()
chunker = cctx.chunker()
- it = chunker.compress(b'foobar')
+ it = chunker.compress(b"foobar")
with self.assertRaises(StopIteration):
next(it)
- it = chunker.compress(b'baz' * 30)
+ it = chunker.compress(b"baz" * 30)
with self.assertRaises(StopIteration):
next(it)
it = chunker.finish()
- self.assertEqual(next(it),
- b'\x28\xb5\x2f\xfd\x00\x58\x7d\x00\x00\x48\x66\x6f'
- b'\x6f\x62\x61\x72\x62\x61\x7a\x01\x00\xe4\xe4\x8e')
+ self.assertEqual(
+ next(it),
+ b"\x28\xb5\x2f\xfd\x00\x58\x7d\x00\x00\x48\x66\x6f"
+ b"\x6f\x62\x61\x72\x62\x61\x7a\x01\x00\xe4\xe4\x8e",
+ )
with self.assertRaises(StopIteration):
next(it)
@@ -1478,57 +1491,60 @@
cctx = zstd.ZstdCompressor()
chunker = cctx.chunker(size=1024)
- it = chunker.compress(b'x' * 1000)
+ it = chunker.compress(b"x" * 1000)
with self.assertRaises(StopIteration):
next(it)
- it = chunker.compress(b'y' * 24)
+ it = chunker.compress(b"y" * 24)
with self.assertRaises(StopIteration):
next(it)
chunks = list(chunker.finish())
- self.assertEqual(chunks, [
- b'\x28\xb5\x2f\xfd\x60\x00\x03\x65\x00\x00\x18\x78\x78\x79\x02\x00'
- b'\xa0\x16\xe3\x2b\x80\x05'
- ])
+ self.assertEqual(
+ chunks,
+ [
+ b"\x28\xb5\x2f\xfd\x60\x00\x03\x65\x00\x00\x18\x78\x78\x79\x02\x00"
+ b"\xa0\x16\xe3\x2b\x80\x05"
+ ],
+ )
dctx = zstd.ZstdDecompressor()
- self.assertEqual(dctx.decompress(b''.join(chunks)),
- (b'x' * 1000) + (b'y' * 24))
+ self.assertEqual(dctx.decompress(b"".join(chunks)), (b"x" * 1000) + (b"y" * 24))
def test_small_chunk_size(self):
cctx = zstd.ZstdCompressor()
chunker = cctx.chunker(chunk_size=1)
- chunks = list(chunker.compress(b'foo' * 1024))
+ chunks = list(chunker.compress(b"foo" * 1024))
self.assertEqual(chunks, [])
chunks = list(chunker.finish())
self.assertTrue(all(len(chunk) == 1 for chunk in chunks))
self.assertEqual(
- b''.join(chunks),
- b'\x28\xb5\x2f\xfd\x00\x58\x55\x00\x00\x18\x66\x6f\x6f\x01\x00'
- b'\xfa\xd3\x77\x43')
+ b"".join(chunks),
+ b"\x28\xb5\x2f\xfd\x00\x58\x55\x00\x00\x18\x66\x6f\x6f\x01\x00"
+ b"\xfa\xd3\x77\x43",
+ )
dctx = zstd.ZstdDecompressor()
- self.assertEqual(dctx.decompress(b''.join(chunks),
- max_output_size=10000),
- b'foo' * 1024)
+ self.assertEqual(
+ dctx.decompress(b"".join(chunks), max_output_size=10000), b"foo" * 1024
+ )
def test_input_types(self):
cctx = zstd.ZstdCompressor()
mutable_array = bytearray(3)
- mutable_array[:] = b'foo'
+ mutable_array[:] = b"foo"
sources = [
- memoryview(b'foo'),
- bytearray(b'foo'),
+ memoryview(b"foo"),
+ bytearray(b"foo"),
mutable_array,
]
@@ -1536,28 +1552,32 @@
chunker = cctx.chunker()
self.assertEqual(list(chunker.compress(source)), [])
- self.assertEqual(list(chunker.finish()), [
- b'\x28\xb5\x2f\xfd\x00\x58\x19\x00\x00\x66\x6f\x6f'
- ])
+ self.assertEqual(
+ list(chunker.finish()),
+ [b"\x28\xb5\x2f\xfd\x00\x58\x19\x00\x00\x66\x6f\x6f"],
+ )
def test_flush(self):
cctx = zstd.ZstdCompressor()
chunker = cctx.chunker()
- self.assertEqual(list(chunker.compress(b'foo' * 1024)), [])
- self.assertEqual(list(chunker.compress(b'bar' * 1024)), [])
+ self.assertEqual(list(chunker.compress(b"foo" * 1024)), [])
+ self.assertEqual(list(chunker.compress(b"bar" * 1024)), [])
chunks1 = list(chunker.flush())
- self.assertEqual(chunks1, [
- b'\x28\xb5\x2f\xfd\x00\x58\x8c\x00\x00\x30\x66\x6f\x6f\x62\x61\x72'
- b'\x02\x00\xfa\x03\xfe\xd0\x9f\xbe\x1b\x02'
- ])
+ self.assertEqual(
+ chunks1,
+ [
+ b"\x28\xb5\x2f\xfd\x00\x58\x8c\x00\x00\x30\x66\x6f\x6f\x62\x61\x72"
+ b"\x02\x00\xfa\x03\xfe\xd0\x9f\xbe\x1b\x02"
+ ],
+ )
self.assertEqual(list(chunker.flush()), [])
self.assertEqual(list(chunker.flush()), [])
- self.assertEqual(list(chunker.compress(b'baz' * 1024)), [])
+ self.assertEqual(list(chunker.compress(b"baz" * 1024)), [])
chunks2 = list(chunker.flush())
self.assertEqual(len(chunks2), 1)
@@ -1567,53 +1587,56 @@
dctx = zstd.ZstdDecompressor()
- self.assertEqual(dctx.decompress(b''.join(chunks1 + chunks2 + chunks3),
- max_output_size=10000),
- (b'foo' * 1024) + (b'bar' * 1024) + (b'baz' * 1024))
+ self.assertEqual(
+ dctx.decompress(
+ b"".join(chunks1 + chunks2 + chunks3), max_output_size=10000
+ ),
+ (b"foo" * 1024) + (b"bar" * 1024) + (b"baz" * 1024),
+ )
def test_compress_after_finish(self):
cctx = zstd.ZstdCompressor()
chunker = cctx.chunker()
- list(chunker.compress(b'foo'))
+ list(chunker.compress(b"foo"))
list(chunker.finish())
- with self.assertRaisesRegexp(
- zstd.ZstdError,
- r'cannot call compress\(\) after compression finished'):
- list(chunker.compress(b'foo'))
+ with self.assertRaisesRegex(
+ zstd.ZstdError, r"cannot call compress\(\) after compression finished"
+ ):
+ list(chunker.compress(b"foo"))
def test_flush_after_finish(self):
cctx = zstd.ZstdCompressor()
chunker = cctx.chunker()
- list(chunker.compress(b'foo'))
+ list(chunker.compress(b"foo"))
list(chunker.finish())
- with self.assertRaisesRegexp(
- zstd.ZstdError,
- r'cannot call flush\(\) after compression finished'):
+ with self.assertRaisesRegex(
+ zstd.ZstdError, r"cannot call flush\(\) after compression finished"
+ ):
list(chunker.flush())
def test_finish_after_finish(self):
cctx = zstd.ZstdCompressor()
chunker = cctx.chunker()
- list(chunker.compress(b'foo'))
+ list(chunker.compress(b"foo"))
list(chunker.finish())
- with self.assertRaisesRegexp(
- zstd.ZstdError,
- r'cannot call finish\(\) after compression finished'):
+ with self.assertRaisesRegex(
+ zstd.ZstdError, r"cannot call finish\(\) after compression finished"
+ ):
list(chunker.finish())
-class TestCompressor_multi_compress_to_buffer(unittest.TestCase):
+class TestCompressor_multi_compress_to_buffer(TestCase):
def test_invalid_inputs(self):
cctx = zstd.ZstdCompressor()
- if not hasattr(cctx, 'multi_compress_to_buffer'):
- self.skipTest('multi_compress_to_buffer not available')
+ if not hasattr(cctx, "multi_compress_to_buffer"):
+ self.skipTest("multi_compress_to_buffer not available")
with self.assertRaises(TypeError):
cctx.multi_compress_to_buffer(True)
@@ -1621,28 +1644,28 @@
with self.assertRaises(TypeError):
cctx.multi_compress_to_buffer((1, 2))
- with self.assertRaisesRegexp(TypeError, 'item 0 not a bytes like object'):
- cctx.multi_compress_to_buffer([u'foo'])
+ with self.assertRaisesRegex(TypeError, "item 0 not a bytes like object"):
+ cctx.multi_compress_to_buffer([u"foo"])
def test_empty_input(self):
cctx = zstd.ZstdCompressor()
- if not hasattr(cctx, 'multi_compress_to_buffer'):
- self.skipTest('multi_compress_to_buffer not available')
+ if not hasattr(cctx, "multi_compress_to_buffer"):
+ self.skipTest("multi_compress_to_buffer not available")
- with self.assertRaisesRegexp(ValueError, 'no source elements found'):
+ with self.assertRaisesRegex(ValueError, "no source elements found"):
cctx.multi_compress_to_buffer([])
- with self.assertRaisesRegexp(ValueError, 'source elements are empty'):
- cctx.multi_compress_to_buffer([b'', b'', b''])
+ with self.assertRaisesRegex(ValueError, "source elements are empty"):
+ cctx.multi_compress_to_buffer([b"", b"", b""])
def test_list_input(self):
cctx = zstd.ZstdCompressor(write_checksum=True)
- if not hasattr(cctx, 'multi_compress_to_buffer'):
- self.skipTest('multi_compress_to_buffer not available')
+ if not hasattr(cctx, "multi_compress_to_buffer"):
+ self.skipTest("multi_compress_to_buffer not available")
- original = [b'foo' * 12, b'bar' * 6]
+ original = [b"foo" * 12, b"bar" * 6]
frames = [cctx.compress(c) for c in original]
b = cctx.multi_compress_to_buffer(original)
@@ -1657,15 +1680,16 @@
def test_buffer_with_segments_input(self):
cctx = zstd.ZstdCompressor(write_checksum=True)
- if not hasattr(cctx, 'multi_compress_to_buffer'):
- self.skipTest('multi_compress_to_buffer not available')
+ if not hasattr(cctx, "multi_compress_to_buffer"):
+ self.skipTest("multi_compress_to_buffer not available")
- original = [b'foo' * 4, b'bar' * 6]
+ original = [b"foo" * 4, b"bar" * 6]
frames = [cctx.compress(c) for c in original]
- offsets = struct.pack('=QQQQ', 0, len(original[0]),
- len(original[0]), len(original[1]))
- segments = zstd.BufferWithSegments(b''.join(original), offsets)
+ offsets = struct.pack(
+ "=QQQQ", 0, len(original[0]), len(original[0]), len(original[1])
+ )
+ segments = zstd.BufferWithSegments(b"".join(original), offsets)
result = cctx.multi_compress_to_buffer(segments)
@@ -1678,28 +1702,39 @@
def test_buffer_with_segments_collection_input(self):
cctx = zstd.ZstdCompressor(write_checksum=True)
- if not hasattr(cctx, 'multi_compress_to_buffer'):
- self.skipTest('multi_compress_to_buffer not available')
+ if not hasattr(cctx, "multi_compress_to_buffer"):
+ self.skipTest("multi_compress_to_buffer not available")
original = [
- b'foo1',
- b'foo2' * 2,
- b'foo3' * 3,
- b'foo4' * 4,
- b'foo5' * 5,
+ b"foo1",
+ b"foo2" * 2,
+ b"foo3" * 3,
+ b"foo4" * 4,
+ b"foo5" * 5,
]
frames = [cctx.compress(c) for c in original]
- b = b''.join([original[0], original[1]])
- b1 = zstd.BufferWithSegments(b, struct.pack('=QQQQ',
- 0, len(original[0]),
- len(original[0]), len(original[1])))
- b = b''.join([original[2], original[3], original[4]])
- b2 = zstd.BufferWithSegments(b, struct.pack('=QQQQQQ',
- 0, len(original[2]),
- len(original[2]), len(original[3]),
- len(original[2]) + len(original[3]), len(original[4])))
+ b = b"".join([original[0], original[1]])
+ b1 = zstd.BufferWithSegments(
+ b,
+ struct.pack(
+ "=QQQQ", 0, len(original[0]), len(original[0]), len(original[1])
+ ),
+ )
+ b = b"".join([original[2], original[3], original[4]])
+ b2 = zstd.BufferWithSegments(
+ b,
+ struct.pack(
+ "=QQQQQQ",
+ 0,
+ len(original[2]),
+ len(original[2]),
+ len(original[3]),
+ len(original[2]) + len(original[3]),
+ len(original[4]),
+ ),
+ )
c = zstd.BufferWithSegmentsCollection(b1, b2)
@@ -1714,16 +1749,16 @@
# threads argument will cause multi-threaded ZSTD APIs to be used, which will
# make output different.
refcctx = zstd.ZstdCompressor(write_checksum=True)
- reference = [refcctx.compress(b'x' * 64), refcctx.compress(b'y' * 64)]
+ reference = [refcctx.compress(b"x" * 64), refcctx.compress(b"y" * 64)]
cctx = zstd.ZstdCompressor(write_checksum=True)
- if not hasattr(cctx, 'multi_compress_to_buffer'):
- self.skipTest('multi_compress_to_buffer not available')
+ if not hasattr(cctx, "multi_compress_to_buffer"):
+ self.skipTest("multi_compress_to_buffer not available")
frames = []
- frames.extend(b'x' * 64 for i in range(256))
- frames.extend(b'y' * 64 for i in range(256))
+ frames.extend(b"x" * 64 for i in range(256))
+ frames.extend(b"y" * 64 for i in range(256))
result = cctx.multi_compress_to_buffer(frames, threads=-1)