2012-10-04 11:26:29 +10:00
|
|
|
"""Tests for cases where the user seeks to obtain packed msgpack objects"""
|
|
|
|
|
2014-02-15 22:16:01 +09:00
|
|
|
import io
|
2012-10-04 11:26:29 +10:00
|
|
|
from msgpack import Unpacker, packb
|
|
|
|
|
2012-12-06 19:15:04 +09:00
|
|
|
|
2012-10-04 11:26:29 +10:00
|
|
|
def test_write_bytes():
|
|
|
|
unpacker = Unpacker()
|
|
|
|
unpacker.feed(b'abc')
|
2014-02-15 22:16:01 +09:00
|
|
|
f = io.BytesIO()
|
2012-12-29 11:24:25 +09:00
|
|
|
assert unpacker.unpack(f.write) == ord('a')
|
|
|
|
assert f.getvalue() == b'a'
|
2014-02-15 22:16:01 +09:00
|
|
|
f = io.BytesIO()
|
2012-12-06 19:15:04 +09:00
|
|
|
assert unpacker.skip(f.write) is None
|
2012-12-29 11:24:25 +09:00
|
|
|
assert f.getvalue() == b'b'
|
2014-02-15 22:16:01 +09:00
|
|
|
f = io.BytesIO()
|
2012-12-06 19:15:04 +09:00
|
|
|
assert unpacker.skip() is None
|
2012-12-29 11:24:25 +09:00
|
|
|
assert f.getvalue() == b''
|
2012-10-04 11:26:29 +10:00
|
|
|
|
2012-12-06 19:15:04 +09:00
|
|
|
|
2012-10-04 11:26:29 +10:00
|
|
|
def test_write_bytes_multi_buffer():
|
|
|
|
long_val = (5) * 100
|
|
|
|
expected = packb(long_val)
|
2014-02-15 22:16:01 +09:00
|
|
|
unpacker = Unpacker(io.BytesIO(expected), read_size=3, max_buffer_size=3)
|
2012-10-04 11:26:29 +10:00
|
|
|
|
2014-02-15 22:16:01 +09:00
|
|
|
f = io.BytesIO()
|
2012-10-04 11:26:29 +10:00
|
|
|
unpacked = unpacker.unpack(f.write)
|
2012-12-29 11:24:25 +09:00
|
|
|
assert unpacked == long_val
|
|
|
|
assert f.getvalue() == expected
|