view contrib/python-zstandard/tests/test_buffer_util.py @ 40450:41f0529b5112 stable

commandserver: get around ETIMEDOUT raised by selectors2 selector.select() should exits with an empty event list on timed out, but selectors2 raises OSError if timeout expires while recovering from EINTR. Spotted while debugging new chg feature.
author Yuya Nishihara <yuya@tcha.org>
date Mon, 03 Dec 2018 21:45:15 +0900
parents b1fb341d8a61
children 675775c33ab6
line wrap: on
line source

import struct
import unittest

import zstandard as zstd

ss = struct.Struct('=QQ')


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.assertRaisesRegexp(ValueError, 'segments array size is not a multiple of 16'):
            zstd.BufferWithSegments(b'foo', b'\x00\x00')

    def test_invalid_offset(self):
        with self.assertRaisesRegexp(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.assertRaisesRegexp(IndexError, 'offset must be non-negative'):
            test = b[-10]

        with self.assertRaisesRegexp(IndexError, 'offset must be less than 1'):
            test = b[1]

        with self.assertRaisesRegexp(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')


class TestBufferWithSegmentsCollection(unittest.TestCase):
    def test_empty_constructor(self):
        with self.assertRaisesRegexp(ValueError, 'must pass at least 1 argument'):
            zstd.BufferWithSegmentsCollection()

    def test_argument_validation(self):
        with self.assertRaisesRegexp(TypeError, 'arguments must be BufferWithSegments'):
            zstd.BufferWithSegmentsCollection(None)

        with self.assertRaisesRegexp(TypeError, 'arguments must be BufferWithSegments'):
            zstd.BufferWithSegmentsCollection(zstd.BufferWithSegments(b'foo', ss.pack(0, 3)),
                                              None)

        with self.assertRaisesRegexp(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.assertRaisesRegexp(IndexError, 'offset must be less than 3'):
            c[3]

        with self.assertRaisesRegexp(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')