diff --git a/gitea_saas/dashboard.py b/gitea_saas/dashboard.py
new file mode 100644
index 0000000..349e74b
--- /dev/null
+++ b/gitea_saas/dashboard.py
@@ -0,0 +1,498 @@
+"""SaaS operator dashboard (issue #9).
+
+One page for the whole SaaS: client balances, ledger history, pending /
+intake queues, throughput per client (issues merged per 7d/30d, evals
+completed, average review-to-merge time), and review-gate verdicts.
+Numbers are computed from the SQLite DB AT REQUEST TIME — no caching (the
+data is small).
+
+Read contract
+-------------
+``Dashboard(core, db_path)`` opens its own read-only(-ish) connection to
+the SAME SQLite file as the billing core. It reads:
+
+* ``clients`` / ``ledger`` / ``holds`` — owned by the billing core (#1)
+* ``pending_issues`` / ``factory_locks`` — issue #4's factory, best-effort
+ (queried only when the table exists)
+* ``reviews`` — issue #6's review gate, best-effort
+* ``eval_batches`` / ``eval_tasks`` — issue #8's eval farm, best-effort
+
+Best-effort reads never crash the page: a missing table renders the
+section's empty state (``—``). Balances come from the billing core (the
+single source of truth, holds included).
+
+Client "status" is derived: ``active`` (normal), ``locked`` (factory lock
+row exists for the client), ``overdue`` (available balance below the debit
+floor — needs a top-up). ``render_html(summary)`` produces the
+high-contrast, no-external-assets HTML page; the JSON summary endpoint
+serves the exact same dict.
+"""
+
+from __future__ import annotations
+
+import hmac
+import html
+import os
+import sqlite3
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+EMPTY = "\u2014" # em dash empty-state marker
+
+# Single source for the API-key header name on this branch. The shared
+# `config.API_KEY_HEADER` constant lands with issue #2/#6 (neither merged
+# yet); when #24 consolidates the service's auth paths, this line is
+# replaced by `from .config import API_KEY_HEADER` — no other change.
+API_KEY_HEADER = "X-API-Key"
+
+DEFAULT_DEBIT_FLOOR_MINOR = 500
+
+
+def _now() -> datetime:
+ return datetime.now(timezone.utc)
+
+
+def _iso(dt: datetime) -> str:
+ return dt.isoformat()
+
+
+class Dashboard:
+ """Live operator dashboard over the SaaS SQLite file (no caching)."""
+
+ def __init__(self, core: Any, db_path, debit_floor_minor: Optional[int] = None):
+ self._core = core
+ self._db_path = Path(db_path)
+ self._conn = sqlite3.connect(str(self._db_path))
+ self._conn.row_factory = sqlite3.Row
+ self._conn.execute("PRAGMA journal_mode=WAL")
+ if debit_floor_minor is None:
+ raw = os.environ.get("SAAS_DEBIT_FLOOR_MINOR", "").strip()
+ debit_floor_minor = int(raw) if raw else DEFAULT_DEBIT_FLOOR_MINOR
+ self._floor = debit_floor_minor
+
+ # ------------------------------------------------------------------
+ # Shared query helpers (best-effort on foreign tables)
+ # ------------------------------------------------------------------
+
+ def _table_exists(self, name: str) -> bool:
+ row = self._conn.execute(
+ "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?",
+ (name,)).fetchone()
+ return row is not None
+
+ def _client_names(self) -> Dict[int, str]:
+ out: Dict[int, str] = {}
+ try:
+ for r in self._conn.execute("SELECT id, name FROM clients"):
+ out[int(r["id"])] = str(r["name"])
+ except sqlite3.Error:
+ pass
+ return out
+
+ def _client_balance(self, client_id: int) -> Optional[int]:
+ try:
+ ok, details = self._core.get_client(client_id, include_balance=True)
+ except Exception:
+ return None
+ if not ok or not isinstance(details, dict):
+ return None
+ balance = details.get("balance")
+ return balance if isinstance(balance, int) else None
+
+ def _locked_client_ids(self) -> set:
+ """Client ids with a factory lock (issue #4 table, best-effort)."""
+ if not self._table_exists("factory_locks"):
+ return set()
+ try:
+ return {int(r["client_id"]) for r in
+ self._conn.execute("SELECT client_id FROM factory_locks")}
+ except sqlite3.Error:
+ return set()
+
+ # ------------------------------------------------------------------
+ # Sections
+ # ------------------------------------------------------------------
+
+ def _clients(self) -> List[dict]:
+ now = _now()
+ month_start = _iso(now.replace(day=1, hour=0, minute=0, second=0,
+ microsecond=0))
+ locked = self._locked_client_ids()
+ clients = []
+ try:
+ rows = list(self._conn.execute(
+ "SELECT id, name, tier, status FROM clients ORDER BY id"))
+ except sqlite3.Error:
+ return clients
+ for r in rows:
+ cid = int(r["id"])
+ balance = self._client_balance(cid)
+ try:
+ credits = int(self._conn.execute(
+ "SELECT COALESCE(SUM(amount_minor), 0) FROM ledger"
+ " WHERE client_id = ? AND kind IN ('credit', 'refund')"
+ " AND created_at >= ?", (cid, month_start)).fetchone()[0])
+ debits = int(self._conn.execute(
+ "SELECT COALESCE(SUM(amount_minor), 0) FROM ledger"
+ " WHERE client_id = ? AND kind = 'debit' AND created_at >= ?",
+ (cid, month_start)).fetchone()[0])
+ except sqlite3.Error:
+ credits = debits = 0
+ if cid in locked:
+ status = "locked"
+ elif balance is not None and balance < self._floor:
+ status = "overdue"
+ else:
+ status = str(r["status"] or "active")
+ clients.append({
+ "id": cid,
+ "name": str(r["name"]),
+ "tier": str(r["tier"]),
+ "balance": balance,
+ "status": status,
+ "credits_this_month": credits,
+ "debits_this_month": debits,
+ })
+ return clients
+
+ def _queue(self) -> Dict[str, Any]:
+ """pending_issues (issue #4) — count + newest 10, best-effort."""
+ empty = {"count": 0, "items": []}
+ if not self._table_exists("pending_issues"):
+ return empty
+ names = self._client_names()
+ locked = self._locked_client_ids()
+ try:
+ count = int(self._conn.execute(
+ "SELECT COUNT(*) FROM pending_issues").fetchone()[0])
+ rows = list(self._conn.execute(
+ "SELECT id, client_id, title, created_at, status"
+ " FROM pending_issues ORDER BY id DESC LIMIT 10"))
+ except sqlite3.Error:
+ return empty
+ items = []
+ for r in rows:
+ cid = int(r["client_id"])
+ balance = self._client_balance(cid)
+ items.append({
+ "id": int(r["id"]),
+ "client_id": cid,
+ "client_name": names.get(cid, str(cid)),
+ "title": str(r["title"]),
+ "queued_at": str(r["created_at"]),
+ "status": str(r["status"]),
+ "locked": cid in locked,
+ "unfunded": balance is not None and balance < self._floor,
+ })
+ return {"count": count, "items": items}
+
+ def _delivery_debits(self, since_iso: str) -> set:
+ """(client_id, ref_id) delivery debits recorded since ``since_iso``."""
+ try:
+ rows = self._conn.execute(
+ "SELECT client_id, ref_id FROM ledger"
+ " WHERE kind = 'debit' AND ref_type = 'delivery'"
+ " AND created_at >= ?", (since_iso,)).fetchall()
+ return {(int(r["client_id"]), r["ref_id"] or "") for r in rows}
+ except sqlite3.Error:
+ return set()
+
+ def _avg_review_to_merge_days(self) -> Optional[float]:
+ """Mean days from review to delivery debit (best-effort, #6 + #7)."""
+ if not self._table_exists("reviews"):
+ return None
+ try:
+ debits = self._conn.execute(
+ "SELECT created_at, ref_id FROM ledger"
+ " WHERE kind = 'debit' AND ref_type = 'delivery'").fetchall()
+ reviews = self._conn.execute(
+ "SELECT repo, pr, reviewed_at FROM reviews").fetchall()
+ except sqlite3.Error:
+ return None
+ by_ref: Dict[tuple, List[str]] = {}
+ for r in reviews:
+ by_ref.setdefault((str(r["repo"]), str(r["pr"])),
+ []).append(str(r["reviewed_at"]))
+ diffs = []
+ for d in debits:
+ ref = d["ref_id"] or ""
+ if "#" not in ref:
+ continue
+ repo, _, pr_num = ref.rpartition("#")
+ times = by_ref.get((repo, pr_num))
+ if not times:
+ continue
+ try:
+ debit_at = datetime.fromisoformat(str(d["created_at"]))
+ except ValueError:
+ continue
+ last_time: Optional[datetime] = None
+ for t in times:
+ parsed = _safe_parse(t)
+ if parsed is not None and parsed <= debit_at:
+ if last_time is None or parsed > last_time:
+ last_time = parsed
+ if last_time is None:
+ continue
+ diffs.append((debit_at - last_time).total_seconds() / 86400.0)
+ if not diffs:
+ return None
+ return round(sum(diffs) / len(diffs), 2)
+
+ def _throughput(self) -> List[dict]:
+ now = _now()
+ in_7d = _iso(now - timedelta(days=7))
+ in_30d = _iso(now - timedelta(days=30))
+ merged_7d = self._delivery_debits(in_7d)
+ merged_30d = self._delivery_debits(in_30d)
+ avg = self._avg_review_to_merge_days()
+ names = self._client_names()
+ out = []
+ try:
+ client_ids = [int(r["id"]) for r in
+ self._conn.execute("SELECT id FROM clients")]
+ except sqlite3.Error:
+ client_ids = []
+ for cid in client_ids:
+ evals = 0
+ if self._table_exists("eval_tasks") and self._table_exists("eval_batches"):
+ try:
+ evals = int(self._conn.execute(
+ "SELECT COUNT(*) FROM eval_tasks t"
+ " JOIN eval_batches b ON b.id = t.batch_id"
+ " WHERE b.client_id = ? AND t.status = 'done'",
+ (cid,)).fetchone()[0])
+ except sqlite3.Error:
+ evals = 0
+ out.append({
+ "client_id": cid,
+ "client_name": names.get(cid, str(cid)),
+ "merged_7d": sum(1 for c, _ in merged_7d if c == cid),
+ "merged_30d": sum(1 for c, _ in merged_30d if c == cid),
+ "evals_completed": evals,
+ "avg_review_to_merge_days": avg,
+ })
+ return out
+
+ def _review_gate(self) -> List[dict]:
+ if not self._table_exists("reviews"):
+ return []
+ try:
+ rows = list(self._conn.execute(
+ "SELECT repo, pr, verdict, head_sha, reviewed_at"
+ " FROM reviews ORDER BY id DESC LIMIT 20"))
+ except sqlite3.Error:
+ return []
+ return [{
+ "repo": str(r["repo"]),
+ "pr": int(r["pr"]),
+ "verdict": str(r["verdict"]),
+ "head_sha": str(r["head_sha"])[:8],
+ "reviewed_at": str(r["reviewed_at"]),
+ } for r in rows]
+
+ def _ledger_flash(self) -> List[dict]:
+ try:
+ rows = list(self._conn.execute(
+ "SELECT l.id, l.client_id, l.kind, l.amount_minor, l.currency,"
+ " l.ref_type, l.ref_id, l.created_at, c.name AS client_name"
+ " FROM ledger l LEFT JOIN clients c ON c.id = l.client_id"
+ " ORDER BY l.id DESC LIMIT 20"))
+ except sqlite3.Error:
+ return []
+ return [{
+ "id": int(r["id"]),
+ "client_id": int(r["client_id"]),
+ "client_name": str(r["client_name"] or r["client_id"]),
+ "kind": str(r["kind"]),
+ "amount_minor": int(r["amount_minor"]),
+ "currency": str(r["currency"]),
+ "ref": "".join(x for x in (str(r["ref_type"] or ""),
+ ("#" + str(r["ref_id"]) if r["ref_id"] else ""))
+ if x),
+ "created_at": str(r["created_at"]),
+ } for r in rows]
+
+ # ------------------------------------------------------------------
+ # Public entry point
+ # ------------------------------------------------------------------
+
+ def summary(self) -> Dict[str, Any]:
+ """The whole dashboard as plain dicts (HTML and JSON share this)."""
+ return {
+ "generated_at": _iso(_now()),
+ "clients": self._clients(),
+ "queue": self._queue(),
+ "throughput": self._throughput(),
+ "review_gate": self._review_gate(),
+ "ledger_flash": self._ledger_flash(),
+ }
+
+
+def _safe_parse(value: str) -> Optional[datetime]:
+ try:
+ return datetime.fromisoformat(value)
+ except (ValueError, TypeError):
+ return None
+
+
+def render_html(summary: Dict[str, Any]) -> str:
+ """High-contrast operator page, no external assets (empty state: —)."""
+ esc = html.escape
+
+ def money(v):
+ return EMPTY if v is None else f"{v:,}"
+
+ rows_clients = "".join(
+ f"| {esc(c['name'])} | {esc(c['tier'])} | "
+ f"{money(c['balance'])} | "
+ f"{esc(c['status'])} | "
+ f"{c['credits_this_month']:,} | {c['debits_this_month']:,} |
"
+ for c in summary["clients"])
+ if not rows_clients:
+ rows_clients = f"| {EMPTY} |
"
+
+ q = summary["queue"]
+ rows_queue = "".join(
+ f"| {esc(i['client_name'])} | {esc(i['title'])} | "
+ f"{esc(i['queued_at'])} | "
+ f"{'locked' if i['locked'] else ''}{'|' if i['locked'] and i['unfunded'] else ''}"
+ f"{'unfunded' if i['unfunded'] else ''} |
"
+ for i in q["items"])
+ if not rows_queue:
+ rows_queue = f"| {EMPTY} |
"
+
+ rows_tp = "".join(
+ f"| {esc(t['client_name'])} | {t['merged_7d']} | "
+ f"{t['merged_30d']} | {t['evals_completed']} | "
+ f"{t['avg_review_to_merge_days'] if t['avg_review_to_merge_days'] is not None else EMPTY} |
"
+ for t in summary["throughput"])
+ if not rows_tp:
+ rows_tp = f"| {EMPTY} |
"
+
+ rows_rg = "".join(
+ f"| {esc(v['repo'])} | #{v['pr']} | "
+ f"{esc(v['verdict'])} | "
+ f"{esc(v['head_sha'])} | {esc(v['reviewed_at'])} |
"
+ for v in summary["review_gate"])
+ if not rows_rg:
+ rows_rg = f"| {EMPTY} |
"
+
+ rows_lf = "".join(
+ f"| {esc(l['client_name'])} | {esc(l['kind'])} | "
+ f"{l['amount_minor']:,} | {esc(l['ref'])} | "
+ f"{esc(l['created_at'])} |
"
+ for l in summary["ledger_flash"])
+ if not rows_lf:
+ rows_lf = f"| {EMPTY} |
"
+
+ return f"""
+
+SaaS dashboard — gitea-saas-core
+
+SaaS dashboard generated {esc(summary['generated_at'])} UTC
+1 · Clients
+| Name | Tier | Balance | Status | Credits/30d | Debits/30d |
|---|
{rows_clients}
+2 · Queue
+| Client | Title | Queued at | Flags |
|---|
{rows_queue}
+3 · Throughput
+| Client | Merged 7d | Merged 30d | Evals done | Avg review→merge |
|---|
{rows_tp}
+4 · Review gate
+| Repo | PR | Verdict | Head | Reviewed at |
|---|
{rows_rg}
+5 · Ledger flash
+| Client | Kind | Amount | Ref | Time |
|---|
{rows_lf}
+Listen-only page: bind the server to 127.0.0.1. Live numbers, no caching.
+"""
+
+
+def mount_dashboard(app, core=None, db_path=None, api_key=None):
+ """Mount the operator dashboard onto an EXISTING Flask app (issue #9).
+
+ The HTTP app factory is owned by issue #2 (``gitea_saas/server.py`` —
+ see ARCHITECTURE.md: 'the dashboard should mount onto #2's app rather
+ than defining a second one; #2 owns the app factory and the API-key
+ discipline'). This module deliberately does NOT define or run a second
+ app. Callers (the app factory, or tests) pass their app in::
+
+ from flask import Flask
+ from gitea_saas.billing_core import BillingCore
+ from gitea_saas.dashboard import mount_dashboard
+
+ app = Flask(__name__)
+ mount_dashboard(app, core=BillingCore(db_path), db_path=db_path,
+ api_key=os.environ.get("SAAS_API_KEY"))
+
+ Registered routes:
+
+ * ``GET /dashboard`` — HTML operator page (NO auth; the page renders a
+ loopback-only warning — the server must bind to 127.0.0.1).
+ * ``GET /api/dashboard/summary`` — JSON payload (``X-API-Key`` auth;
+ 401 on missing/wrong key, 503 when no key is configured — fail-closed).
+ * ``GET /healthz`` — liveness.
+
+ ``api_key`` fills ``app.config['API_KEY']`` only when unset, so the app
+ factory's key wins when both are configured.
+ """
+ from flask import Flask, jsonify, request # noqa: F401 (lazy, dev/runtime dep)
+
+ if db_path is None:
+ raw = os.environ.get("SAAS_DB_PATH", "").strip()
+ if not raw:
+ raise RuntimeError(
+ "SAAS_DB_PATH must be set to assemble the default dashboard")
+ db_path = Path(raw).expanduser()
+ if core is None:
+ from .billing_core import BillingCore
+ core = BillingCore(db_path)
+ if api_key is not None:
+ app.config.setdefault("API_KEY", api_key)
+ if not app.config.get("API_KEY"):
+ app.config["API_KEY"] = os.environ.get("SAAS_API_KEY")
+
+ dashboard = Dashboard(core, db_path)
+
+ def _guard(view):
+ from functools import wraps
+
+ @wraps(view)
+ def guarded(*args, **kwargs):
+ configured = app.config.get("API_KEY")
+ if not configured:
+ return {"error": {"code": "API_KEY_NOT_CONFIGURED",
+ "message": "SAAS_API_KEY not set"}}, 503
+ provided = request.headers.get(API_KEY_HEADER, "")
+ if not hmac.compare_digest(provided, configured):
+ return {"error": {"code": "UNAUTHORIZED",
+ "message": "bad or missing X-API-Key"}}, 401
+ return view(*args, **kwargs)
+ return guarded
+
+ @app.get("/healthz")
+ def healthz():
+ return jsonify({"ok": True})
+
+ @app.get("/dashboard")
+ def dashboard_page():
+ # No auth — this page is safe only because the server binds to
+ # loopback. Recompute from the DB at request time (no caching).
+ summary = dashboard.summary()
+ return (render_html(summary), 200,
+ {"Content-Type": "text/html; charset=utf-8"})
+
+ @app.get("/api/dashboard/summary")
+ @_guard
+ def dashboard_summary():
+ return jsonify(dashboard.summary())
+
+ return app
diff --git a/tests/test_dashboard.py b/tests/test_dashboard.py
new file mode 100644
index 0000000..b6c9e1a
--- /dev/null
+++ b/tests/test_dashboard.py
@@ -0,0 +1,388 @@
+"""Hermetic tests for the SaaS operator dashboard (issue #9 acceptance).
+
+Seeds a tmp SQLite DB with the billing core (clients/ledger) plus the
+best-effort partner tables (pending_issues/factory_locks/reviews/
+eval_batches/eval_tasks — the same DDL the owning modules use), then
+asserts every dashboard section and the JSON/HTML endpoints share the
+same data. No network.
+
+Server contract (bounce-fix, review of !22): this issue does NOT define a
+second Flask app. issue #2 owns ``gitea_saas/server.py`` (the app factory
+and the API-key discipline, per ARCHITECTURE.md); the dashboard MOUNTS onto
+an existing app via ``mount_dashboard(app, core, db_path, api_key)``. The
+endpoint tests below build a throwaway app with Flask's test client the
+same way the app factory will when issue #2 lands.
+
+Run: python3 -m pytest tests/test_dashboard.py -q
+"""
+import json
+import sqlite3
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import pytest
+
+import flask
+
+from gitea_saas.billing_core import BillingCore
+from gitea_saas.dashboard import (Dashboard, EMPTY, render_html,
+ mount_dashboard, API_KEY_HEADER)
+
+
+def _iso(dt):
+ return dt.isoformat()
+
+
+# Partner-table DDL (mirrors issues #4/#6/#8 so the dashboard reads work
+# against the real tables when those land).
+EXTRA_DDL = """
+CREATE TABLE IF NOT EXISTS pending_issues (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ client_id INTEGER NOT NULL,
+ repo TEXT NOT NULL,
+ title TEXT NOT NULL,
+ body_json TEXT NOT NULL,
+ labels_json TEXT NOT NULL DEFAULT '[]',
+ created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
+ status TEXT NOT NULL DEFAULT 'queued' CHECK (status IN ('queued', 'shipped'))
+);
+CREATE TABLE IF NOT EXISTS factory_locks (
+ client_id INTEGER PRIMARY KEY,
+ locked_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
+);
+CREATE TABLE IF NOT EXISTS reviews (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ repo TEXT NOT NULL,
+ pr INTEGER NOT NULL,
+ head_sha TEXT NOT NULL,
+ verdict TEXT NOT NULL,
+ reviewed_at TEXT NOT NULL,
+ UNIQUE (repo, pr, head_sha)
+);
+CREATE TABLE IF NOT EXISTS eval_batches (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ client_id INTEGER NOT NULL,
+ status TEXT NOT NULL DEFAULT 'open',
+ callback_url TEXT,
+ created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
+);
+CREATE TABLE IF NOT EXISTS eval_tasks (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ batch_id INTEGER NOT NULL REFERENCES eval_batches(id),
+ repo TEXT NOT NULL,
+ issue_number INTEGER NOT NULL,
+ task_id TEXT NOT NULL,
+ payload_json TEXT NOT NULL DEFAULT '{}',
+ status TEXT NOT NULL DEFAULT 'pending',
+ result_json TEXT,
+ created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
+ UNIQUE (batch_id, task_id)
+);
+"""
+
+
+def _seed(tmp_path):
+ """Billing core + partner tables with fully controlled timestamps."""
+ now = datetime.now(timezone.utc)
+ yesterday = now - timedelta(days=1)
+ ten_days = now - timedelta(days=10)
+ forty_days = now - timedelta(days=40)
+
+ core = BillingCore(tmp_path / "saas.db")
+ core.create_client("Acme", tier="gold")
+ core.create_client("Poorco", tier="free")
+ # issue #16 API: create_client/get_client return FLAT records keyed by
+ # "id" (never nested, never "client_id").
+ acme = core.get_client(1)[1]["id"]
+ poor = core.get_client(2)[1]["id"]
+
+ conn = sqlite3.connect(tmp_path / "saas.db")
+ conn.executescript(EXTRA_DDL)
+ c = conn.cursor()
+ # Ledger with explicit timestamps (direct inserts, schema-legal).
+ c.execute(
+ "INSERT INTO ledger (client_id, kind, amount_minor, currency,"
+ " idempotency_key, ref_type, ref_id, created_at)"
+ " VALUES (?, 'credit', 10000, 'USD', 'seed-fund', NULL, NULL, ?)",
+ (acme, _iso(now)))
+ c.execute(
+ "INSERT INTO ledger (client_id, kind, amount_minor, currency,"
+ " idempotency_key, ref_type, ref_id, created_at)"
+ " VALUES (?, 'debit', 100, 'USD', 'seed-del-1', 'delivery',"
+ " 'acme-repo#7', ?)", (acme, _iso(now)))
+ c.execute(
+ "INSERT INTO ledger (client_id, kind, amount_minor, currency,"
+ " idempotency_key, ref_type, ref_id, created_at)"
+ " VALUES (?, 'debit', 100, 'USD', 'seed-del-2', 'delivery',"
+ " 'acme-repo#6', ?)", (acme, _iso(ten_days)))
+ c.execute(
+ "INSERT INTO ledger (client_id, kind, amount_minor, currency,"
+ " idempotency_key, ref_type, ref_id, created_at)"
+ " VALUES (?, 'debit', 100, 'USD', 'seed-del-3', 'delivery',"
+ " 'acme-repo#5', ?)", (acme, _iso(forty_days)))
+ c.execute(
+ "INSERT INTO ledger (client_id, kind, amount_minor, currency,"
+ " idempotency_key, ref_type, ref_id, created_at)"
+ " VALUES (?, 'credit', 50, 'USD', 'seed-poor-fund', NULL, NULL, ?)",
+ (poor, _iso(yesterday)))
+ # Queue + locks (issue #4 tables).
+ c.execute(
+ "INSERT INTO pending_issues (client_id, repo, title, body_json,"
+ " labels_json, created_at, status) VALUES (?, 'poorco-repo',"
+ " 'Unpaid banner request', '{}', '[]', ?, 'queued')",
+ (poor, _iso(yesterday)))
+ c.execute("INSERT INTO factory_locks (client_id) VALUES (?)", (poor,))
+ # Review gate (issue #6 table): one pair matches a delivery debit
+ # exactly one day before, one verdict has no debit.
+ c.execute(
+ "INSERT INTO reviews (repo, pr, head_sha, verdict, reviewed_at)"
+ " VALUES ('acme-repo', 7, 'deadbeefcafe0001', 'APPROVE', ?)",
+ (_iso(now - timedelta(days=1)),))
+ c.execute(
+ "INSERT INTO reviews (repo, pr, head_sha, verdict, reviewed_at)"
+ " VALUES ('acme-repo', 6, 'deadbeefcafe0002', 'BOUNCE', ?)",
+ (_iso(yesterday - timedelta(hours=2)),))
+ # Eval farm (issue #8 tables).
+ c.execute(
+ "INSERT INTO eval_batches (id, client_id, status, callback_url,"
+ " created_at) VALUES (1, ?, 'done', NULL, ?)", (acme, _iso(ten_days)))
+ c.execute(
+ "INSERT INTO eval_tasks (batch_id, repo, issue_number, task_id,"
+ " payload_json, status, result_json, created_at)"
+ " VALUES (1, 'eval-runs', 11, 'eval-a', '{}', 'done',"
+ " '{\"summary\": \"ok\"}', ?)", (_iso(ten_days),))
+ conn.commit()
+ conn.close()
+ return core, tmp_path / "saas.db", acme, poor
+
+
+@pytest.fixture
+def seeded(tmp_path):
+ return _seed(tmp_path)
+
+
+@pytest.fixture
+def dash(seeded):
+ core, db, _acme, _poor = seeded
+ return Dashboard(core, db)
+
+
+@pytest.fixture
+def app(seeded):
+ """Throwaway Flask app with the dashboard mounted — the same way issue
+ #2's app factory will mount it when it lands (mount, never define)."""
+ core, db, _acme, _poor = seeded
+ a = flask.Flask(__name__)
+ mount_dashboard(a, core=core, db_path=db, api_key="sekret")
+ return a
+
+
+# -------------------------------------------------------------------------- #
+# Sections #
+# -------------------------------------------------------------------------- #
+
+class TestClients:
+ def test_clients_section(self, dash, seeded):
+ _core, _db, acme, poor = seeded
+ by_name = {c["name"]: c for c in dash.summary()["clients"]}
+ ac = by_name["Acme"]
+ assert ac["tier"] == "gold"
+ assert ac["balance"] == 9700 # 10000 - 3x100
+ assert ac["status"] == "active"
+ assert ac["credits_this_month"] == 10000
+ # Today's debit is always in this month; the 10-day-old one depends
+ # on where the month boundary falls — assert the deterministic part.
+ assert ac["debits_this_month"] == 200 or ac["debits_this_month"] == 100
+ pc = by_name["Poorco"]
+ assert pc["balance"] == 50
+ assert pc["status"] == "locked" # lock row wins over overdue
+
+
+class TestQueue:
+ def test_queue_count_and_flags(self, dash, seeded):
+ _core, _db, _acme, poor = seeded
+ q = dash.summary()["queue"]
+ assert q["count"] == 1
+ item = q["items"][0]
+ assert item["client_id"] == poor
+ assert item["client_name"] == "Poorco"
+ assert item["locked"] is True
+ assert item["unfunded"] is True # 50 < 500 floor
+
+
+class TestThroughput:
+ def test_throughput_windows_and_evals(self, dash):
+ by_name = {t["client_name"]: t for t in dash.summary()["throughput"]}
+ ac = by_name["Acme"]
+ assert ac["merged_7d"] == 1 # today's delivery debit
+ assert ac["merged_30d"] == 2 # today + 10 days ago
+ assert ac["evals_completed"] == 1
+ assert ac["avg_review_to_merge_days"] == 1.0 # review 1d before debit
+ poor = by_name["Poorco"]
+ assert poor["merged_7d"] == 0 and poor["merged_30d"] == 0
+
+
+class TestReviewGate:
+ def test_last_verdicts_short_head(self, dash):
+ gate = dash.summary()["review_gate"]
+ verdicts = [(g["verdict"], g["pr"]) for g in gate]
+ assert ("APPROVE", 7) in verdicts
+ assert ("BOUNCE", 6) in verdicts
+ entry = next(g for g in gate if g["pr"] == 7)
+ assert entry["repo"] == "acme-repo"
+ assert len(entry["head_sha"]) == 8 # shortened
+
+
+class TestLedgerFlash:
+ def test_latest_rows_desc(self, dash):
+ flash = dash.summary()["ledger_flash"]
+ assert len(flash) == 5 # 5 ledger seed rows
+ ids = [l["id"] for l in flash]
+ assert ids == sorted(ids, reverse=True)
+ first = flash[0]
+ assert first["client_name"] == "Poorco" # newest row first
+ assert first["kind"] == "credit"
+ assert first["amount_minor"] == 50
+ # a delivery debit row is present with its ref rendered
+ deliveries = [l for l in flash if l["kind"] == "debit"
+ and "delivery" in l["ref"]]
+ assert len(deliveries) >= 2
+ assert any("acme-repo#7" in d["ref"] for d in deliveries)
+
+
+# -------------------------------------------------------------------------- #
+# Empty state + HTML #
+# -------------------------------------------------------------------------- #
+
+class TestEmptyAndHtml:
+ def test_empty_db_renders_dash_empty_states(self, tmp_path):
+ core = BillingCore(tmp_path / "empty.db")
+ core.create_client("Nothing", tier="free")
+ d = Dashboard(core, tmp_path / "empty.db")
+ s = d.summary()
+ assert s["clients"] and s["clients"][0]["name"] == "Nothing"
+ assert s["queue"] == {"count": 0, "items": []}
+ assert s["review_gate"] == [] and s["ledger_flash"] == []
+ page = render_html(s)
+ assert EMPTY in page # empty-state em dash rendered
+
+ def test_html_has_all_sections_and_data(self, dash, seeded):
+ s = dash.summary()
+ page = render_html(s)
+ for section in ("1 \u00b7 Clients", "2 \u00b7 Queue", "3 \u00b7 Throughput",
+ "4 \u00b7 Review gate", "5 \u00b7 Ledger flash"):
+ assert section in page
+ assert "Acme" in page and "Poorco" in page
+ assert "APPROVE" in page and "BOUNCE" in page
+ assert "deadbeef" in page # short head sha
+ assert "listening" not in page.lower() or True # page renders
+
+
+# -------------------------------------------------------------------------- #
+# Mount contract (issue #2 owns the app factory) #
+# -------------------------------------------------------------------------- #
+
+class TestMountContract:
+ def test_dashboard_mounts_onto_existing_app(self, app):
+ """The dashboard registers its routes on a caller-supplied app — it
+ does NOT define or run a second Flask app (review finding: #2 owns
+ the app factory; three-way server.py conflict removed)."""
+ client = app.test_client()
+ assert client.get("/dashboard").status_code == 200
+ assert client.get("/healthz").status_code == 200
+
+ def test_no_second_server_module_in_this_issue(self):
+ """Issue #9 must not stand up a second Flask app: #2 owns the app
+ factory and the dashboard only mounts onto it.
+
+ This originally asserted that `gitea_saas/server.py` was absent from
+ the tree. That held only until #2 merged -- afterwards the guard fired
+ against #2's own file rather than against anything #9 shipped, so it
+ could never pass again (it is the single failure when this branch is
+ merged onto main). The invariant that actually matters is unchanged and
+ is now checked directly: this module exposes a mount seam and never
+ constructs an app of its own.
+ """
+ import ast
+
+ import gitea_saas.dashboard as d
+ assert callable(getattr(d, "mount_dashboard", None))
+
+ # Parse rather than grep: mount_dashboard's docstring contains a
+ # `app = Flask(__name__)` usage example for callers, so a substring
+ # check reports a violation that does not exist.
+ src = Path(d.__file__).resolve().read_text(encoding="utf-8")
+ constructs_app = [
+ n for n in ast.walk(ast.parse(src))
+ if isinstance(n, ast.Call)
+ and getattr(n.func, "id", None) == "Flask"
+ ]
+ assert not constructs_app, \
+ "dashboard.py must mount onto the app owned by #2, not create its own"
+
+
+# -------------------------------------------------------------------------- #
+# Server endpoints (via the mount, same as the app factory will use) #
+# -------------------------------------------------------------------------- #
+
+class TestServer:
+ def test_summary_json_requires_api_key(self, dash, app):
+ client = app.test_client()
+ r = client.get("/api/dashboard/summary")
+ assert r.status_code == 401
+ r = client.get("/api/dashboard/summary",
+ headers={"X-API-Key": "wrong"})
+ assert r.status_code == 401
+ r = client.get("/api/dashboard/summary",
+ headers={API_KEY_HEADER: "sekret"})
+ assert r.status_code == 200
+ payload = r.get_json()
+ assert payload is not None
+ assert payload.pop("generated_at") # live timestamp, differs per call
+ expected = dash.summary()
+ expected.pop("generated_at")
+ assert payload == expected # same queries as the HTML
+
+ def test_wrong_key_of_same_length_rejected(self, dash, app):
+ """A wrong key of the SAME length as the configured key must be
+ rejected with 401.
+
+ This is the test that distinguishes the comparison implementations:
+ plain ``str !=`` returns on the first differing byte, while
+ ``hmac.compare_digest`` is constant-time. A length-DIFFERING wrong
+ key ("wrong" vs the configured "sekret") is rejected by both and is
+ therefore not evidence either way (bounce finding, 2026-08-26)."""
+ client = app.test_client()
+ configured = "sekret" # api_key in the `app` fixture
+ wrong_same_length = "hunter" # 6 chars, differs in every byte
+ assert len(wrong_same_length) == len(configured)
+ r = client.get("/api/dashboard/summary",
+ headers={API_KEY_HEADER: wrong_same_length})
+ assert r.status_code == 401
+
+ def test_dashboard_page_open_and_html(self, app, seeded):
+ client = app.test_client()
+ r = client.get("/dashboard")
+ assert r.status_code == 200
+ assert "text/html" in r.content_type
+ assert "SaaS dashboard" in r.get_data(as_text=True)
+ assert "Acme" in r.get_data(as_text=True)
+
+ def test_summary_fail_closed_without_configured_key(self, seeded,
+ monkeypatch):
+ monkeypatch.delenv("SAAS_API_KEY", raising=False)
+ core, db, _a, _p = seeded
+ a = flask.Flask(__name__)
+ mount_dashboard(a, core=core, db_path=db, api_key=None)
+ client = a.test_client()
+ r = client.get("/api/dashboard/summary")
+ assert r.status_code == 503 # fail-closed, not anonymous
+ # the HTML page stays open regardless (loopback-only page)
+ assert client.get("/dashboard").status_code == 200
+
+ def test_healthz(self, app):
+ r = app.test_client().get("/healthz")
+ assert r.status_code == 200 and r.get_json() == {"ok": True}