test(agent-team): unit tests for status_page render + read-only reader

2026-06-23 14:15:34 -04:00 · 2026-06-23 14:15:34 -04:00 · de215dfb64
commit de215dfb64
parent 97a4befd24
1 changed files with 265 additions and 0 deletions
--- a/agent-team/tests/test_status_page.py
+++ b/agent-team/tests/test_status_page.py
@ -0,0 +1,265 @@
+"""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 sqlite3
+from contextlib import closing
+from pathlib import Path
+
+import pytest
+
+from agent_team.status_page import (
+    Snapshot,
+    TaskView,
+    build_snapshot,
+    render_html,
+)
+
+# --- 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 "&lt;script&gt;" 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