2024-05-06 03:33:48 +09:00
|
|
|
from pytest import raises
|
2010-11-02 14:09:50 +09:00
|
|
|
|
2024-05-06 11:46:31 +09:00
|
|
|
from msgpack import Packer, packb, unpackb
|
2010-11-02 14:09:50 +09:00
|
|
|
|
2012-12-29 11:24:25 +09:00
|
|
|
|
2010-11-02 14:09:50 +09:00
|
|
|
def test_unpack_buffer():
|
|
|
|
from array import array
|
2019-12-05 18:51:45 +09:00
|
|
|
|
|
|
|
buf = array("b")
|
2019-12-05 20:47:20 +09:00
|
|
|
buf.frombytes(packb((b"foo", b"bar")))
|
2012-09-24 02:12:55 +09:00
|
|
|
obj = unpackb(buf, use_list=1)
|
2019-12-05 18:51:45 +09:00
|
|
|
assert [b"foo", b"bar"] == obj
|
2010-11-02 14:09:50 +09:00
|
|
|
|
2013-12-15 16:22:39 +00:00
|
|
|
|
|
|
|
def test_unpack_bytearray():
|
2019-12-05 21:34:10 +09:00
|
|
|
buf = bytearray(packb((b"foo", b"bar")))
|
2013-12-15 16:22:39 +00:00
|
|
|
obj = unpackb(buf, use_list=1)
|
2019-12-05 18:51:45 +09:00
|
|
|
assert [b"foo", b"bar"] == obj
|
2014-02-17 04:06:58 +09:00
|
|
|
expected_type = bytes
|
2025-10-08 17:56:20 +09:00
|
|
|
assert all(type(s) is expected_type for s in obj)
|
2016-05-07 15:18:20 +02:00
|
|
|
|
|
|
|
|
|
|
|
def test_unpack_memoryview():
|
2019-12-05 21:34:10 +09:00
|
|
|
buf = bytearray(packb((b"foo", b"bar")))
|
2016-05-07 15:18:20 +02:00
|
|
|
view = memoryview(buf)
|
|
|
|
obj = unpackb(view, use_list=1)
|
2019-12-05 18:51:45 +09:00
|
|
|
assert [b"foo", b"bar"] == obj
|
2016-05-07 15:18:20 +02:00
|
|
|
expected_type = bytes
|
2025-10-08 17:56:20 +09:00
|
|
|
assert all(type(s) is expected_type for s in obj)
|
2024-05-06 03:33:48 +09:00
|
|
|
|
|
|
|
|
|
|
|
def test_packer_getbuffer():
|
|
|
|
packer = Packer(autoreset=False)
|
|
|
|
packer.pack_array_header(2)
|
|
|
|
packer.pack(42)
|
|
|
|
packer.pack("hello")
|
|
|
|
buffer = packer.getbuffer()
|
|
|
|
assert isinstance(buffer, memoryview)
|
|
|
|
assert bytes(buffer) == b"\x92*\xa5hello"
|
|
|
|
|
|
|
|
if Packer.__module__ == "msgpack._cmsgpack": # only for Cython
|
|
|
|
# cython Packer supports buffer protocol directly
|
|
|
|
assert bytes(packer) == b"\x92*\xa5hello"
|
|
|
|
|
|
|
|
with raises(BufferError):
|
|
|
|
packer.pack(42)
|
|
|
|
buffer.release()
|
|
|
|
packer.pack(42)
|
|
|
|
assert bytes(packer) == b"\x92*\xa5hello*"
|