75e0e7e84c
- Needed to include cleanup_global_schema_parser() to handle duplicate imports of the same .capnp file * Duplicate IDs are a problem as pytest does not fully cleanup between tests - Marked some tests as xfail as I'm not sure the test is supposed to work anymore with recent versions of capnproto
120 lines
3.1 KiB
Python
120 lines
3.1 KiB
Python
import pytest
|
|
import capnp
|
|
import socket
|
|
|
|
import test_capability_capnp
|
|
|
|
|
|
class Server(test_capability_capnp.TestInterface.Server):
|
|
|
|
def __init__(self, val=1):
|
|
self.val = val
|
|
|
|
def foo(self, i, j, **kwargs):
|
|
return str(i * 5 + self.val)
|
|
|
|
|
|
def restore_func(ref_id):
|
|
return Server(100)
|
|
|
|
|
|
class SimpleRestorer(test_capability_capnp.TestSturdyRefObjectId.Restorer):
|
|
|
|
def restore(self, ref_id):
|
|
assert ref_id.tag == 'testInterface'
|
|
return Server(100)
|
|
|
|
|
|
def test_simple_rpc():
|
|
read, write = socket.socketpair(socket.AF_UNIX)
|
|
|
|
restorer = SimpleRestorer()
|
|
_ = capnp.TwoPartyServer(write, restorer)
|
|
client = capnp.TwoPartyClient(read)
|
|
|
|
ref = test_capability_capnp.TestSturdyRefObjectId.new_message(tag='testInterface')
|
|
cap = client.restore(ref)
|
|
cap = cap.cast_as(test_capability_capnp.TestInterface)
|
|
|
|
remote = cap.foo(i=5)
|
|
response = remote.wait()
|
|
|
|
assert response.x == '125'
|
|
|
|
|
|
def test_simple_rpc_with_options():
|
|
read, write = socket.socketpair(socket.AF_UNIX)
|
|
|
|
restorer = SimpleRestorer()
|
|
_ = capnp.TwoPartyServer(write, restorer)
|
|
# This traversal limit is too low to receive the response in, so we expect
|
|
# an exception during the call.
|
|
client = capnp.TwoPartyClient(read, traversal_limit_in_words=1)
|
|
|
|
ref = test_capability_capnp.TestSturdyRefObjectId.new_message(tag='testInterface')
|
|
with pytest.raises(capnp.KjException):
|
|
cap = client.restore(ref)
|
|
cap = cap.cast_as(test_capability_capnp.TestInterface)
|
|
|
|
remote = cap.foo(i=5)
|
|
_ = remote.wait()
|
|
|
|
|
|
def test_simple_rpc_restore_func():
|
|
read, write = socket.socketpair(socket.AF_UNIX)
|
|
|
|
_ = capnp.TwoPartyServer(write, restore_func)
|
|
client = capnp.TwoPartyClient(read)
|
|
|
|
ref = test_capability_capnp.TestSturdyRefObjectId.new_message(tag='testInterface')
|
|
cap = client.restore(ref)
|
|
cap = cap.cast_as(test_capability_capnp.TestInterface)
|
|
|
|
remote = cap.foo(i=5)
|
|
response = remote.wait()
|
|
|
|
assert response.x == '125'
|
|
|
|
|
|
def text_restore_func(objectId):
|
|
text = objectId.as_text()
|
|
assert text == 'testInterface'
|
|
return Server(100)
|
|
|
|
|
|
def test_ez_rpc():
|
|
read, write = socket.socketpair(socket.AF_UNIX)
|
|
|
|
_ = capnp.TwoPartyServer(write, text_restore_func)
|
|
client = capnp.TwoPartyClient(read)
|
|
|
|
cap = client.ez_restore('testInterface')
|
|
cap = cap.cast_as(test_capability_capnp.TestInterface)
|
|
|
|
remote = cap.foo(i=5)
|
|
response = remote.wait()
|
|
|
|
assert response.x == '125'
|
|
|
|
cap = client.restore(test_capability_capnp.TestSturdyRefObjectId.new_message())
|
|
cap = cap.cast_as(test_capability_capnp.TestInterface)
|
|
|
|
remote = cap.foo(i=5)
|
|
|
|
with pytest.raises(capnp.KjException):
|
|
response = remote.wait()
|
|
|
|
def test_simple_rpc_bootstrap():
|
|
read, write = socket.socketpair(socket.AF_UNIX)
|
|
|
|
_ = capnp.TwoPartyServer(write, bootstrap=Server(100))
|
|
client = capnp.TwoPartyClient(read)
|
|
|
|
cap = client.bootstrap()
|
|
cap = cap.cast_as(test_capability_capnp.TestInterface)
|
|
|
|
remote = cap.foo(i=5)
|
|
response = remote.wait()
|
|
|
|
assert response.x == '125'
|