mirror of
https://github.com/msgpack/msgpack-python.git
synced 2025-10-25 06:34:10 +00:00
implement unpack_one also for the cython version, and add a test for it
This commit is contained in:
parent
a7485eccb2
commit
ff858387d3
2 changed files with 32 additions and 1 deletions
|
|
@ -359,6 +359,24 @@ cdef class Unpacker(object):
|
||||||
"""
|
"""
|
||||||
return self._unpack(unpack_construct, write_bytes)
|
return self._unpack(unpack_construct, write_bytes)
|
||||||
|
|
||||||
|
def unpack_one(self, object write_bytes=None):
|
||||||
|
"""
|
||||||
|
unpack one object
|
||||||
|
|
||||||
|
If write_bytes is not None, it will be called with parts of the raw
|
||||||
|
message as it is unpacked.
|
||||||
|
|
||||||
|
Raises `UnpackValueError` if there are no more bytes to unpack.
|
||||||
|
Raises ``ExtraData`` if there are still bytes left after the unpacking.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
result = self.unpack()
|
||||||
|
except OutOfData:
|
||||||
|
raise UnpackValueError("Data is not enough")
|
||||||
|
if self.buf_head < self.buf_tail:
|
||||||
|
raise ExtraData(result, self.buf[self.buf_head:])
|
||||||
|
return result
|
||||||
|
|
||||||
def skip(self, object write_bytes=None):
|
def skip(self, object write_bytes=None):
|
||||||
"""
|
"""
|
||||||
read and ignore one object, returning None
|
read and ignore one object, returning None
|
||||||
|
|
|
||||||
|
|
@ -1,9 +1,10 @@
|
||||||
#!/usr/bin/env python
|
#!/usr/bin/env python
|
||||||
# coding: utf-8
|
# coding: utf-8
|
||||||
|
|
||||||
|
import py
|
||||||
import six
|
import six
|
||||||
from msgpack import Unpacker, BufferFull
|
from msgpack import Unpacker, BufferFull
|
||||||
from msgpack.exceptions import OutOfData
|
from msgpack.exceptions import OutOfData, ExtraData, UnpackValueError
|
||||||
from pytest import raises
|
from pytest import raises
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -85,3 +86,15 @@ def test_readbytes():
|
||||||
assert unpacker.unpack() == ord(b'a')
|
assert unpacker.unpack() == ord(b'a')
|
||||||
assert unpacker.unpack() == ord(b'r')
|
assert unpacker.unpack() == ord(b'r')
|
||||||
|
|
||||||
|
def test_unpack_one():
|
||||||
|
unpacker = Unpacker()
|
||||||
|
unpacker.feed('\xda\x00\x03abc')
|
||||||
|
assert unpacker.unpack_one() == 'abc'
|
||||||
|
#
|
||||||
|
unpacker = Unpacker()
|
||||||
|
unpacker.feed('\xda\x00\x03abcd')
|
||||||
|
py.test.raises(ExtraData, "unpacker.unpack_one()")
|
||||||
|
#
|
||||||
|
unpacker = Unpacker()
|
||||||
|
unpacker.feed('\xda\x00\x03ab')
|
||||||
|
py.test.raises(UnpackValueError, "unpacker.unpack_one()")
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue