summaryrefslogtreecommitdiff
path: root/test/test_sequnpack.py
blob: b1b80b2db4b0723979b7f3df808d043ce212e60a (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
#!/usr/bin/env python
# coding: utf-8

from msgpack import Unpacker, BufferFull
import nose

def test_foobar():
    unpacker = Unpacker(read_size=3)
    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')
    try:
        o = unpacker.unpack()
        assert 0, "should raise exception"
    except StopIteration:
        assert 1, "ok"

    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_maxbuffersize():
    nose.tools.assert_raises(ValueError, Unpacker, read_size=5, max_buffer_size=3)
    unpacker = Unpacker(read_size=3, max_buffer_size=3)
    unpacker.feed(b'fo')
    nose.tools.assert_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)


if __name__ == '__main__':
    nose.main()