feat: event and webhook on keylist update stored

Signed-off-by: Daniel Bluhm <dbluhm@pm.me>

Author: dbluhm

aries_cloudagent/admin/server.py

@@ -52,6 +52,7 @@
     "acapy::actionmenu::received": "actionmenu",
     "acapy::actionmenu::get-active-menu": "get-active-menu",
     "acapy::actionmenu::perform-menu-action": "perform-menu-action",
+    "acapy::keylist::updated": "keylist",
 }
 
 

aries_cloudagent/core/event_bus.py

@@ -15,6 +15,7 @@
     Optional,
     Pattern,
     TYPE_CHECKING,
+    Tuple,
 )
 from functools import partial
 
@@ -193,7 +194,7 @@ class MockEventBus(EventBus):
     def __init__(self):
         """Initialize MockEventBus."""
         super().__init__()
-        self.events = []
+        self.events: List[Tuple[Profile, Event]] = []
 
     async def notify(self, profile: "Profile", event: Event):
         """Append the event to MockEventBus.events."""

aries_cloudagent/protocols/coordinate_mediation/v1_0/manager.py

@@ -60,6 +60,7 @@ class MediationManager:
     SET_TO_DEFAULT_ON_GRANTED = "set_to_default_on_granted"
     METADATA_KEY = "mediation"
     METADATA_ID = "id"
+    KEYLIST_UPDATED_EVENT = "acapy::keylist::updated"
 
     def __init__(self, profile: Profile):
         """Initialize Mediation Manager.
@@ -534,63 +535,74 @@ async def store_update_results(
             session: An active profile session
 
         """
-        session = await self._profile.session()
         to_save: Sequence[RouteRecord] = []
         to_remove: Sequence[RouteRecord] = []
-        for updated in results:
-            if updated.result != KeylistUpdated.RESULT_SUCCESS:
-                # TODO better handle different results?
-                LOGGER.warning(
-                    "Keylist update failure: %s(%s): %s",
-                    updated.action,
-                    updated.recipient_key,
-                    updated.result,
-                )
-                continue
-            if updated.action == KeylistUpdateRule.RULE_ADD:
-                # Multi-tenancy uses route record for internal relaying of wallets
-                # So the record could already exist. We update in that case
-                try:
-                    record = await RouteRecord.retrieve_by_recipient_key(
-                        session, updated.recipient_key
-                    )
-                    record.connection_id = connection_id
-                    record.role = RouteRecord.ROLE_CLIENT
-                except StorageNotFoundError:
-                    record = RouteRecord(
-                        role=RouteRecord.ROLE_CLIENT,
-                        recipient_key=updated.recipient_key,
-                        connection_id=connection_id,
-                    )
-                to_save.append(record)
-            elif updated.action == KeylistUpdateRule.RULE_REMOVE:
-                try:
-                    records = await RouteRecord.query(
-                        session,
-                        {
-                            "role": RouteRecord.ROLE_CLIENT,
-                            "connection_id": connection_id,
-                            "recipient_key": updated.recipient_key,
-                        },
-                    )
-                except StorageNotFoundError as err:
-                    LOGGER.error(
-                        "No route found while processing keylist update response: %s",
-                        err,
+
+        async with self._profile.session() as session:
+            for updated in results:
+                if updated.result != KeylistUpdated.RESULT_SUCCESS:
+                    # TODO better handle different results?
+                    LOGGER.warning(
+                        "Keylist update failure: %s(%s): %s",
+                        updated.action,
+                        updated.recipient_key,
+                        updated.result,
                     )
-                else:
-                    if len(records) > 1:
+                    continue
+                if updated.action == KeylistUpdateRule.RULE_ADD:
+                    # Multi-tenancy uses route record for internal relaying of wallets
+                    # So the record could already exist. We update in that case
+                    try:
+                        record = await RouteRecord.retrieve_by_recipient_key(
+                            session, updated.recipient_key
+                        )
+                        record.connection_id = connection_id
+                        record.role = RouteRecord.ROLE_CLIENT
+                    except StorageNotFoundError:
+                        record = RouteRecord(
+                            role=RouteRecord.ROLE_CLIENT,
+                            recipient_key=updated.recipient_key,
+                            connection_id=connection_id,
+                        )
+                    to_save.append(record)
+                elif updated.action == KeylistUpdateRule.RULE_REMOVE:
+                    try:
+                        records = await RouteRecord.query(
+                            session,
+                            {
+                                "role": RouteRecord.ROLE_CLIENT,
+                                "connection_id": connection_id,
+                                "recipient_key": updated.recipient_key,
+                            },
+                        )
+                    except StorageNotFoundError as err:
                         LOGGER.error(
-                            f"Too many ({len(records)}) routes found "
-                            "while processing keylist update response"
+                            "No route found while processing keylist update response: %s",
+                            err,
                         )
-                    record = records[0]
-                    to_remove.append(record)
-
-        for record_for_saving in to_save:
-            await record_for_saving.save(session, reason="Route successfully added.")
-        for record_for_removal in to_remove:
-            await record_for_removal.delete_record(session)
+                    else:
+                        if len(records) > 1:
+                            LOGGER.error(
+                                f"Too many ({len(records)}) routes found "
+                                "while processing keylist update response"
+                            )
+                        record = records[0]
+                        to_remove.append(record)
+
+            for record_for_saving in to_save:
+                await record_for_saving.save(
+                    session, reason="Route successfully added."
+                )
+            for record_for_removal in to_remove:
+                await record_for_removal.delete_record(session)
+
+        await self._profile.notify(
+            self.KEYLIST_UPDATED_EVENT,
+            {
+                "connection_id": connection_id,
+                "updated": [update.serialize() for update in results],
+            },
+        )
 
     async def get_my_keylist(
         self, connection_id: Optional[str] = None

aries_cloudagent/protocols/coordinate_mediation/v1_0/tests/test_mediation_manager.py

@@ -1,11 +1,14 @@
 """Test MediationManager."""
 import logging
+from typing import AsyncIterable, Iterable
 
 import pytest
 
 from asynctest import mock as async_mock
 
 from .....core.profile import Profile, ProfileSession
+from .....core.in_memory import InMemoryProfile
+from .....core.event_bus import EventBus, MockEventBus
 from .....connections.models.conn_record import ConnRecord
 from .....messaging.request_context import RequestContext
 from .....storage.error import StorageNotFoundError
@@ -36,29 +39,32 @@
 
 
 @pytest.fixture
-async def profile() -> Profile:
+def profile() -> Iterable[Profile]:
     """Fixture for profile used in tests."""
     # pylint: disable=W0621
-    context = RequestContext.test_context()
-    context.message_receipt = MessageReceipt(sender_verkey=TEST_VERKEY)
-    context.connection_record = ConnRecord(connection_id=TEST_CONN_ID)
-    yield context.profile
+    yield InMemoryProfile.test_profile(bind={EventBus: MockEventBus()})
 
 
 @pytest.fixture
-async def session(profile) -> ProfileSession:  # pylint: disable=W0621
+def mock_event_bus(profile: Profile):
+    yield profile.inject(EventBus)
+
+
+@pytest.fixture
+async def session(profile) -> AsyncIterable[ProfileSession]:  # pylint: disable=W0621
     """Fixture for profile session used in tests."""
-    yield await profile.session()
+    async with profile.session() as session:
+        yield session
 
 
 @pytest.fixture
-async def manager(profile) -> MediationManager:  # pylint: disable=W0621
+def manager(profile) -> Iterable[MediationManager]:  # pylint: disable=W0621
     """Fixture for manager used in tests."""
     yield MediationManager(profile)
 
 
 @pytest.fixture
-def record() -> MediationRecord:
+def record() -> Iterable[MediationRecord]:
     """Fixture for record used in tests."""
     yield MediationRecord(
         state=MediationRecord.STATE_GRANTED, connection_id=TEST_CONN_ID
@@ -71,7 +77,7 @@ class TestMediationManager:  # pylint: disable=R0904,W0621
     async def test_create_manager_no_profile(self):
         """test_create_manager_no_profile."""
         with pytest.raises(MediationManagerError):
-            await MediationManager(None)
+            MediationManager(None)
 
     async def test_create_did(self, manager, session):
         """test_create_did."""
@@ -363,7 +369,12 @@ async def test_add_remove_key_mix(self, manager):
         assert update.updates[0].recipient_key == TEST_VERKEY
         assert update.updates[1].recipient_key == TEST_ROUTE_VERKEY
 
-    async def test_store_update_results(self, session, manager):
+    async def test_store_update_results(
+        self,
+        session: ProfileSession,
+        manager: MediationManager,
+        mock_event_bus: MockEventBus,
+    ):
         """test_store_update_results."""
         await RouteRecord(
             role=RouteRecord.ROLE_CLIENT,
@@ -383,6 +394,12 @@ async def test_store_update_results(self, session, manager):
             ),
         ]
         await manager.store_update_results(TEST_CONN_ID, results)
+        assert mock_event_bus.events
+        assert mock_event_bus.events[0][1].topic == manager.KEYLIST_UPDATED_EVENT
+        assert mock_event_bus.events[0][1].payload == {
+            "connection_id": TEST_CONN_ID,
+            "updated": [result.serialize() for result in results],
+        }
         routes = await RouteRecord.query(session)
 
         assert len(routes) == 1