feat: include thread id in keylist update recv webhooks

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

Author: dbluhm

aries_cloudagent/protocols/coordinate_mediation/v1_0/handlers/keylist_update_handler.py

@@ -32,6 +32,7 @@ async def handle(self, context: RequestContext, responder: BaseResponder):
                     session, context.connection_record.connection_id
                 )
             response = await mgr.update_keylist(record, updates=context.message.updates)
+            response.assign_thread_from(context.message)
             await responder.send_reply(response)
         except (StorageNotFoundError, MediationNotGrantedError):
             reply = CMProblemReport(

aries_cloudagent/protocols/coordinate_mediation/v1_0/handlers/keylist_update_response_handler.py

@@ -25,3 +25,6 @@ async def handle(self, context: RequestContext, responder: BaseResponder):
         await mgr.store_update_results(
             context.connection_record.connection_id, context.message.updated
         )
+        await mgr.notify_keylist_updated(
+            context.connection_record.connection_id, context.message
+        )

aries_cloudagent/protocols/coordinate_mediation/v1_0/handlers/tests/test_keylist_update_response_handler.py

@@ -46,6 +46,9 @@ async def test_handler(self):
         handler, responder = KeylistUpdateResponseHandler(), MockResponder()
         with async_mock.patch.object(
             MediationManager, "store_update_results"
-        ) as mock_method:
+        ) as mock_store, async_mock.patch.object(
+            MediationManager, "notify_keylist_updated"
+        ) as mock_notify:
             await handler.handle(self.context, responder)
-            mock_method.assert_called_once_with(TEST_CONN_ID, self.updated)
+            mock_store.assert_called_once_with(TEST_CONN_ID, self.updated)
+            mock_notify.assert_called_once_with(TEST_CONN_ID, self.context.message)

aries_cloudagent/protocols/coordinate_mediation/v1_0/manager.py

@@ -596,11 +596,16 @@ async def store_update_results(
             for record_for_removal in to_remove:
                 await record_for_removal.delete_record(session)
 
+    async def notify_keylist_updated(
+        self, connection_id: str, response: KeylistUpdateResponse
+    ):
+        """Notify of keylist update response received."""
         await self._profile.notify(
             self.KEYLIST_UPDATED_EVENT,
             {
                 "connection_id": connection_id,
-                "updated": [update.serialize() for update in results],
+                "thread_id": response._thread_id,
+                "updated": [update.serialize() for update in response.updated],
             },
         )
 

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

@@ -2,21 +2,15 @@
 import logging
 from typing import AsyncIterable, Iterable
 
-import pytest
-
 from asynctest import mock as async_mock
+import pytest
 
-from .....core.profile import Profile, ProfileSession
-from .....core.in_memory import InMemoryProfile
+from .. import manager as test_module
 from .....core.event_bus import EventBus, MockEventBus
-from .....connections.models.conn_record import ConnRecord
-from .....messaging.request_context import RequestContext
+from .....core.in_memory import InMemoryProfile
+from .....core.profile import Profile, ProfileSession
 from .....storage.error import StorageNotFoundError
-from .....transport.inbound.receipt import MessageReceipt
-
 from ....routing.v1_0.models.route_record import RouteRecord
-
-from .. import manager as test_module
 from ..manager import (
     MediationAlreadyExists,
     MediationManager,
@@ -25,12 +19,14 @@
 )
 from ..messages.inner.keylist_update_rule import KeylistUpdateRule
 from ..messages.inner.keylist_updated import KeylistUpdated
+from ..messages.keylist_update_response import KeylistUpdateResponse
 from ..messages.mediate_deny import MediationDeny
 from ..messages.mediate_grant import MediationGrant
 from ..messages.mediate_request import MediationRequest
 from ..models.mediation_record import MediationRecord
 
 TEST_CONN_ID = "conn-id"
+TEST_THREAD_ID = "thread-id"
 TEST_ENDPOINT = "https://example.com"
 TEST_VERKEY = "3Dn1SJNPaCXcvvJvSbsFWP2xaCjMom3can8CQNhWrTRx"
 TEST_ROUTE_VERKEY = "9WCgWKUaAJj3VWxxtzvvMQN3AoFxoBtBDo9ntwJnVVCC"
@@ -373,7 +369,6 @@ async def test_store_update_results(
         self,
         session: ProfileSession,
         manager: MediationManager,
-        mock_event_bus: MockEventBus,
     ):
         """test_store_update_results."""
         await RouteRecord(
@@ -394,12 +389,6 @@ async def test_store_update_results(
             ),
         ]
         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
@@ -484,6 +473,34 @@ async def test_store_update_results_errors(self, caplog, manager):
         assert "server_error" in caplog.text
         print(caplog.text)
 
+    async def test_notify_keylist_updated(
+        self, manager: MediationManager, mock_event_bus: MockEventBus
+    ):
+        """test notify_keylist_updated."""
+        response = KeylistUpdateResponse(
+            updated=[
+                KeylistUpdated(
+                    recipient_key=TEST_ROUTE_VERKEY,
+                    action=KeylistUpdateRule.RULE_ADD,
+                    result=KeylistUpdated.RESULT_SUCCESS,
+                ),
+                KeylistUpdated(
+                    recipient_key=TEST_VERKEY,
+                    action=KeylistUpdateRule.RULE_REMOVE,
+                    result=KeylistUpdated.RESULT_SUCCESS,
+                ),
+            ],
+        )
+        response.assign_thread_id(TEST_THREAD_ID)
+        await manager.notify_keylist_updated(TEST_CONN_ID, response)
+        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,
+            "thread_id": TEST_THREAD_ID,
+            "updated": [result.serialize() for result in response.updated],
+        }
+
     async def test_get_my_keylist(self, session, manager):
         """test_get_my_keylist."""
         await RouteRecord(