blob: 9db14ca7e53acefd032f0f5964edd348f6d8be86 (
plain)
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
|
#!/usr/bin/env python
# coding: utf-8
import six
from msgpack import Unpacker, BufferFull
from msgpack.exceptions import OutOfData
from pytest import raises
def test_partialdata():
unpacker = Unpacker()
unpacker.feed(b'\xa5')
with raises(StopIteration): next(iter(unpacker))
unpacker.feed(b'h')
with raises(StopIteration): next(iter(unpacker))
unpacker.feed(b'a')
with raises(StopIteration): next(iter(unpacker))
unpacker.feed(b'l')
with raises(StopIteration): next(iter(unpacker))
unpacker.feed(b'l')
with raises(StopIteration): next(iter(unpacker))
unpacker.feed(b'o')
assert next(iter(unpacker)) == b'hallo'
def test_foobar():
unpacker = Unpacker(read_size=3, use_list=1)
unpacker.feed(b'foobar')
assert unpacker.unpack() == ord(b'f')
assert unpacker.unpack() == ord(b'o')
assert unpacker.unpack() == ord(b'o')
assert unpacker.unpack() == ord(b'b')
assert unpacker.unpack() == ord(b'a')
assert unpacker.unpack() == ord(b'r')
with raises(OutOfData):
unpacker.unpack()
unpacker.feed(b'foo')
unpacker.feed(b'bar')
k = 0
for o, e in zip(unpacker, 'foobarbaz'):
assert o == ord(e)
k += 1
assert k == len(b'foobar')
def test_foobar_skip():
unpacker = Unpacker(read_size=3, use_list=1)
unpacker.feed(b'foobar')
assert unpacker.unpack() == ord(b'f')
unpacker.skip()
assert unpacker.unpack() == ord(b'o')
unpacker.skip()
assert unpacker.unpack() == ord(b'a')
unpacker.skip()
with raises(OutOfData):
unpacker.unpack()
def test_maxbuffersize():
with raises(ValueError):
Unpacker(read_size=5, max_buffer_size=3)
unpacker = Unpacker(read_size=3, max_buffer_size=3, use_list=1)
unpacker.feed(b'fo')
with raises(BufferFull):
unpacker.feed(b'ob')
unpacker.feed(b'o')
assert ord('f') == next(unpacker)
unpacker.feed(b'b')
assert ord('o') == next(unpacker)
assert ord('o') == next(unpacker)
assert ord('b') == next(unpacker)
def test_readbytes():
unpacker = Unpacker(read_size=3)
unpacker.feed(b'foobar')
assert unpacker.unpack() == ord(b'f')
assert unpacker.read_bytes(3) == b'oob'
assert unpacker.unpack() == ord(b'a')
assert unpacker.unpack() == ord(b'r')
# Test buffer refill
unpacker = Unpacker(six.BytesIO(b'foobar'), read_size=3)
assert unpacker.unpack() == ord(b'f')
assert unpacker.read_bytes(3) == b'oob'
assert unpacker.unpack() == ord(b'a')
assert unpacker.unpack() == ord(b'r')
|
