feat: updates for caboodle demo

examples/10-cleaning-demo/cleaner.py

@@ -44,8 +44,8 @@ async def startup():
 
     await Availability.create(
         provider=provider,
-        time_start=utc.localize(datetime.fromisoformat("2022-12-31 10:00:00")),
-        time_end=utc.localize(datetime.fromisoformat("2022-12-31 22:00:00")),
+        time_start=utc.localize(datetime.fromisoformat("2022-01-31 00:00:00")),
+        time_end=utc.localize(datetime.fromisoformat("2023-02-01 00:00:00")),
         max_distance=10,
         min_hourly_price=5,
     )

examples/10-cleaning-demo/protocols/cleaning/__init__.py

@@ -78,7 +78,7 @@ async def handle_query_request(ctx: Context, sender: str, msg: ServiceRequest):
 @cleaning_proto.on_message(model=ServiceBooking, replies=BookingResponse)
 async def handle_book_request(ctx: Context, sender: str, msg: ServiceBooking):
 
-    provider = await Provider.get(name=ctx.name)
+    provider = await Provider.filter(name=ctx.name).first()
     availability = await Availability.get(provider=provider)
     services = [int(service.type) for service in await provider.services]
 

src/nexus/agent.py

@@ -14,7 +14,7 @@
     MsgDigest,
 )
 from nexus.crypto import Identity, derive_key_from_seed, is_user_address
-from nexus.dispatch import Sink, dispatcher
+from nexus.dispatch import Sink, dispatcher, JsonStr
 from nexus.models import Model, ErrorMessage
 from nexus.protocol import Protocol
 from nexus.resolver import Resolver, AlmanacResolver
@@ -57,7 +57,7 @@ def __init__(
         self._name = name
         self._intervals: List[Tuple[float, Any]] = []
         self._port = port if port is not None else 8000
-        self._background_tasks = set()
+        self._background_tasks: Set[asyncio.Task] = set()
         self._resolver = resolve if resolve is not None else AlmanacResolver()
         self._loop = asyncio.get_event_loop_policy().get_event_loop()
         if seed is None:
@@ -111,11 +111,6 @@ def __init__(
         # register with the dispatcher
         self._dispatcher.register(self.address, self)
 
-        # start the background message queue processor
-        task = self._loop.create_task(self._process_message_queue())
-        self._background_tasks.add(task)
-        task.add_done_callback(self._background_tasks.discard)
-
         self._server = ASGIServer(self._port, self._loop, self._queries)
 
     @property
@@ -287,7 +282,7 @@ def include(self, protocol: Protocol):
         if protocol.digest is not None:
             self.protocols[protocol.canonical_name] = protocol.digest
 
-    async def handle_message(self, sender, schema_digest: str, message: Any):
+    async def handle_message(self, sender, schema_digest: str, message: JsonStr):
         await self._message_queue.put((schema_digest, sender, message))
 
     async def startup(self):
@@ -302,6 +297,11 @@ def setup(self):
         # register the internal agent protocol
         self.include(self._protocol)
 
+        # start the background message queue processor
+        task = self._loop.create_task(self._process_message_queue())
+        self._background_tasks.add(task)
+        task.add_done_callback(self._background_tasks.discard)
+
         # start the contract registration update loop
         self._loop.create_task(
             _run_interval(self.register, self._ctx, REG_UPDATE_INTERVAL_SECONDS)
@@ -321,7 +321,7 @@ async def _process_message_queue(self):
             schema_digest, sender, message = await self._message_queue.get()
 
             # lookup the model definition
-            model_class = self._models.get(schema_digest)
+            model_class: Model = self._models.get(schema_digest)
             if model_class is None:
                 continue
 

src/nexus/asgi.py

@@ -63,7 +63,7 @@ async def __call__(self, scope, receive, send):
             return
 
         headers = dict(scope.get("headers", {}))
-        if headers[b"content-type"] != b"application/json":
+        if b"application/json" not in headers[b"content-type"]:
             await send(
                 {
                     "type": "http.response.start",
@@ -143,14 +143,15 @@ async def __call__(self, scope, receive, send):
             return
 
         await dispatcher.dispatch(
-            env.sender, env.target, env.protocol, env.decode_payload()
+            env.sender, env.target, env.protocol, json.dumps(env.decode_payload())
         )
 
         # wait for any queries to be resolved
         if expects_response:
             response_msg: Model = await self._queries[env.sender]
-            if datetime.now() > datetime.fromtimestamp(env.expires):
-                response_msg = ErrorMessage("Query envelope expired")
+            if env.expires is not None:
+                if datetime.now() > datetime.fromtimestamp(env.expires):
+                    response_msg = ErrorMessage("Query envelope expired")
             sender = env.target
             response = enclose_response(response_msg, sender, env.session)
         else:

src/nexus/context.py

@@ -1,4 +1,5 @@
 import asyncio
+import json
 import logging
 import uuid
 from dataclasses import dataclass
@@ -10,7 +11,7 @@
 from cosmpy.aerial.wallet import LocalWallet
 
 from nexus.config import DEFAULT_ENVELOPE_TIMEOUT_SECONDS
-from nexus.crypto import Identity, is_user_address
+from nexus.crypto import Identity
 from nexus.dispatch import dispatcher
 from nexus.envelope import Envelope
 from nexus.models import Model, ErrorMessage
@@ -105,15 +106,12 @@ async def send(
         # handle local dispatch of messages
         if dispatcher.contains(destination):
             await dispatcher.dispatch(
-                self.address, destination, schema_digest, json_message
+                self.address, destination, schema_digest, json.dumps(json_message)
             )
             return
 
         # handle queries waiting for a response
-        if is_user_address(destination):
-            if destination not in self._queries:
-                logging.exception(f"Unable to resolve query to user {destination}")
-                return
+        if destination in self._queries:
             self._queries[destination].set_result(message)
             del self._queries[destination]
             return

src/nexus/dispatch.py

@@ -1,10 +1,12 @@
 from abc import ABC, abstractmethod
-from typing import Dict, Set, Any
+from typing import Dict, Set
+
+JsonStr = str
 
 
 class Sink(ABC):
     @abstractmethod
-    async def handle_message(self, sender: str, schema_digest: str, message: Any):
+    async def handle_message(self, sender: str, schema_digest: str, message: JsonStr):
         pass
 
 
@@ -26,7 +28,7 @@ def contains(self, address: str) -> bool:
         return address in self._sinks
 
     async def dispatch(
-        self, sender: str, destination: str, schema_digest: str, message: Any
+        self, sender: str, destination: str, schema_digest: str, message: JsonStr
     ):
         for handler in self._sinks.get(destination, set()):
             await handler.handle_message(sender, schema_digest, message)

src/nexus/query.py

@@ -73,5 +73,5 @@ def enclose_response(message: Model, sender: str, session: str) -> dict:
         session=session,
         protocol=Model.build_schema_digest(message),
     )
-    response_env.encode_payload(message.json())
+    response_env.encode_payload(message.dict())
     return response_env.json()

tests/test_agent_registration.py

@@ -0,0 +1,46 @@
+# pylint: disable=protected-access
+import unittest
+
+from nexus import Agent
+from nexus.setup import fund_agent_if_low
+
+
+class TestVerify(unittest.TestCase):
+    def test_agent_registration(self):
+
+        agent = Agent(name="alice")
+
+        reg_fee = "500000000000000000atestfet"
+
+        fund_agent_if_low(agent.wallet.address())
+
+        sequence = agent.get_registration_sequence()
+
+        signature = agent._identity.sign_registration(
+            agent._reg_contract.address, agent.get_registration_sequence()
+        )
+
+        msg = {
+            "register": {
+                "record": {
+                    "service": {
+                        "protocols": [],
+                        "endpoints": [{"url": agent._endpoint, "weight": 1}],
+                    }
+                },
+                "signature": signature,
+                "sequence": sequence,
+                "agent_address": agent.address,
+            }
+        }
+
+        transaction = agent._reg_contract.execute(msg, agent.wallet, funds=reg_fee)
+        transaction.wait_to_complete()
+
+        is_registered = agent.registration_status()
+
+        self.assertEqual(is_registered, True, "Registration failed")
+
+
+if __name__ == "__main__":
+    unittest.main()