mercurial-scm/hg: contrib/python-zstandard/tests/test

comparison contrib/python-zstandard/tests/test_decompressor.py @ 44147:5e84a96d865b

python-zstandard: blacken at 80 characters I made this change upstream and it will make it into the next release of python-zstandard. I figured I'd send it Mercurial's way because it will allow us to drop this directory from the black exclusion list. # skip-blame blackening Differential Revision: https://phab.mercurial-scm.org/D7937

author	Gregory Szorc <gregory.szorc@gmail.com>
date	Wed, 22 Jan 2020 22:23:04 -0800
parents	de7838053207
children	493034cc3265

comparison

equal deleted inserted replaced

-:45ec64d93b3a
+:5e84a96d865b
 zstd.ZstdError, "decompression error: did not decompress full frame"
 ):
 dctx.decompress(compressed, max_output_size=len(source) - 1)
 # Input size + 1 works
-decompressed = dctx.decompress(compressed, max_output_size=len(source) + 1)
+decompressed = dctx.decompress(
+compressed, max_output_size=len(source) + 1
+)
 self.assertEqual(decompressed, source)
 # A much larger buffer works.
-decompressed = dctx.decompress(compressed, max_output_size=len(source) * 64)
+decompressed = dctx.decompress(
+compressed, max_output_size=len(source) * 64
+)
 self.assertEqual(decompressed, source)
 def test_stupidly_large_output_buffer(self):
 cctx = zstd.ZstdCompressor(write_content_size=False)
 compressed = cctx.compress(b"foobar" * 256)
 frame = cctx.compress(source)
 dctx = zstd.ZstdDecompressor(max_window_size=2 ** zstd.WINDOWLOG_MIN)
 with self.assertRaisesRegex(
-zstd.ZstdError, "decompression error: Frame requires too much memory"
+zstd.ZstdError,
+"decompression error: Frame requires too much memory",
 ):
 dctx.decompress(frame, max_output_size=len(source))
 @make_cffi
 self.assertEqual(r, len(compressed.getvalue()))
 self.assertEqual(w, len(source.getvalue()))
 def test_read_write_size(self):
-source = OpCountingBytesIO(zstd.ZstdCompressor().compress(b"foobarfoobar"))
+source = OpCountingBytesIO(
+zstd.ZstdCompressor().compress(b"foobarfoobar")
+)
 dest = OpCountingBytesIO()
 dctx = zstd.ZstdDecompressor()
 r, w = dctx.copy_stream(source, dest, read_size=1, write_size=1)
 class TestDecompressor_stream_reader(TestCase):
 def test_context_manager(self):
 dctx = zstd.ZstdDecompressor()
 with dctx.stream_reader(b"foo") as reader:
-with self.assertRaisesRegex(ValueError, "cannot __enter__ multiple times"):
+with self.assertRaisesRegex(
+ValueError, "cannot __enter__ multiple times"
+):
 with reader as reader2:
 pass
 def test_not_implemented(self):
 dctx = zstd.ZstdDecompressor()
 frame = cctx.compress(b"foo" * 60)
 dctx = zstd.ZstdDecompressor()
 with dctx.stream_reader(frame) as reader:
-with self.assertRaisesRegex(ValueError, "cannot seek to negative position"):
+with self.assertRaisesRegex(
+ValueError, "cannot seek to negative position"
+):
 reader.seek(-1, os.SEEK_SET)
 reader.read(1)
 with self.assertRaisesRegex(
 ValueError, "cannot seek zstd decompression stream backwards"
 ):
 reader.seek(-1, os.SEEK_CUR)
 with self.assertRaisesRegex(
-ValueError, "zstd decompression streams cannot be seeked with SEEK_END"
+ValueError,
+"zstd decompression streams cannot be seeked with SEEK_END",
 ):
 reader.seek(0, os.SEEK_END)
 reader.close()
 self.assertEqual(reader.read1(1), b"")
 self.assertEqual(b._read_count, 2)
 def test_read_lines(self):
 cctx = zstd.ZstdCompressor()
-source = b"\n".join(("line %d" % i).encode("ascii") for i in range(1024))
+source = b"\n".join(
+("line %d" % i).encode("ascii") for i in range(1024)
+)
 frame = cctx.compress(source)
 dctx = zstd.ZstdDecompressor()
 reader = dctx.stream_reader(frame)
 dctx = zstd.ZstdDecompressor()
 dobj = dctx.decompressobj()
 dobj.decompress(data)
-with self.assertRaisesRegex(zstd.ZstdError, "cannot use a decompressobj"):
+with self.assertRaisesRegex(
+zstd.ZstdError, "cannot use a decompressobj"
+):
 dobj.decompress(data)
 self.assertIsNone(dobj.flush())
 def test_bad_write_size(self):
 dctx = zstd.ZstdDecompressor()
 dctx.read_to_iter(io.BytesIO())
 # Buffer protocol works.
 dctx.read_to_iter(b"foobar")
-with self.assertRaisesRegex(ValueError, "must pass an object with a read"):
+with self.assertRaisesRegex(
+ValueError, "must pass an object with a read"
+):
 b"".join(dctx.read_to_iter(True))
 def test_empty_input(self):
 dctx = zstd.ZstdDecompressor()
 next(it)
 decompressed = b"".join(chunks)
 self.assertEqual(decompressed, source.getvalue())
-@unittest.skipUnless("ZSTD_SLOW_TESTS" in os.environ, "ZSTD_SLOW_TESTS not set")
+@unittest.skipUnless(
+"ZSTD_SLOW_TESTS" in os.environ, "ZSTD_SLOW_TESTS not set"
+)
 def test_large_input(self):
 bytes = list(struct.Struct(">B").pack(i) for i in range(256))
 compressed = NonClosingBytesIO()
 input_size = 0
 cctx = zstd.ZstdCompressor(level=1)
 have_compressed = (
 len(compressed.getvalue())
 > zstd.DECOMPRESSION_RECOMMENDED_INPUT_SIZE
 )
-have_raw = input_size > zstd.DECOMPRESSION_RECOMMENDED_OUTPUT_SIZE * 2
+have_raw = (
+input_size > zstd.DECOMPRESSION_RECOMMENDED_OUTPUT_SIZE * 2
+)
 if have_compressed and have_raw:
 break
 compressed = io.BytesIO(compressed.getvalue())
 self.assertGreater(
-len(compressed.getvalue()), zstd.DECOMPRESSION_RECOMMENDED_INPUT_SIZE
+len(compressed.getvalue()),
+zstd.DECOMPRESSION_RECOMMENDED_INPUT_SIZE,
 )
 dctx = zstd.ZstdDecompressor()
 it = dctx.read_to_iter(compressed)
 compressed = io.BytesIO(compressed.getvalue())
 streamed = b"".join(dctx.read_to_iter(compressed))
 self.assertEqual(streamed, source.getvalue())
 def test_read_write_size(self):
-source = OpCountingBytesIO(zstd.ZstdCompressor().compress(b"foobarfoobar"))
+source = OpCountingBytesIO(
+zstd.ZstdCompressor().compress(b"foobarfoobar")
+)
 dctx = zstd.ZstdDecompressor()
 for chunk in dctx.read_to_iter(source, read_size=1, write_size=1):
 self.assertEqual(len(chunk), 1)
 self.assertEqual(source._read_count, len(source.getvalue()))
 with self.assertRaisesRegex(
 ValueError, "chunk 0 is too small to contain a zstd frame"
 ):
 dctx.decompress_content_dict_chain([zstd.FRAME_HEADER])
-with self.assertRaisesRegex(ValueError, "chunk 0 is not a valid zstd frame"):
+with self.assertRaisesRegex(
+ValueError, "chunk 0 is not a valid zstd frame"
+):
 dctx.decompress_content_dict_chain([b"foo" * 8])
-no_size = zstd.ZstdCompressor(write_content_size=False).compress(b"foo" * 64)
+no_size = zstd.ZstdCompressor(write_content_size=False).compress(
+b"foo" * 64
+)
 with self.assertRaisesRegex(
 ValueError, "chunk 0 missing content size in frame"
 ):
 dctx.decompress_content_dict_chain([no_size])
 with self.assertRaisesRegex(
 ValueError, "chunk 1 is too small to contain a zstd frame"
 ):
 dctx.decompress_content_dict_chain([initial, zstd.FRAME_HEADER])
-with self.assertRaisesRegex(ValueError, "chunk 1 is not a valid zstd frame"):
+with self.assertRaisesRegex(
+ValueError, "chunk 1 is not a valid zstd frame"
+):
 dctx.decompress_content_dict_chain([initial, b"foo" * 8])
-no_size = zstd.ZstdCompressor(write_content_size=False).compress(b"foo" * 64)
+no_size = zstd.ZstdCompressor(write_content_size=False).compress(
+b"foo" * 64
+)
 with self.assertRaisesRegex(
 ValueError, "chunk 1 missing content size in frame"
 ):
 dctx.decompress_content_dict_chain([initial, no_size])
 # Corrupt second frame.
-cctx = zstd.ZstdCompressor(dict_data=zstd.ZstdCompressionDict(b"foo" * 64))
+cctx = zstd.ZstdCompressor(
+dict_data=zstd.ZstdCompressionDict(b"foo" * 64)
+)
 frame = cctx.compress(b"bar" * 64)
 frame = frame[0:12] + frame[15:]
 with self.assertRaisesRegex(
 zstd.ZstdError, "chunk 1 did not decompress full frame"
 dctx.multi_decompress_to_buffer(True)
 with self.assertRaises(TypeError):
 dctx.multi_decompress_to_buffer((1, 2))
-with self.assertRaisesRegex(TypeError, "item 0 not a bytes like object"):
+with self.assertRaisesRegex(
+TypeError, "item 0 not a bytes like object"
+):
 dctx.multi_decompress_to_buffer([u"foo"])
 with self.assertRaisesRegex(
 ValueError, "could not determine decompressed size of item 0"
 ):
 dctx = zstd.ZstdDecompressor()
 if not hasattr(dctx, "multi_decompress_to_buffer"):
 self.skipTest("multi_decompress_to_buffer not available")
-result = dctx.multi_decompress_to_buffer(frames, decompressed_sizes=sizes)
+result = dctx.multi_decompress_to_buffer(
+frames, decompressed_sizes=sizes
+)
 self.assertEqual(len(result), len(frames))
 self.assertEqual(result.size(), sum(map(len, original)))
 for i, data in enumerate(original):
 self.assertEqual(data, decompressed[i].tobytes())
 # And a manual mode.
 b = b"".join([frames[0].tobytes(), frames[1].tobytes()])
 b1 = zstd.BufferWithSegments(
-b, struct.pack("=QQQQ", 0, len(frames[0]), len(frames[0]), len(frames[1]))
+b,
+struct.pack(
+"=QQQQ", 0, len(frames[0]), len(frames[0]), len(frames[1])
+),
 )
-b = b"".join([frames[2].tobytes(), frames[3].tobytes(), frames[4].tobytes()])
+b = b"".join(
+[frames[2].tobytes(), frames[3].tobytes(), frames[4].tobytes()]
+)
 b2 = zstd.BufferWithSegments(
 b,
 struct.pack(
 "=QQQQQQ",
 0,

Mercurial > public > mercurial-scm > hg

comparison contrib/python-zstandard/tests/test_decompressor.py @ 44147:5e84a96d865b