• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2016 The Brotli Authors. All rights reserved.
2#
3# Distributed under MIT license.
4# See file LICENSE for detail or copy at https://opensource.org/licenses/MIT
5
6import functools
7import unittest
8
9from . import _test_utils
10import brotli
11
12
13# Do not inherit from TestCase here to ensure that test methods
14# are not run automatically and instead are run as part of a specific
15# configuration below.
16class _TestCompressor(object):
17
18    CHUNK_SIZE = 2048
19
20    def _check_decompression(self, test_data):
21        # Write decompression to temp file and verify it matches the original.
22        temp_uncompressed = _test_utils.get_temp_uncompressed_name(test_data)
23        temp_compressed = _test_utils.get_temp_compressed_name(test_data)
24        original = test_data
25        with open(temp_uncompressed, 'wb') as out_file:
26            with open(temp_compressed, 'rb') as in_file:
27                out_file.write(brotli.decompress(in_file.read()))
28        self.assertFilesMatch(temp_uncompressed, original)
29
30    def _test_single_process(self, test_data):
31        # Write single-shot compression to temp file.
32        temp_compressed = _test_utils.get_temp_compressed_name(test_data)
33        with open(temp_compressed, 'wb') as out_file:
34            with open(test_data, 'rb') as in_file:
35                out_file.write(self.compressor.process(in_file.read()))
36            out_file.write(self.compressor.finish())
37        self._check_decompression(test_data)
38
39    def _test_multiple_process(self, test_data):
40        # Write chunked compression to temp file.
41        temp_compressed = _test_utils.get_temp_compressed_name(test_data)
42        with open(temp_compressed, 'wb') as out_file:
43            with open(test_data, 'rb') as in_file:
44                read_chunk = functools.partial(in_file.read, self.CHUNK_SIZE)
45                for data in iter(read_chunk, b''):
46                    out_file.write(self.compressor.process(data))
47            out_file.write(self.compressor.finish())
48        self._check_decompression(test_data)
49
50    def _test_multiple_process_and_flush(self, test_data):
51        # Write chunked and flushed compression to temp file.
52        temp_compressed = _test_utils.get_temp_compressed_name(test_data)
53        with open(temp_compressed, 'wb') as out_file:
54            with open(test_data, 'rb') as in_file:
55                read_chunk = functools.partial(in_file.read, self.CHUNK_SIZE)
56                for data in iter(read_chunk, b''):
57                    out_file.write(self.compressor.process(data))
58                    out_file.write(self.compressor.flush())
59            out_file.write(self.compressor.finish())
60        self._check_decompression(test_data)
61
62
63_test_utils.generate_test_methods(_TestCompressor)
64
65
66class TestCompressorQuality1(_TestCompressor, _test_utils.TestCase):
67
68    def setUp(self):
69        self.compressor = brotli.Compressor(quality=1)
70
71
72class TestCompressorQuality6(_TestCompressor, _test_utils.TestCase):
73
74    def setUp(self):
75        self.compressor = brotli.Compressor(quality=6)
76
77
78class TestCompressorQuality9(_TestCompressor, _test_utils.TestCase):
79
80    def setUp(self):
81        self.compressor = brotli.Compressor(quality=9)
82
83
84class TestCompressorQuality11(_TestCompressor, _test_utils.TestCase):
85
86    def setUp(self):
87        self.compressor = brotli.Compressor(quality=11)
88
89
90if __name__ == '__main__':
91    unittest.main()
92