File size: 2,517 Bytes
31c8075 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 | import pytest
from arena.event_bus import EventBus
from arena.run_journal import InvalidRunStatusTransition, RunJournal
from arena.run_models import RunEvent, RunRequest, RunTask
from arena.run_store import InMemoryRunStore
def test_run_journal_owns_status_transitions_and_streams() -> None:
journal = RunJournal(run_store=InMemoryRunStore(), event_bus=EventBus())
access = journal.create_run(RunRequest(prompt="Build a CLI"))
queue = journal.subscribe_events(access.run.id)
journal.transition(access.run.id, "planning")
journal.publish(access.run.id, {"kind": "run_started", "run_id": access.run.id})
view = journal.get_run(access.run.id)
assert view.status == "planning"
assert queue.get(timeout=1)["kind"] == "run_started"
def test_run_journal_rejects_invalid_status_transition() -> None:
journal = RunJournal(run_store=InMemoryRunStore(), event_bus=EventBus())
access = journal.create_run(RunRequest(prompt="Build a CLI"))
with pytest.raises(InvalidRunStatusTransition):
journal.transition(access.run.id, "completed")
def test_run_journal_persists_runtime_task_and_event_updates() -> None:
journal = RunJournal(run_store=InMemoryRunStore(), event_bus=EventBus())
access = journal.create_run(RunRequest(prompt="Build a CLI"))
task = RunTask(id="task-1", title="Build codebase", status="active")
event = RunEvent(id="event-1", message="Coordinator started work")
journal.publish(access.run.id, {"kind": "task_created", "task": task.model_dump()})
journal.publish(access.run.id, {"kind": "event", "event": event.model_dump(mode="json")})
view = journal.get_run(access.run.id)
assert view.tasks == [task]
assert [item.id for item in view.events] == ["event-1"]
def test_run_journal_cancels_active_tasks_and_closes_stream() -> None:
journal = RunJournal(run_store=InMemoryRunStore(), event_bus=EventBus())
access = journal.create_run(RunRequest(prompt="Build a CLI"))
queue = journal.subscribe_events(access.run.id)
journal.transition(access.run.id, "working")
journal.publish(
access.run.id,
{
"kind": "task_created",
"task": RunTask(id="task-1", title="Build codebase", status="active").model_dump(),
},
)
view = journal.cancel_run(access.run.id)
assert view.status == "interrupted"
assert view.tasks[0].status == "interrupted"
assert queue.get(timeout=1)["kind"] == "task_created"
assert queue.get(timeout=1) is None
|