1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
|
import io
import os
from unittest import TestCase
from snappy import snappy_formats as formats
from snappy.snappy import _CHUNK_MAX, UncompressError
class TestFormatBase(TestCase):
compress_format = formats.FORMAT_AUTO
decompress_format = formats.FORMAT_AUTO
success = True
def runTest(self):
data = os.urandom(1024 * 256 * 2) + os.urandom(13245 * 2)
compress_func = formats.get_compress_function(self.compress_format)
instream = io.BytesIO(data)
compressed_stream = io.BytesIO()
compress_func(instream, compressed_stream)
compressed_stream.seek(0)
if not self.success:
with self.assertRaises(UncompressError) as err:
decompress_func, read_chunk = formats.get_decompress_function(
self.decompress_format, compressed_stream
)
decompressed_stream = io.BytesIO()
decompress_func(
compressed_stream,
decompressed_stream,
start_chunk=read_chunk
)
return
decompress_func, read_chunk = formats.get_decompress_function(
self.decompress_format, compressed_stream
)
decompressed_stream = io.BytesIO()
decompress_func(
compressed_stream,
decompressed_stream,
start_chunk=read_chunk
)
decompressed_stream.seek(0)
self.assertEqual(data, decompressed_stream.read())
class TestFormatFramingFraming(TestFormatBase):
compress_format = formats.FRAMING_FORMAT
decompress_format = formats.FRAMING_FORMAT
success = True
class TestFormatFramingHadoop(TestFormatBase):
compress_format = formats.FRAMING_FORMAT
decompress_format = formats.HADOOP_FORMAT
success = False
class TestFormatFramingAuto(TestFormatBase):
compress_format = formats.FRAMING_FORMAT
decompress_format = formats.FORMAT_AUTO
success = True
class TestFormatHadoopHadoop(TestFormatBase):
compress_format = formats.HADOOP_FORMAT
decompress_format = formats.HADOOP_FORMAT
success = True
class TestFormatHadoopFraming(TestFormatBase):
compress_format = formats.HADOOP_FORMAT
decompress_format = formats.FRAMING_FORMAT
success = False
class TestFormatHadoopAuto(TestFormatBase):
compress_format = formats.HADOOP_FORMAT
decompress_format = formats.FORMAT_AUTO
success = True
class TestFormatAutoFraming(TestFormatBase):
compress_format = formats.FORMAT_AUTO
decompress_format = formats.FRAMING_FORMAT
success = True
class TestFormatAutoHadoop(TestFormatBase):
compress_format = formats.FORMAT_AUTO
decompress_format = formats.HADOOP_FORMAT
success = False
if __name__ == "__main__":
import unittest
unittest.main()
|
