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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
import struct
import unittest
import zstandard as zstd
ss = struct.Struct("=QQ")
@unittest.skipUnless(
"buffer_types" in zstd.backend_features, "buffer types not available"
)
class TestBufferWithSegments(unittest.TestCase):
def test_arguments(self):
with self.assertRaises(TypeError):
zstd.BufferWithSegments()
with self.assertRaises(TypeError):
zstd.BufferWithSegments(b"foo")
# Segments data should be a multiple of 16.
with self.assertRaisesRegex(
ValueError, "segments array size is not a multiple of 16"
):
zstd.BufferWithSegments(b"foo", b"\x00\x00")
def test_invalid_offset(self):
with self.assertRaisesRegex(
ValueError, "offset within segments array references memory"
):
zstd.BufferWithSegments(b"foo", ss.pack(0, 4))
def test_invalid_getitem(self):
b = zstd.BufferWithSegments(b"foo", ss.pack(0, 3))
with self.assertRaisesRegex(IndexError, "offset must be non-negative"):
test = b[-10]
with self.assertRaisesRegex(IndexError, "offset must be less than 1"):
test = b[1]
with self.assertRaisesRegex(IndexError, "offset must be less than 1"):
test = b[2]
def test_single(self):
b = zstd.BufferWithSegments(b"foo", ss.pack(0, 3))
self.assertEqual(len(b), 1)
self.assertEqual(b.size, 3)
self.assertEqual(b.tobytes(), b"foo")
self.assertEqual(len(b[0]), 3)
self.assertEqual(b[0].offset, 0)
self.assertEqual(b[0].tobytes(), b"foo")
def test_multiple(self):
b = zstd.BufferWithSegments(
b"foofooxfooxy",
b"".join([ss.pack(0, 3), ss.pack(3, 4), ss.pack(7, 5)]),
)
self.assertEqual(len(b), 3)
self.assertEqual(b.size, 12)
self.assertEqual(b.tobytes(), b"foofooxfooxy")
self.assertEqual(b[0].tobytes(), b"foo")
self.assertEqual(b[1].tobytes(), b"foox")
self.assertEqual(b[2].tobytes(), b"fooxy")
@unittest.skipUnless(
"buffer_types" in zstd.backend_features, "buffer types not available"
)
class TestBufferWithSegmentsCollection(unittest.TestCase):
def test_empty_constructor(self):
with self.assertRaisesRegex(
ValueError, "must pass at least 1 argument"
):
zstd.BufferWithSegmentsCollection()
def test_argument_validation(self):
with self.assertRaisesRegex(
TypeError, "arguments must be BufferWithSegments"
):
zstd.BufferWithSegmentsCollection(None)
with self.assertRaisesRegex(
TypeError, "arguments must be BufferWithSegments"
):
zstd.BufferWithSegmentsCollection(
zstd.BufferWithSegments(b"foo", ss.pack(0, 3)), None
)
with self.assertRaisesRegex(
ValueError, "ZstdBufferWithSegments cannot be empty"
):
zstd.BufferWithSegmentsCollection(zstd.BufferWithSegments(b"", b""))
def test_length(self):
b1 = zstd.BufferWithSegments(b"foo", ss.pack(0, 3))
b2 = zstd.BufferWithSegments(
b"barbaz", b"".join([ss.pack(0, 3), ss.pack(3, 3)])
)
c = zstd.BufferWithSegmentsCollection(b1)
self.assertEqual(len(c), 1)
self.assertEqual(c.size(), 3)
c = zstd.BufferWithSegmentsCollection(b2)
self.assertEqual(len(c), 2)
self.assertEqual(c.size(), 6)
c = zstd.BufferWithSegmentsCollection(b1, b2)
self.assertEqual(len(c), 3)
self.assertEqual(c.size(), 9)
def test_getitem(self):
b1 = zstd.BufferWithSegments(b"foo", ss.pack(0, 3))
b2 = zstd.BufferWithSegments(
b"barbaz", b"".join([ss.pack(0, 3), ss.pack(3, 3)])
)
c = zstd.BufferWithSegmentsCollection(b1, b2)
with self.assertRaisesRegex(IndexError, "offset must be less than 3"):
c[3]
with self.assertRaisesRegex(IndexError, "offset must be less than 3"):
c[4]
self.assertEqual(c[0].tobytes(), b"foo")
self.assertEqual(c[1].tobytes(), b"bar")
self.assertEqual(c[2].tobytes(), b"baz")
|
