"""Tests for the status dashboard DATA LAYER (``agent_team.status_page``).
After the WebUI makeover, ``status_page`` no longer renders HTML — the dashboard
is a React/Vite SPA served by ``agent_team.dashboard`` (see ``test_dashboard.py``
for the FastAPI endpoints). What remains here is the read-only reader
(:func:`build_snapshot`) and the ``/api/state`` payload builder
(:func:`snapshot_to_dict`), exercised against fabricated :class:`Snapshot`
objects and a temp SQLite ledger seeded with real rows. ``conftest.py`` puts
``agent-team/`` on ``sys.path``.
"""
from __future__ import annotations
import json
import sqlite3
from contextlib import closing
from pathlib import Path
import pytest
from agent_team.status_page import (
STAGES,
Snapshot,
TaskView,
build_snapshot,
snapshot_to_dict,
)
def _sample_snapshot() -> Snapshot:
return Snapshot(
generated_at="2026-06-23 12:00:00 UTC",
db_path="/home/adam/orchestrator/agent-team/state/agent_team.sqlite",
ok=True,
tasks=[
TaskView(
thread_id="abc123def456",
short_id="abc123de",
task="remediate CVE-2026-0001 in payments-svc",
current_phase="clarify",
status="waiting_human",
waiting=True,
waiting_since="2026-06-23T11:50:00+00:00",
),
TaskView(
thread_id="ffff0000aaaa",
short_id="ffff0000",
task="add a smoke-test file",
current_phase="plan",
status="active",
waiting=False,
),
TaskView(
thread_id="dead0000beef",
short_id="dead0000",
task="stalled task",
current_phase="parked",
status="parked",
waiting=False,
),
],
question_counts={"open": 1, "answered": 4, "expired": 1},
recent_spend=[
{
"recorded_at": "2026-06-23T11:00:00+00:00",
"stage": "plan",
"model": "claude-opus-4",
"usd_cost": 0.1234,
}
],
spend_total_usd=1.5,
budget_available=True,
)
# --- read-only reader tests against a temp seeded SQLite ledger. -------------
def seed_ledger(db_path: Path) -> None:
"""Create the agent-team schema and seed a couple of rows.
Seeds one open pending question and one budget row via the foundation
schema, and one LangGraph checkpoint per thread via the real ``SqliteSaver``
so the reader's checkpoint path is exercised faithfully. Imported by
``test_dashboard.py`` as well.
"""
from agent_team.db.schema import init_db
from langgraph.checkpoint.base import empty_checkpoint
from langgraph.checkpoint.sqlite import SqliteSaver
init_db(db_path)
with closing(sqlite3.connect(str(db_path))) as conn:
conn.execute(
"INSERT INTO pending_questions "
"(question_id, thread_id, turn, status, transport, posted_at) "
"VALUES (?, ?, ?, 'open', 'slack', ?)",
("q1", "thread-waiting", 0, "2026-06-23T11:50:00+00:00"),
)
conn.execute(
"INSERT INTO budget_ledger "
"(thread_id, stage, model, billing_mode, usd_cost, recorded_at, day_bucket) "
"VALUES (?, ?, ?, ?, ?, ?, ?)",
(
"thread-waiting",
"plan",
"claude-opus-4",
"subscription",
0.25,
"2026-06-23T11:00:00+00:00",
"2026-06-23",
),
)
conn.commit()
with closing(sqlite3.connect(str(db_path), check_same_thread=False)) as conn:
saver = SqliteSaver(conn)
for tid, values in (
(
"thread-waiting",
{
"task": "remediate CVE",
"current_phase": "clarify",
"status": "waiting_human",
},
),
(
"thread-active",
{
"task": "add a file",
"current_phase": "plan",
"status": "active",
},
),
):
ck = empty_checkpoint()
ck["channel_values"] = values
saver.put(
{"configurable": {"thread_id": tid, "checkpoint_ns": ""}},
ck,
{},
{},
)
def test_build_snapshot_reads_seeded_ledger(tmp_path: Path) -> None:
db = tmp_path / "agent_team.sqlite"
seed_ledger(db)
snap = build_snapshot(db)
assert snap.ok is True
assert snap.error is None
by_id = {t.thread_id: t for t in snap.tasks}
assert set(by_id) == {"thread-waiting", "thread-active"}
waiting = by_id["thread-waiting"]
assert waiting.task == "remediate CVE"
assert waiting.current_phase == "clarify"
assert waiting.status == "waiting_human"
assert waiting.waiting is True
assert waiting.waiting_since == "2026-06-23T11:50:00+00:00"
active = by_id["thread-active"]
assert active.waiting is False
assert active.status == "active"
assert snap.active_count == 2 # active + waiting_human
assert snap.parked_count == 0
assert snap.question_counts.get("open") == 1
assert snap.budget_available is True
assert snap.spend_total_usd == pytest.approx(0.25)
assert snap.recent_spend[0]["model"] == "claude-opus-4"
def test_build_snapshot_missing_db_is_fail_safe(tmp_path: Path) -> None:
snap = build_snapshot(tmp_path / "does-not-exist.sqlite")
assert snap.ok is False
assert snap.error and "not found" in snap.error
def test_build_snapshot_never_writes(tmp_path: Path) -> None:
"""The reader must open mode=ro; it can never create the DB file."""
missing = tmp_path / "nope.sqlite"
build_snapshot(missing)
assert not missing.exists() # mode=ro did not create it
# --- /api/state JSON payload (snapshot_to_dict) contract ---------------------
def test_snapshot_to_dict_shape_and_grouping() -> None:
"""The /api/state payload keeps its keys + correct stage grouping."""
payload = snapshot_to_dict(_sample_snapshot())
assert payload["ok"] is True
assert set(payload) >= {
"ok",
"generated_at",
"stages",
"tasks",
"waiting",
"question_counts",
"summary",
"budget",
}
assert [s["key"] for s in payload["stages"]] == [s.key for s in STAGES]
by_key = {s["key"]: s for s in payload["stages"]}
clarify = by_key["clarify"]
assert clarify["count"] == 1
assert clarify["state"] == "awaiting_human"
assert clarify["tasks"][0]["short_id"] == "abc123de"
assert by_key["plan"]["state"] == "active"
assert by_key["plan"]["count"] == 1
# The parked task maps onto no pipeline node.
assert all(
t["thread_id"] != "dead0000beef" for s in payload["stages"] for t in s["tasks"]
)
assert by_key["intake"]["state"] == "idle"
assert by_key["intake"]["count"] == 0
assert payload["summary"]["active"] == 2
assert payload["summary"]["waiting"] == 1
assert payload["summary"]["open_questions"] == 1
assert payload["summary"]["total"] == 3
def test_snapshot_to_dict_not_ok_is_serializable() -> None:
snap = Snapshot(generated_at="now", db_path="/x.sqlite", ok=False, error="boom")
payload = snapshot_to_dict(snap)
assert payload["ok"] is False
assert payload["error"] == "boom"
assert [s["key"] for s in payload["stages"]] == [s.key for s in STAGES]
assert all(s["count"] == 0 for s in payload["stages"])
assert json.loads(json.dumps(payload))["ok"] is False
def test_snapshot_to_dict_carries_raw_description_for_textcontent() -> None:
"""The JSON sidecar carries the raw description; the SPA renders it safely."""
snap = Snapshot(
generated_at="now",
db_path="/x.sqlite",
ok=True,
tasks=[
TaskView(
thread_id="x",
short_id="x",
task="",
current_phase="plan",
status="active",
)
],
)
payload = snapshot_to_dict(snap)
parsed = json.loads(json.dumps(payload))
assert parsed["tasks"][0]["task"] == ""