23 lines
688 B
Python
23 lines
688 B
Python
from app.agent_event_bus import AgentEventBus
|
|
from 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
|