mirror of
https://github.com/capnproto/pycapnp.git
synced 2025-03-04 16:35:04 +01:00
99 lines
2.9 KiB
Python
99 lines
2.9 KiB
Python
import pytest
|
|
import platform
|
|
import capnp
|
|
import os
|
|
import tempfile
|
|
import sys
|
|
|
|
this_dir = os.path.dirname(__file__)
|
|
|
|
|
|
@pytest.fixture
|
|
def test_capnp():
|
|
return capnp.load(os.path.join(this_dir, 'test_large_read.capnp'))
|
|
|
|
|
|
def test_large_read(test_capnp):
|
|
f = tempfile.TemporaryFile()
|
|
|
|
array = test_capnp.MultiArray.new_message()
|
|
|
|
row = array.init('rows', 1)[0]
|
|
values = row.init('values', 10000)
|
|
for i in range(len(values)):
|
|
values[i] = i
|
|
|
|
array.write_packed(f)
|
|
f.seek(0)
|
|
|
|
array = test_capnp.MultiArray.read_packed(f)
|
|
del f
|
|
assert array.rows[0].values[9000] == 9000
|
|
|
|
def test_large_read_multiple(test_capnp):
|
|
f = tempfile.TemporaryFile()
|
|
msg1 = test_capnp.Msg.new_message()
|
|
msg1.data = [0x41] * 8192
|
|
msg1.write(f)
|
|
msg2 = test_capnp.Msg.new_message()
|
|
msg2.write(f)
|
|
f.seek(0)
|
|
|
|
for m in test_capnp.Msg.read_multiple(f):
|
|
pass
|
|
|
|
def get_two_adjacent_messages(test_capnp):
|
|
msg1 = test_capnp.Msg.new_message()
|
|
msg1.data = [0x41] * 8192
|
|
m1 = msg1.to_bytes()
|
|
msg2 = test_capnp.Msg.new_message()
|
|
m2 = msg2.to_bytes()
|
|
|
|
return m1 + m2
|
|
|
|
def test_large_read_multiple_bytes(test_capnp):
|
|
data = get_two_adjacent_messages(test_capnp)
|
|
for m in test_capnp.Msg.read_multiple_bytes(data):
|
|
pass
|
|
|
|
with pytest.raises(capnp.KjException):
|
|
data = get_two_adjacent_messages(test_capnp)[:-1]
|
|
for m in test_capnp.Msg.read_multiple_bytes(data):
|
|
pass
|
|
|
|
with pytest.raises(capnp.KjException):
|
|
data = get_two_adjacent_messages(test_capnp) + b' '
|
|
for m in test_capnp.Msg.read_multiple_bytes(data):
|
|
pass
|
|
|
|
@pytest.mark.skipif(platform.python_implementation() == 'PyPy', reason="PyPy memoryview support is limited")
|
|
def test_large_read_mutltiple_bytes_memoryview(test_capnp):
|
|
data = get_two_adjacent_messages(test_capnp)
|
|
for m in test_capnp.Msg.read_multiple_bytes(memoryview(data)):
|
|
pass
|
|
|
|
with pytest.raises(capnp.KjException):
|
|
data = get_two_adjacent_messages(test_capnp)[:-1]
|
|
for m in test_capnp.Msg.read_multiple_bytes(memoryview(data)):
|
|
pass
|
|
|
|
with pytest.raises(capnp.KjException):
|
|
data = get_two_adjacent_messages(test_capnp) + b' '
|
|
for m in test_capnp.Msg.read_multiple_bytes(memoryview(data)):
|
|
pass
|
|
|
|
@pytest.mark.skipif(sys.version_info[0] == 3, reason="Legacy buffer support only for python 2.7")
|
|
def test_large_read_mutltiple_bytes_buffer(test_capnp):
|
|
data = get_two_adjacent_messages(test_capnp)
|
|
for m in test_capnp.Msg.read_multiple_bytes(buffer(data)):
|
|
pass
|
|
|
|
with pytest.raises(capnp.KjException):
|
|
data = get_two_adjacent_messages(test_capnp)[:-1]
|
|
for m in test_capnp.Msg.read_multiple_bytes(buffer(data)):
|
|
pass
|
|
|
|
with pytest.raises(capnp.KjException):
|
|
data = get_two_adjacent_messages(test_capnp) + b' '
|
|
for m in test_capnp.Msg.read_multiple_bytes(buffer(data)):
|
|
pass
|