diff options
Diffstat (limited to 'python/tests/compressor_test.py')
| -rw-r--r-- | python/tests/compressor_test.py | 104 |
1 files changed, 27 insertions, 77 deletions
diff --git a/python/tests/compressor_test.py b/python/tests/compressor_test.py index 6983c31..5924e4c 100644 --- a/python/tests/compressor_test.py +++ b/python/tests/compressor_test.py @@ -3,135 +3,85 @@ # Distributed under MIT license. # See file LICENSE for detail or copy at https://opensource.org/licenses/MIT -import filecmp import functools -import os -import sys -import types import unittest -import test_utils - +import _test_utils import brotli -TEST_DATA_FILES = [ - 'empty', # Empty file - '10x10y', # Small text - 'alice29.txt', # Large text - 'random_org_10k.bin', # Small data - 'mapsdatazrh', # Large data -] -TEST_DATA_PATHS = [os.path.join(test_utils.TESTDATA_DIR, f) - for f in TEST_DATA_FILES] - - -def get_compressed_name(filename): - return filename + '.compressed' - - -def get_temp_compressed_name(filename): - return filename + '.bro' - - -def get_temp_uncompressed_name(filename): - return filename + '.unbro' - - -def bind_method_args(method, *args): - return lambda self: method(self, *args) - - -# Do not inherit from unittest.TestCase here to ensure that test methods +# Do not inherit from TestCase here to ensure that test methods # are not run automatically and instead are run as part of a specific # configuration below. class _TestCompressor(object): - def _check_decompression_matches(self, test_data): + CHUNK_SIZE = 2048 + + def _check_decompression(self, test_data): # Write decompression to temp file and verify it matches the original. - with open(get_temp_uncompressed_name(test_data), 'wb') as out_file: - with open(get_temp_compressed_name(test_data), 'rb') as in_file: + temp_uncompressed = _test_utils.get_temp_uncompressed_name(test_data) + temp_compressed = _test_utils.get_temp_compressed_name(test_data) + original = test_data + with open(temp_uncompressed, 'wb') as out_file: + with open(temp_compressed, 'rb') as in_file: out_file.write(brotli.decompress(in_file.read())) - self.assertTrue( - filecmp.cmp(get_temp_uncompressed_name(test_data), - test_data, - shallow=False)) + self.assertFilesMatch(temp_uncompressed, original) def _test_single_process(self, test_data): # Write single-shot compression to temp file. - with open(get_temp_compressed_name(test_data), 'wb') as out_file: + temp_compressed = _test_utils.get_temp_compressed_name(test_data) + with open(temp_compressed, 'wb') as out_file: with open(test_data, 'rb') as in_file: out_file.write(self.compressor.process(in_file.read())) out_file.write(self.compressor.finish()) - self._check_decompression_matches(test_data) + self._check_decompression(test_data) def _test_multiple_process(self, test_data): # Write chunked compression to temp file. - chunk_size = 2048 - with open(get_temp_compressed_name(test_data), 'wb') as out_file: + temp_compressed = _test_utils.get_temp_compressed_name(test_data) + with open(temp_compressed, 'wb') as out_file: with open(test_data, 'rb') as in_file: - read_chunk = functools.partial(in_file.read, chunk_size) + read_chunk = functools.partial(in_file.read, self.CHUNK_SIZE) for data in iter(read_chunk, b''): out_file.write(self.compressor.process(data)) out_file.write(self.compressor.finish()) - self._check_decompression_matches(test_data) + self._check_decompression(test_data) def _test_multiple_process_and_flush(self, test_data): # Write chunked and flushed compression to temp file. - chunk_size = 2048 - with open(get_temp_compressed_name(test_data), 'wb') as out_file: + temp_compressed = _test_utils.get_temp_compressed_name(test_data) + with open(temp_compressed, 'wb') as out_file: with open(test_data, 'rb') as in_file: - read_chunk = functools.partial(in_file.read, chunk_size) + read_chunk = functools.partial(in_file.read, self.CHUNK_SIZE) for data in iter(read_chunk, b''): out_file.write(self.compressor.process(data)) out_file.write(self.compressor.flush()) out_file.write(self.compressor.finish()) - self._check_decompression_matches(test_data) - - -# Add test methods for each test data file. This makes identifying problems -# with specific compression scenarios easier. -for methodname in [m for m in dir(_TestCompressor) if m.startswith('_test')]: - for test_data in TEST_DATA_PATHS: - filename = os.path.splitext(os.path.basename(test_data))[0] - name = 'test_{method}_{file}'.format(method=methodname, file=filename) - func = bind_method_args(getattr(_TestCompressor, methodname), test_data) - setattr(_TestCompressor, name, func) - + self._check_decompression(test_data) -class _CompressionTestCase(unittest.TestCase): - def tearDown(self): - for f in TEST_DATA_PATHS: - try: - os.unlink(get_temp_compressed_name(f)) - except OSError: - pass - try: - os.unlink(get_temp_uncompressed_name(f)) - except OSError: - pass +_test_utils.generate_test_methods(_TestCompressor) -class TestCompressorQuality1(_TestCompressor, _CompressionTestCase): +class TestCompressorQuality1(_TestCompressor, _test_utils.TestCase): def setUp(self): self.compressor = brotli.Compressor(quality=1) -class TestCompressorQuality6(_TestCompressor, _CompressionTestCase): +class TestCompressorQuality6(_TestCompressor, _test_utils.TestCase): def setUp(self): self.compressor = brotli.Compressor(quality=6) -class TestCompressorQuality9(_TestCompressor, _CompressionTestCase): +class TestCompressorQuality9(_TestCompressor, _test_utils.TestCase): def setUp(self): self.compressor = brotli.Compressor(quality=9) -class TestCompressorQuality11(_TestCompressor, _CompressionTestCase): +class TestCompressorQuality11(_TestCompressor, _test_utils.TestCase): def setUp(self): self.compressor = brotli.Compressor(quality=11) |