mirror of
https://github.com/msgpack/msgpack-python.git
synced 2025-10-20 12:23:16 +00:00
e381032641
Merge remote-tracking branch 'jnothman/object_pairs_hook' into 0.2-maint Conflicts: msgpack/_msgpack.pyx test/test_pack.py test/test_sequnpack.py
60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
#!/usr/bin/env python
|
|
# coding: utf-8
|
|
|
|
from msgpack import Unpacker, BufferFull
|
|
import nose
|
|
|
|
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')
|
|
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_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()
|
|
try:
|
|
o = unpacker.unpack()
|
|
assert 0, "should raise exception"
|
|
except StopIteration:
|
|
assert 1, "ok"
|
|
|
|
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, use_list=1)
|
|
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()
|