test_buffer_util.py
146 lines
| 5.0 KiB
| text/x-python
|
PythonLexer
|
r31796 | import struct | ||
|
r37513 | import unittest | ||
|
r31796 | |||
|
r37513 | import zstandard as zstd | ||
|
r31796 | |||
|
r44446 | from .common import TestCase | ||
ss = struct.Struct("=QQ") | ||||
|
r31796 | |||
|
r44446 | class TestBufferWithSegments(TestCase): | ||
|
r31796 | def test_arguments(self): | ||
|
r44446 | if not hasattr(zstd, "BufferWithSegments"): | ||
self.skipTest("BufferWithSegments not available") | ||||
|
r42237 | |||
|
r31796 | with self.assertRaises(TypeError): | ||
zstd.BufferWithSegments() | ||||
with self.assertRaises(TypeError): | ||||
|
r44446 | zstd.BufferWithSegments(b"foo") | ||
|
r31796 | |||
# Segments data should be a multiple of 16. | ||||
|
r44446 | with self.assertRaisesRegex( | ||
ValueError, "segments array size is not a multiple of 16" | ||||
): | ||||
zstd.BufferWithSegments(b"foo", b"\x00\x00") | ||||
|
r31796 | |||
def test_invalid_offset(self): | ||||
|
r44446 | if not hasattr(zstd, "BufferWithSegments"): | ||
self.skipTest("BufferWithSegments not available") | ||||
|
r42237 | |||
|
r44446 | with self.assertRaisesRegex( | ||
ValueError, "offset within segments array references memory" | ||||
): | ||||
zstd.BufferWithSegments(b"foo", ss.pack(0, 4)) | ||||
|
r31796 | |||
def test_invalid_getitem(self): | ||||
|
r44446 | if not hasattr(zstd, "BufferWithSegments"): | ||
self.skipTest("BufferWithSegments not available") | ||||
|
r42237 | |||
|
r44446 | b = zstd.BufferWithSegments(b"foo", ss.pack(0, 3)) | ||
|
r31796 | |||
|
r44446 | with self.assertRaisesRegex(IndexError, "offset must be non-negative"): | ||
|
r31796 | test = b[-10] | ||
|
r44446 | with self.assertRaisesRegex(IndexError, "offset must be less than 1"): | ||
|
r31796 | test = b[1] | ||
|
r44446 | with self.assertRaisesRegex(IndexError, "offset must be less than 1"): | ||
|
r31796 | test = b[2] | ||
def test_single(self): | ||||
|
r44446 | if not hasattr(zstd, "BufferWithSegments"): | ||
self.skipTest("BufferWithSegments not available") | ||||
|
r42237 | |||
|
r44446 | b = zstd.BufferWithSegments(b"foo", ss.pack(0, 3)) | ||
|
r31796 | self.assertEqual(len(b), 1) | ||
self.assertEqual(b.size, 3) | ||||
|
r44446 | self.assertEqual(b.tobytes(), b"foo") | ||
|
r31796 | |||
self.assertEqual(len(b[0]), 3) | ||||
self.assertEqual(b[0].offset, 0) | ||||
|
r44446 | self.assertEqual(b[0].tobytes(), b"foo") | ||
|
r31796 | |||
def test_multiple(self): | ||||
|
r44446 | if not hasattr(zstd, "BufferWithSegments"): | ||
self.skipTest("BufferWithSegments not available") | ||||
|
r42237 | |||
|
r44446 | b = zstd.BufferWithSegments( | ||
b"foofooxfooxy", b"".join([ss.pack(0, 3), ss.pack(3, 4), ss.pack(7, 5)]) | ||||
) | ||||
|
r31796 | self.assertEqual(len(b), 3) | ||
self.assertEqual(b.size, 12) | ||||
|
r44446 | self.assertEqual(b.tobytes(), b"foofooxfooxy") | ||
|
r31796 | |||
|
r44446 | self.assertEqual(b[0].tobytes(), b"foo") | ||
self.assertEqual(b[1].tobytes(), b"foox") | ||||
self.assertEqual(b[2].tobytes(), b"fooxy") | ||||
|
r31796 | |||
|
r44446 | class TestBufferWithSegmentsCollection(TestCase): | ||
|
r31796 | def test_empty_constructor(self): | ||
|
r44446 | if not hasattr(zstd, "BufferWithSegmentsCollection"): | ||
self.skipTest("BufferWithSegmentsCollection not available") | ||||
|
r42237 | |||
|
r44446 | with self.assertRaisesRegex(ValueError, "must pass at least 1 argument"): | ||
|
r31796 | zstd.BufferWithSegmentsCollection() | ||
def test_argument_validation(self): | ||||
|
r44446 | if not hasattr(zstd, "BufferWithSegmentsCollection"): | ||
self.skipTest("BufferWithSegmentsCollection not available") | ||||
|
r42237 | |||
|
r44446 | with self.assertRaisesRegex(TypeError, "arguments must be BufferWithSegments"): | ||
|
r31796 | zstd.BufferWithSegmentsCollection(None) | ||
|
r44446 | with self.assertRaisesRegex(TypeError, "arguments must be BufferWithSegments"): | ||
zstd.BufferWithSegmentsCollection( | ||||
zstd.BufferWithSegments(b"foo", ss.pack(0, 3)), None | ||||
) | ||||
|
r31796 | |||
|
r44446 | with self.assertRaisesRegex( | ||
ValueError, "ZstdBufferWithSegments cannot be empty" | ||||
): | ||||
zstd.BufferWithSegmentsCollection(zstd.BufferWithSegments(b"", b"")) | ||||
|
r31796 | |||
def test_length(self): | ||||
|
r44446 | if not hasattr(zstd, "BufferWithSegmentsCollection"): | ||
self.skipTest("BufferWithSegmentsCollection not available") | ||||
|
r42237 | |||
|
r44446 | b1 = zstd.BufferWithSegments(b"foo", ss.pack(0, 3)) | ||
b2 = zstd.BufferWithSegments( | ||||
b"barbaz", b"".join([ss.pack(0, 3), ss.pack(3, 3)]) | ||||
) | ||||
|
r31796 | |||
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): | ||||
|
r44446 | if not hasattr(zstd, "BufferWithSegmentsCollection"): | ||
self.skipTest("BufferWithSegmentsCollection not available") | ||||
|
r42237 | |||
|
r44446 | b1 = zstd.BufferWithSegments(b"foo", ss.pack(0, 3)) | ||
b2 = zstd.BufferWithSegments( | ||||
b"barbaz", b"".join([ss.pack(0, 3), ss.pack(3, 3)]) | ||||
) | ||||
|
r31796 | |||
c = zstd.BufferWithSegmentsCollection(b1, b2) | ||||
|
r44446 | with self.assertRaisesRegex(IndexError, "offset must be less than 3"): | ||
|
r31796 | c[3] | ||
|
r44446 | with self.assertRaisesRegex(IndexError, "offset must be less than 3"): | ||
|
r31796 | c[4] | ||
|
r44446 | self.assertEqual(c[0].tobytes(), b"foo") | ||
self.assertEqual(c[1].tobytes(), b"bar") | ||||
self.assertEqual(c[2].tobytes(), b"baz") | ||||