a5c29a74d2
Cap'n Proto provides a schema loader, which can be used to dynamically load schemas during runtime. To port this functionality to pycapnp, a new class is provided `C_SchemaLoader`, which exposes the Cap'n Proto C++ interface, and `SchemaLoader`, which is part of the pycapnp library. The specific use case for this is when a capnp message contains a Node.Reader: The schema for a yet unseen message can be loaded dynamically, allowing the future message to be properly processed. If the message is a struct containing other structs, all the schemas for every struct must be loaded to correctly parse the message. See https://github.com/DaneSlattery/capnp_generic_poc for a proof-of-concept. Add docs and cleanup Add more docs Reduce changes Fix flake8 formatting Fix get datatype
149 lines
3.5 KiB
Python
149 lines
3.5 KiB
Python
import pytest
|
|
import capnp
|
|
import os
|
|
import sys
|
|
|
|
this_dir = os.path.dirname(__file__)
|
|
|
|
|
|
@pytest.fixture
|
|
def addressbook():
|
|
return capnp.load(os.path.join(this_dir, "addressbook.capnp"))
|
|
|
|
|
|
@pytest.fixture
|
|
def foo():
|
|
return capnp.load(os.path.join(this_dir, "foo.capnp"))
|
|
|
|
|
|
@pytest.fixture
|
|
def bar():
|
|
return capnp.load(os.path.join(this_dir, "bar.capnp"))
|
|
|
|
|
|
def test_basic_load():
|
|
capnp.load(os.path.join(this_dir, "addressbook.capnp"))
|
|
|
|
|
|
def test_constants(addressbook):
|
|
assert addressbook.qux == 123
|
|
|
|
|
|
def test_classes(addressbook):
|
|
assert addressbook.AddressBook
|
|
assert addressbook.Person
|
|
|
|
|
|
def test_import(foo, bar):
|
|
m = capnp._MallocMessageBuilder()
|
|
foo = m.init_root(foo.Foo)
|
|
m2 = capnp._MallocMessageBuilder()
|
|
bar = m2.init_root(bar.Bar)
|
|
|
|
foo.name = "foo"
|
|
bar.foo = foo
|
|
|
|
assert bar.foo.name == "foo"
|
|
|
|
|
|
def test_failed_import():
|
|
s = capnp.SchemaParser()
|
|
s2 = capnp.SchemaParser()
|
|
|
|
foo = s.load(os.path.join(this_dir, "foo.capnp"))
|
|
bar = s2.load(os.path.join(this_dir, "bar.capnp"))
|
|
|
|
m = capnp._MallocMessageBuilder()
|
|
foo = m.init_root(foo.Foo)
|
|
m2 = capnp._MallocMessageBuilder()
|
|
bar = m2.init_root(bar.Bar)
|
|
|
|
foo.name = "foo"
|
|
|
|
with pytest.raises(Exception):
|
|
bar.foo = foo
|
|
|
|
|
|
def test_defualt_import_hook():
|
|
# Make sure any previous imports of addressbook_capnp are gone
|
|
capnp.cleanup_global_schema_parser()
|
|
|
|
import addressbook_capnp # noqa: F401
|
|
|
|
|
|
def test_dash_import():
|
|
import addressbook_with_dashes_capnp # noqa: F401
|
|
|
|
|
|
def test_spaces_import():
|
|
import addressbook_with_spaces_capnp # noqa: F401
|
|
|
|
|
|
def test_add_import_hook():
|
|
capnp.add_import_hook([this_dir])
|
|
|
|
# Make sure any previous imports of addressbook_capnp are gone
|
|
capnp.cleanup_global_schema_parser()
|
|
|
|
import addressbook_capnp
|
|
|
|
addressbook_capnp.AddressBook.new_message()
|
|
|
|
|
|
def test_multiple_add_import_hook():
|
|
capnp.add_import_hook()
|
|
capnp.add_import_hook()
|
|
capnp.add_import_hook([this_dir])
|
|
|
|
# Make sure any previous imports of addressbook_capnp are gone
|
|
capnp.cleanup_global_schema_parser()
|
|
|
|
import addressbook_capnp
|
|
|
|
addressbook_capnp.AddressBook.new_message()
|
|
|
|
|
|
def test_remove_import_hook():
|
|
capnp.add_import_hook([this_dir])
|
|
capnp.remove_import_hook()
|
|
|
|
if "addressbook_capnp" in sys.modules:
|
|
# hack to deal with it being imported already
|
|
del sys.modules["addressbook_capnp"]
|
|
|
|
with pytest.raises(ImportError):
|
|
import addressbook_capnp # noqa: F401
|
|
|
|
|
|
def test_bundled_import_hook():
|
|
# stream.capnp should be bundled, or provided by the system capnproto
|
|
capnp.add_import_hook()
|
|
import stream_capnp # noqa: F401
|
|
|
|
|
|
def test_load_capnp(foo):
|
|
# test dynamically loading
|
|
loader = capnp.SchemaLoader()
|
|
loader.load(foo.Baz.schema.get_proto())
|
|
loader.load_dynamic(foo.Qux.schema.get_proto().node)
|
|
|
|
schema = loader.get(foo.Baz.schema.get_proto().node.id).as_struct()
|
|
assert "text" in schema.fieldnames
|
|
assert "qux" in schema.fieldnames
|
|
assert schema.fields["qux"].proto.slot.type.which == "struct"
|
|
|
|
class Wrapper(foo.Wrapper.Server):
|
|
def wrapped(self, object, **kwargs):
|
|
assert isinstance(object, capnp.lib.capnp._DynamicObjectReader)
|
|
baz_ = object.as_struct(schema)
|
|
assert baz_.text == "test"
|
|
assert baz_.qux.id == 2
|
|
|
|
# test calling into the wrapper with a Baz message.
|
|
baz_ = foo.Baz.new_message()
|
|
baz_.text = "test"
|
|
baz_.qux.id = 2
|
|
|
|
wrapper = foo.Wrapper._new_client(Wrapper())
|
|
wrapper.wrapped(baz_).wait()
|