32 lines
1.2 KiB
Python
32 lines
1.2 KiB
Python
from active_runtime.app.agent_event_bus import AgentEventBus
|
|
from active_runtime.app.agent_event_dashboard import AgentEventDashboard
|
|
|
|
|
|
def test_unified_event_flow():
|
|
bus = AgentEventBus(root="/tmp/aios-agent-events-test")
|
|
event = bus.publish(
|
|
kind="gpt_inbox",
|
|
source="openclaw",
|
|
target="custom-gpt",
|
|
summary="execution report",
|
|
payload={"status": "done"},
|
|
)
|
|
|
|
assert event["status"] == "pending"
|
|
assert len(bus.list_events(kind="gpt_inbox")) == 1
|
|
|
|
done = bus.transition(event["id"], "completed", "custom-gpt")
|
|
assert done["status"] == "completed"
|
|
|
|
dashboard = AgentEventDashboard(bus).overview()
|
|
assert dashboard["audit"]["enabled"] is True
|
|
|
|
|
|
def test_target_changes_event_identity(tmp_path):
|
|
bus = AgentEventBus(tmp_path)
|
|
first = bus.publish('test', 'same', 'user-a', 'same summary', {'value': 1})
|
|
second = bus.publish('test', 'same', 'user-b', 'same summary', {'value': 1})
|
|
first_id = first.get('id') if isinstance(first, dict) else getattr(first, 'id', first)
|
|
second_id = second.get('id') if isinstance(second, dict) else getattr(second, 'id', second)
|
|
assert first_id != second_id
|