Adam Moussa
4e75a0bf93
- render_html includes the inline SVG map + inline fetch('/api/state') poller
+ tooltip mount + live clock, stays offline (no CDN), keeps the <noscript>
meta-refresh fallback, and renders the map even on a not-ok snapshot.
- snapshot_to_dict: expected top-level keys, one entry per STAGE in order,
per-phase grouping, awaiting-human classification (open pending_question),
per-node model/agent role labels, summary counts, not-ok serializability.
- Descriptions escaped in BOTH the HTML and the JSON-in-script seed
(</script><script> breakout neutralised; \u003c form present); the raw
/api/state JSON round-trips the description for textContent rendering.
- End-to-end snapshot_to_dict over the seeded ledger.
416 lines
14 KiB
Python
416 lines
14 KiB
Python
"""Tests for the LAN-only read-only status dashboard (``agent_team.status_page``).
|
|
|
|
No live socket is bound: :func:`render_html` is exercised against fabricated
|
|
:class:`Snapshot` objects, and the read-only reader (:func:`build_snapshot`) is
|
|
exercised against a temp SQLite ledger seeded with a couple of rows (a real
|
|
``SqliteSaver`` checkpoint plus ``pending_questions`` / ``budget_ledger`` rows).
|
|
``conftest.py`` already 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,
|
|
render_html,
|
|
snapshot_to_dict,
|
|
)
|
|
|
|
# --- pure render_html tests (no DB, no socket). ------------------------------
|
|
|
|
|
|
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,
|
|
)
|
|
|
|
|
|
def test_render_html_returns_complete_document() -> None:
|
|
html_out = render_html(_sample_snapshot())
|
|
assert html_out.startswith("<!doctype html>")
|
|
assert html_out.rstrip().endswith("</html>")
|
|
# Auto-refresh + offline (no external CDN / http references).
|
|
assert '<meta http-equiv="refresh" content="10">' in html_out
|
|
assert "http://" not in html_out and "https://" not in html_out
|
|
assert "<style>" in html_out # inline CSS only
|
|
|
|
|
|
def test_render_html_shows_counts_and_waiting() -> None:
|
|
snap = _sample_snapshot()
|
|
html_out = render_html(snap)
|
|
assert snap.active_count == 2 # active + waiting_human
|
|
assert snap.parked_count == 1
|
|
assert len(snap.waiting_tasks) == 1
|
|
# The waiting task and its phase are surfaced.
|
|
assert "abc123de" in html_out
|
|
assert "remediate CVE-2026-0001 in payments-svc" in html_out
|
|
assert "Waiting on the human gate" in html_out
|
|
# Budget panel present with a total.
|
|
assert "Budget" in html_out
|
|
assert "claude-opus-4" in html_out
|
|
|
|
|
|
def test_render_html_escapes_task_descriptions() -> None:
|
|
snap = Snapshot(
|
|
generated_at="now",
|
|
db_path="/x.sqlite",
|
|
ok=True,
|
|
tasks=[
|
|
TaskView(
|
|
thread_id="x",
|
|
short_id="x",
|
|
task="<script>alert('xss')</script>",
|
|
current_phase="intake",
|
|
status="active",
|
|
)
|
|
],
|
|
)
|
|
html_out = render_html(snap)
|
|
assert "<script>alert" not in html_out
|
|
assert "<script>" in html_out
|
|
|
|
|
|
def test_render_html_no_data_page_on_not_ok() -> None:
|
|
snap = Snapshot(
|
|
generated_at="now",
|
|
db_path="/missing.sqlite",
|
|
ok=False,
|
|
error="ledger not found",
|
|
)
|
|
html_out = render_html(snap)
|
|
assert "No data" in html_out
|
|
assert "ledger not found" in html_out
|
|
assert html_out.rstrip().endswith("</html>")
|
|
|
|
|
|
def test_render_html_omits_budget_when_unavailable() -> None:
|
|
snap = Snapshot(
|
|
generated_at="now",
|
|
db_path="/x.sqlite",
|
|
ok=True,
|
|
tasks=[],
|
|
budget_available=False,
|
|
)
|
|
html_out = render_html(snap)
|
|
assert "Budget" not in html_out
|
|
|
|
|
|
# --- read-only reader tests against a temp seeded SQLite ledger. -------------
|
|
|
|
|
|
def _seed_db(db_path: Path) -> None:
|
|
"""Create the agent-team schema and seed a couple of rows.
|
|
|
|
Seeds one pending question (open) 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.
|
|
"""
|
|
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()
|
|
|
|
# Seed checkpoints for two threads via the real saver.
|
|
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_db(db)
|
|
|
|
snap = build_snapshot(db)
|
|
|
|
assert snap.ok is True
|
|
assert snap.error is None
|
|
# Two checkpointed threads enumerated.
|
|
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 # has an open pending_question
|
|
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"
|
|
|
|
# Queue summary + question counts.
|
|
assert snap.active_count == 2 # active + waiting_human
|
|
assert snap.parked_count == 0
|
|
assert snap.question_counts.get("open") == 1
|
|
|
|
# Budget read.
|
|
assert snap.budget_available is True
|
|
assert snap.spend_total_usd == pytest.approx(0.25)
|
|
assert snap.recent_spend[0]["model"] == "claude-opus-4"
|
|
|
|
# The whole thing renders without error.
|
|
html_out = render_html(snap)
|
|
assert "remediate CVE" in html_out
|
|
|
|
|
|
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
|
|
# Still renders a friendly page rather than raising.
|
|
html_out = render_html(snap)
|
|
assert "No data" in html_out
|
|
|
|
|
|
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
|
|
|
|
|
|
# --- pipeline-map + /api/state JSON contract tests ---------------------------
|
|
|
|
|
|
def test_render_html_includes_svg_map_and_poller() -> None:
|
|
"""The page must carry the inline SVG map and the inline JS poller."""
|
|
html_out = render_html(_sample_snapshot())
|
|
# Inline SVG map (hand-rolled, not fetched).
|
|
assert '<svg class="map"' in html_out
|
|
assert 'data-stage="clarify"' in html_out
|
|
# Inline poller that fetches the JSON sidecar — no external libraries.
|
|
assert "<script>" in html_out
|
|
assert "fetch('/api/state'" in html_out
|
|
assert "setInterval(poll" in html_out
|
|
# Tooltip mount + live clock the poller updates.
|
|
assert 'id="tip"' in html_out
|
|
assert 'id="clock"' in html_out
|
|
# Legend present.
|
|
assert "awaiting human" in html_out
|
|
# Still fully offline: no CDN / external references.
|
|
assert "http://" not in html_out and "https://" not in html_out
|
|
# The no-JS fallback is the meta-refresh, now inside <noscript>.
|
|
assert "<noscript>" in html_out
|
|
assert '<meta http-equiv="refresh" content="10">' in html_out
|
|
|
|
|
|
def test_render_html_renders_map_even_with_no_data() -> None:
|
|
"""A not-ok snapshot still renders the map + poller so it goes live later."""
|
|
snap = Snapshot(generated_at="now", db_path="/x.sqlite", ok=False, error="boom")
|
|
html_out = render_html(snap)
|
|
assert '<svg class="map"' in html_out
|
|
assert "fetch('/api/state'" in html_out
|
|
assert "No data" in html_out
|
|
assert html_out.rstrip().endswith("</html>")
|
|
|
|
|
|
def test_snapshot_to_dict_shape_and_grouping() -> None:
|
|
"""The /api/state payload has the expected 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",
|
|
}
|
|
# One stage entry per declared STAGE, in declared order.
|
|
assert [s["key"] for s in payload["stages"]] == [s.key for s in STAGES]
|
|
|
|
by_key = {s["key"]: s for s in payload["stages"]}
|
|
# The waiting clarify task lands on the clarify node and flags awaiting_human.
|
|
clarify = by_key["clarify"]
|
|
assert clarify["count"] == 1
|
|
assert clarify["state"] == "awaiting_human"
|
|
assert clarify["tasks"][0]["short_id"] == "abc123de"
|
|
# Each stage carries its model/agent role for the per-node label.
|
|
assert clarify["agent"] == "Claude (sub)"
|
|
assert by_key["review"]["agent"] == "GPT-4.1 (cross)"
|
|
assert by_key["build"]["agent"] == "DeepSeek (fast)"
|
|
# The active plan task lands on the plan node as 'active'.
|
|
assert by_key["plan"]["state"] == "active"
|
|
assert by_key["plan"]["count"] == 1
|
|
# The parked task maps onto the (terminal-ish) parked side — no stage owns
|
|
# the 'parked' phase, so no pipeline node claims it.
|
|
assert all(
|
|
t["thread_id"] != "dead0000beef" for s in payload["stages"] for t in s["tasks"]
|
|
)
|
|
# An idle stage with no tasks.
|
|
assert by_key["intake"]["state"] == "idle"
|
|
assert by_key["intake"]["count"] == 0
|
|
|
|
# Summary mirrors the snapshot counts.
|
|
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:
|
|
"""A not-ok snapshot still serializes to a valid, JSON-dumpable payload."""
|
|
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"
|
|
# Stages still present (all idle) so the client can render the empty map.
|
|
assert [s["key"] for s in payload["stages"]] == [s.key for s in STAGES]
|
|
assert all(s["count"] == 0 for s in payload["stages"])
|
|
# Round-trips through json.
|
|
assert json.loads(json.dumps(payload))["ok"] is False
|
|
|
|
|
|
def test_descriptions_escaped_in_html_and_json_seed() -> None:
|
|
"""Hostile descriptions must not break out of HTML *or* the inline JSON."""
|
|
snap = Snapshot(
|
|
generated_at="now",
|
|
db_path="/x.sqlite",
|
|
ok=True,
|
|
tasks=[
|
|
TaskView(
|
|
thread_id="x",
|
|
short_id="x",
|
|
task="</script><script>alert(1)</script>",
|
|
current_phase="plan",
|
|
status="active",
|
|
)
|
|
],
|
|
)
|
|
html_out = render_html(snap)
|
|
# No raw script tag survives anywhere in the document (table render escapes
|
|
# it; the inline JSON seed escapes < and > to \\uXXXX).
|
|
assert "<script>alert(1)" not in html_out
|
|
assert "</script><script>" not in html_out
|
|
assert "\\u003cscript\\u003e" in html_out # JSON seed escaped form
|
|
|
|
# The JSON sidecar itself is valid JSON carrying the raw description (the
|
|
# consumer renders it via textContent, never as markup).
|
|
payload = snapshot_to_dict(snap)
|
|
body = json.dumps(payload)
|
|
parsed = json.loads(body)
|
|
assert parsed["tasks"][0]["task"] == "</script><script>alert(1)</script>"
|
|
|
|
|
|
def test_build_snapshot_api_payload_against_seeded_ledger(tmp_path: Path) -> None:
|
|
"""End-to-end: read a seeded ledger, then assert the /api/state payload."""
|
|
db = tmp_path / "agent_team.sqlite"
|
|
_seed_db(db)
|
|
|
|
payload = snapshot_to_dict(build_snapshot(db))
|
|
|
|
assert payload["ok"] is True
|
|
by_key = {s["key"]: s for s in payload["stages"]}
|
|
# thread-waiting (clarify, open question) -> clarify node, awaiting_human.
|
|
assert by_key["clarify"]["state"] == "awaiting_human"
|
|
assert by_key["clarify"]["count"] == 1
|
|
assert by_key["clarify"]["tasks"][0]["thread_id"] == "thread-waiting"
|
|
# thread-active (plan, active) -> plan node, active.
|
|
assert by_key["plan"]["state"] == "active"
|
|
assert by_key["plan"]["count"] == 1
|
|
# Budget surfaced.
|
|
assert payload["budget"]["available"] is True
|
|
assert payload["budget"]["total_usd"] == pytest.approx(0.25)
|
|
# Whole payload is JSON-serializable.
|
|
assert json.loads(json.dumps(payload))["summary"]["total"] == 2
|