76 lines
2.8 KiB
Python
76 lines
2.8 KiB
Python
"""
|
|
Tests for directed-message delivery on the MessageBus.
|
|
|
|
Regression guard for the dead-letter bug: directed messages used to be appended
|
|
to an internal queue that nothing ever drained. They must now be delivered live
|
|
to the recipient agent — task-bearing payloads become tasks, notifications reach
|
|
handle_message, and an absent recipient still retains the message in the pull
|
|
queue (never silently lost).
|
|
|
|
Run:
|
|
python -m unittest discover -s tests
|
|
"""
|
|
import unittest
|
|
|
|
from core.agent import Agent
|
|
from core.message_bus import message_bus
|
|
from core.types import MessageType
|
|
|
|
|
|
class Recorder(Agent):
|
|
"""Minimal concrete agent that records what it was handed."""
|
|
def __init__(self, agent_id):
|
|
super().__init__(agent_id, "test", "recorder") # registers with the bus
|
|
self.notifications = []
|
|
|
|
async def handle_task(self, task):
|
|
return {"ok": True}
|
|
|
|
async def handle_message(self, message):
|
|
self.notifications.append(message)
|
|
|
|
|
|
class TestDirectedDelivery(unittest.IsolatedAsyncioTestCase):
|
|
async def test_task_message_enqueued_as_task(self):
|
|
agent = Recorder("REC_TASK")
|
|
try:
|
|
await message_bus.send_to_agent(
|
|
sender="X", recipient="REC_TASK",
|
|
message_type=MessageType.AGENT_TASK,
|
|
payload={"task_type": "do_thing", "booking_id": "b1"},
|
|
)
|
|
task = agent._task_queue.get_nowait() # delivered, not dead-lettered
|
|
self.assertEqual(task.task_type, "do_thing")
|
|
self.assertEqual(task.data["booking_id"], "b1")
|
|
finally:
|
|
message_bus.unregister_agent("REC_TASK")
|
|
|
|
async def test_notification_goes_to_handle_message(self):
|
|
agent = Recorder("REC_NOTIF")
|
|
try:
|
|
await message_bus.send_to_agent(
|
|
sender="X", recipient="REC_NOTIF",
|
|
message_type=MessageType.EXCEPTION_DETECTED,
|
|
payload={"order_id": "b9"}, # no task_type
|
|
)
|
|
self.assertEqual(len(agent.notifications), 1)
|
|
self.assertEqual(agent.notifications[0].message_type, MessageType.EXCEPTION_DETECTED)
|
|
self.assertTrue(agent._task_queue.empty()) # a notification is not a task
|
|
finally:
|
|
message_bus.unregister_agent("REC_NOTIF")
|
|
|
|
async def test_absent_recipient_retained_in_pull_queue(self):
|
|
# Nobody registered under this id -> message kept, not lost.
|
|
await message_bus.send_to_agent(
|
|
sender="X", recipient="NOBODY_HOME",
|
|
message_type=MessageType.AGENT_TASK,
|
|
payload={"task_type": "x"},
|
|
)
|
|
msgs = await message_bus.get_messages("NOBODY_HOME")
|
|
self.assertEqual(len(msgs), 1)
|
|
self.assertEqual(msgs[0].recipient, "NOBODY_HOME")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|