diff --git a/capnp/lib/capnp.pyx b/capnp/lib/capnp.pyx index aa23137..864993a 100644 --- a/capnp/lib/capnp.pyx +++ b/capnp/lib/capnp.pyx @@ -15,7 +15,7 @@ from libc.stdlib cimport malloc, free from libc.string cimport memcpy from cython.operator cimport dereference as deref from cpython.exc cimport PyErr_Clear -from cpython cimport Py_buffer +from cpython cimport Py_buffer, PyObject_CheckBuffer from cpython.buffer cimport PyBUF_SIMPLE from types import ModuleType as _ModuleType @@ -31,7 +31,6 @@ import threading as _threading import socket as _socket import random as _random import collections as _collections -import mmap as _mmap _CAPNP_VERSION_MAJOR = capnp.CAPNP_VERSION_MAJOR _CAPNP_VERSION_MINOR = capnp.CAPNP_VERSION_MINOR @@ -3756,11 +3755,7 @@ cdef class _FlatArrayMessageReader(_MessageReader): raise ValueError("input length must be a multiple of eight bytes") cdef char * ptr - if type(buf) == _mmap.mmap: - view = _BufferView(buf) - ptr = view.buf - self._object_to_pin = view - else: + if isinstance(buf, bytes): ptr = buf if (ptr) % 8 != 0: aligned = _AlignedBuffer(buf) @@ -3768,6 +3763,12 @@ cdef class _FlatArrayMessageReader(_MessageReader): self._object_to_pin = aligned else: self._object_to_pin = buf + elif PyObject_CheckBuffer(buf): + view = _BufferView(buf) + ptr = view.buf + self._object_to_pin = view + else: + raise TypeError('expected buffer-like object in FlatArrayMessageReader') self.thisptr = new schema_cpp.FlatArrayMessageReader( schema_cpp.WordArrayPtr(ptr, sz//8), diff --git a/test/test_serialization.py b/test/test_serialization.py index 392bbd2..a3ff183 100644 --- a/test/test_serialization.py +++ b/test/test_serialization.py @@ -65,6 +65,20 @@ def test_roundtrip_bytes_mmap(all_types): msg = all_types.TestAllTypes.from_bytes(memory) test_regression.check_all_types(msg) +@pytest.mark.skipif(sys.version_info[0] < 3, reason="memoryview is a builtin on Python 3") +def test_roundtrip_bytes_buffer(all_types): + msg = all_types.TestAllTypes.new_message() + test_regression.init_all_types(msg) + + b = msg.to_bytes() + v = memoryview(b) + msg = all_types.TestAllTypes.from_bytes(v) + test_regression.check_all_types(msg) + +def test_roundtrip_bytes_fail(all_types): + with pytest.raises(TypeError): + all_types.TestAllTypes.from_bytes(42) + @pytest.mark.skipif(platform.python_implementation() == 'PyPy', reason="This works in PyPy 4.0.1 but travisci's version of PyPy has some bug that fails this test.") def test_roundtrip_bytes_packed(all_types): msg = all_types.TestAllTypes.new_message()