use capnproto codec in tests

Author: Mahad-10

pyproject.toml

@@ -39,7 +39,7 @@ test = [
   "pytest-timeout",
 ]
 publish = ["twine", "build"]
-dev = ["mkdocs-material[imaging]", "wampproto@git+https://github.com/xconnio/wampproto-python", "protobuf"]
+dev = ["mkdocs-material[imaging]", "wampproto@git+https://github.com/xconnio/wampproto-python", "protobuf", "pycapnp"]
 capnproto = [
     "wampproto-messages-capnproto@git+https://github.com/xconnio/wampproto-messages-capnproto@main#subdirectory=python"
 ]

tests/codec_test.py

@@ -1,6 +1,8 @@
 import base64
-from typing import Type
+from pathlib import Path
+from typing import Type, TypeVar, Any
 
+import capnp
 from google.protobuf.message import Message
 
 from xconn.client import connect_anonymous
@@ -179,3 +181,109 @@ def get_profile_handler() -> ProfileGet:
     assert profile.created_at == "2025-10-30T17:00:00Z"
 
     session.leave()
+
+
+T = TypeVar("T")
+SCHEMA_PATH = Path(__file__).parent / "user.capnp"
+user_capnp = capnp.load(str(SCHEMA_PATH))
+
+UserCreate = user_capnp.UserCreate
+UserGet = user_capnp.UserGet
+
+
+class CapnpProtoCodec(codec.Codec[T]):
+    def name(self) -> str:
+        return "capnproto"
+
+    def encode(self, obj: Any) -> bytes:
+        return obj.to_bytes_packed()
+
+    def decode(self, data: bytes, out_type: Type[T]) -> T:
+        return out_type.from_bytes_packed(data)
+
+
+def test_register_object_capnproto():
+    session = connect_anonymous("ws://localhost:8080/ws", "realm1")
+    session.set_payload_codec(CapnpProtoCodec())
+
+    def create_handler(user_create: UserCreate) -> UserGet:
+        user_get = UserGet.new_message()
+        user_get.id = 999
+        user_get.name = user_create.name
+        user_get.email = user_create.email
+        user_get.age = user_create.age
+        user_get.isAdmin = False
+
+        return user_get
+
+    session.register_object("io.xconn.user.create", create_handler)
+
+    new_user = UserCreate.new_message()
+    new_user.name = "john"
+    new_user.email = "[email protected]"
+    new_user.age = 35
+
+    result = session.call("io.xconn.user.create", [new_user.to_bytes_packed()])
+    user = UserGet.from_bytes_packed(result.args[0])
+
+    assert user.id == 999
+    assert user.name == "john"
+    assert user.email == "[email protected]"
+    assert user.age == 35
+    assert not user.isAdmin
+
+    session.leave()
+
+
+def test_call_object_capnproto():
+    session = connect_anonymous("ws://localhost:8080/ws", "realm1")
+    session.set_payload_codec(CapnpProtoCodec())
+
+    def invocation_handler(inv: Invocation) -> Result:
+        user_create = UserCreate.from_bytes_packed(inv.args[0])
+
+        user_get = UserGet.new_message()
+        user_get.id = 78
+        user_get.name = user_create.name
+        user_get.email = user_create.email
+        user_get.age = user_create.age
+        user_get.isAdmin = True
+
+        return Result(args=[user_get.to_bytes_packed()])
+
+    session.register("io.xconn.user.create", invocation_handler)
+    new_user = UserCreate.new_message()
+    new_user.name = "alice"
+    new_user.email = "[email protected]"
+    new_user.age = 23
+
+    result: UserGet = session.call_object("io.xconn.user.create", new_user, UserGet)
+    assert result.id == 78
+    assert result.name == "alice"
+    assert result.email == "[email protected]"
+    assert result.age == 23
+    assert result.isAdmin
+
+    session.leave()
+
+
+def test_pubsub_capnproto():
+    session = connect_anonymous("ws://localhost:8080/ws", "realm1")
+    session.set_payload_codec(CapnpProtoCodec())
+
+    def event_handler(event: Event):
+        user: UserGet = event.args[0]
+        assert user.name == "alice"
+        assert user.email == "[email protected]"
+        assert user.age == 21
+
+    session.subscribe_object("io.xconn.object", event_handler, UserCreate)
+
+    new_user = UserCreate.new_message()
+    new_user.name = "alice"
+    new_user.email = "[email protected]"
+    new_user.age = 21
+
+    session.publish_object("io.xconn.object", new_user)
+
+    session.leave()

tests/user.capnp

@@ -0,0 +1,14 @@
+@0x9d0356c942e754d3;
+struct UserCreate {
+    name     @0 :Text;
+    email    @1 :Text;
+    age      @2 :UInt16;
+}
+
+struct UserGet {
+    id         @0 :UInt32;
+    name       @1 :Text;
+    email      @2 :Text;
+    age        @3 :UInt16;
+    isAdmin    @4 :Bool;
+}