|
|
|
|
|
from __future__ import absolute_import |
|
|
from __future__ import print_function |
|
|
from __future__ import unicode_literals |
|
|
from io import BytesIO |
|
|
from unittest import TestCase |
|
|
import os |
|
|
import zlib |
|
|
|
|
|
from hwp5.compressed import ZLibIncrementalDecoder |
|
|
from hwp5.compressed import decompress |
|
|
from hwp5.compressed import decompress_gen |
|
|
from hwp5.utils import cached_property |
|
|
|
|
|
|
|
|
class TestUncompress(TestCase): |
|
|
|
|
|
@cached_property |
|
|
def original_data(self): |
|
|
return os.urandom(16384) |
|
|
|
|
|
@cached_property |
|
|
def compressed_data(self): |
|
|
return zlib.compress(self.original_data) |
|
|
|
|
|
def test_incremental_decode(self): |
|
|
compressed_data = self.compressed_data |
|
|
|
|
|
dec = ZLibIncrementalDecoder(wbits=-15) |
|
|
data = dec.decode(compressed_data[2:2048]) |
|
|
data += dec.decode(compressed_data[2048:2048 + 1024]) |
|
|
data += dec.decode(compressed_data[2048 + 1024:2048 + 1024 + 4096]) |
|
|
data += dec.decode(compressed_data[2048 + 1024 + 4096:], True) |
|
|
|
|
|
self.assertEqual(self.original_data, data) |
|
|
|
|
|
def test_decompress(self): |
|
|
|
|
|
gen = decompress_gen(BytesIO(self.compressed_data[2:])) |
|
|
self.assertEqual(self.original_data, b''.join(gen)) |
|
|
|
|
|
|
|
|
|
|
|
f = decompress(BytesIO(self.compressed_data[2:])) |
|
|
g = BytesIO(self.original_data) |
|
|
|
|
|
self.assertEqual(f.read(2048), g.read(2048)) |
|
|
self.assertEqual(f.read(1024), g.read(1024)) |
|
|
self.assertEqual(f.read(4096), g.read(4096)) |
|
|
self.assertEqual(f.read(), g.read()) |
|
|
|
