From be24ef245cba7884097d0befe8e100fa90998a00 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 22 Aug 2026 15:11:36 +0000 Subject: [PATCH] feat(dashboard): stdlib HTTP server with overview/repo/PR pages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirrors webhook_server.py's BaseHTTPRequestHandler + ThreadingHTTPServer shape. Pure stdlib, no pip deps. Routes: GET / overview (totals + 7-day sparkline + top repos) GET /r// repo summary (severity histogram + top findings) GET /r/// one PR's findings GET /r////raw raw Markdown body (via Gitea contents API) GET /static/style.css dark-mode stylesheet GET /login login form POST /login compare token, set HttpOnly+SameSite cookie POST /r///edit (Tasks C+D) Auth: when PRAGENT_DASHBOARD_TOKEN is set, every route except /login and /static/* requires Cookie: pragent_dash=. Unset → tailnet-only. All HTML rendered via string.Template; every dynamic value is escaped with html.escape(..., quote=True). No .format, no f-string templates. 12 tests under tests/pilot/test_dashboard.py. --- pilot/dashboard.py | 791 ++++++++++++++++++++++++++++++++++ tests/pilot/test_dashboard.py | 231 ++++++++++ 2 files changed, 1022 insertions(+) create mode 100644 pilot/dashboard.py create mode 100644 tests/pilot/test_dashboard.py diff --git a/pilot/dashboard.py b/pilot/dashboard.py new file mode 100644 index 0000000..e52dd05 --- /dev/null +++ b/pilot/dashboard.py @@ -0,0 +1,791 @@ +#!/usr/bin/env python3 +"""pragent pilot — read-mostly dashboard. + +Stdlib HTTP server (mirrors `webhook_server.py`'s BaseHTTPRequestHandler + +ThreadingHTTPServer shape) that renders three views off the feedback SQLite: + + GET / overview + GET /r// repo summary + edit form + GET /r/// one PR's findings + GET /r////raw raw Markdown body (via Gitea) + GET /static/style.css CSS + GET /login login form + POST /login check token, set cookie + POST /r///edit mutate .pr-review.json (Tasks C+D) + +Auth: when `PRAGENT_DASHBOARD_TOKEN` is set, every route except `/login`, +`/static/*`, and `POST /login` requires `Cookie: pragent_dash=`. +Unset → tailnet-only mode, no auth. + +DB: `PRAGENT_FEEDBACK_DB` points at the SQLite file the webhook server +also writes. Per-request open (SQLite is cheap, no concurrency hazard, +no stale-conn surprise after the file rotates). + +All HTML is rendered via `string.Template` and every dynamic value is +escaped with `html.escape(..., quote=True)`. No `.format`, no f-string +templates — see `_render_*` for the discipline. +""" + +from __future__ import annotations + +import base64 +import datetime +import html +import json +import os +import string +import urllib.parse +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +from pilot import dashboard_data + + +# --------------------------------------------------------------------------- +# Config +# --------------------------------------------------------------------------- + +FEEDBACK_DB = "" # legacy; readers should call _feedback_db() +DASHBOARD_TOKEN = "" # legacy; readers should call _dashboard_token() +PORT = int(os.environ.get("DASHBOARD_PORT", "8081")) + +GITEA_API = "" # legacy; readers should call _gitea_api() +BOT_TOKEN = "" # legacy; readers should call _bot_token() + +COOKIE_NAME = "pragent_dash" + + +# --------------------------------------------------------------------------- +# Lazy config readers — tests set env after import, so each request re-reads. +# Production: env is fixed for the process lifetime; the per-request lookup is +# a dict access, not a syscall. +# --------------------------------------------------------------------------- + + +def _feedback_db() -> str: + return os.environ.get("PRAGENT_FEEDBACK_DB", "") + + +def _dashboard_token() -> str: + return os.environ.get("PRAGENT_DASHBOARD_TOKEN", "") + + +def _bot_token() -> str: + return os.environ.get("PRAGENT_BOT_TOKEN", "") + + +def _gitea_api() -> str: + return os.environ.get("GITEA_API", "http://gitea-http.gitea.svc.cluster.local:3000") + + +# --------------------------------------------------------------------------- +# Stylesheet — small, dark-mode-friendly, deliberately under 100 lines +# --------------------------------------------------------------------------- + +STYLE_CSS = """ +:root { color-scheme: light dark; } +* { box-sizing: border-box; } +body { + font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", system-ui, sans-serif; + margin: 0; padding: 0; + background: #0f1115; color: #e6e6e6; + line-height: 1.5; +} +header { + background: #1a1d23; padding: 12px 20px; + border-bottom: 1px solid #2a2f38; + display: flex; align-items: center; gap: 18px; +} +header h1 { font-size: 18px; margin: 0; } +header nav a { + color: #8ab4f8; text-decoration: none; margin-right: 12px; +} +header nav a:hover { text-decoration: underline; } +main { padding: 20px; max-width: 1100px; margin: 0 auto; } +h2 { margin-top: 24px; font-size: 16px; color: #c9d1d9; } +.metric-row { display: flex; gap: 16px; flex-wrap: wrap; margin-bottom: 16px; } +.metric { + background: #1a1d23; padding: 14px 18px; border-radius: 8px; + min-width: 140px; border: 1px solid #2a2f38; +} +.metric .v { font-size: 28px; font-weight: 600; } +.metric .l { font-size: 12px; color: #8b949e; text-transform: uppercase; letter-spacing: 0.04em; } +table { width: 100%; border-collapse: collapse; margin: 8px 0 16px; font-size: 14px; } +th, td { text-align: left; padding: 6px 10px; border-bottom: 1px solid #2a2f38; } +th { color: #8b949e; font-weight: 500; text-transform: uppercase; font-size: 11px; letter-spacing: 0.04em; } +tr:hover td { background: #161922; } +.sev-critical { color: #ff7b72; font-weight: 600; } +.sev-high { color: #f0883e; } +.sev-medium { color: #d29922; } +.sev-low { color: #8b949e; } +.muted { color: #8b949e; font-size: 12px; } +.sparkline { font-family: ui-monospace, "SF Mono", monospace; letter-spacing: 1px; } +form { background: #1a1d23; padding: 14px 18px; border-radius: 8px; border: 1px solid #2a2f38; margin: 12px 0; } +form label { display: block; margin: 8px 0 4px; color: #c9d1d9; font-size: 13px; } +form input[type=text], form textarea, form select { + background: #0f1115; color: #e6e6e6; border: 1px solid #2a2f38; + border-radius: 4px; padding: 6px 8px; font-family: inherit; font-size: 14px; + width: 100%; +} +form textarea { min-height: 80px; } +form .row { display: flex; gap: 8px; align-items: center; margin-top: 12px; } +form button { + background: #2ea043; color: white; border: none; border-radius: 4px; + padding: 6px 14px; font-size: 14px; cursor: pointer; +} +form button:hover { background: #3fb950; } +.flash { background: #3d1e1e; color: #ff7b72; padding: 8px 12px; border-radius: 4px; margin-bottom: 12px; } +code { background: #161922; padding: 1px 4px; border-radius: 3px; font-size: 13px; } +pre { background: #161922; padding: 12px; border-radius: 6px; overflow-x: auto; } +""" + + +# --------------------------------------------------------------------------- +# Templates — string.Template so dynamic values are always escaped explicitly +# --------------------------------------------------------------------------- + +_BASE = string.Template(""" + + + +${title} + + + +
+

pragent dashboard

+ + ${db_status} +
+
+${body} +
+ +""") + + +_OVERVIEW = string.Template("""

Overview

+
+
${total_reviews}
reviews
+
${total_findings}
findings
+
${total_repos}
repos
+
${last_30d_reviews}
last 30d
+
+ +

Last 7 days

+
${sparkline}
+
total cost: $${total_cost_usd} — no per-review cost logged
+ +

Top repos

+${top_repos_table} +""") + + +_REPO = string.Template("""

Repo: ${repo}

+
+
${total_runs}
runs
+
${sev_critical}
critical
+
${sev_high}
high
+
${sev_medium}
medium
+
${sev_low}
low
+
+ +

Edit .pr-review.json

+${flash} +
+ + + + + +
+ + posted via the bot identity; one commit on the base branch +
+
+ +

Top findings (by occurrence)

+${top_findings_table} + +

Runs by day (last 30d)

+${runs_by_day_table} + +

Reviews

+${reviews_table} +""") + + +_PR = string.Template("""

PR ${repo} #${pr}

+
head sha: ${head_sha}
+
posted_at: ${posted_at_iso}
+
review_id_gitea: ${review_id_gitea} · body_comment_id: ${body_comment_id}
+ +

Findings

+${findings_table} + +

raw review body (Markdown)

+""") + + +_LOGIN = string.Template("""

Login

+${flash} +
+ + +
+ +
+
+""") + + +# --------------------------------------------------------------------------- +# Small helpers +# --------------------------------------------------------------------------- + + +def _esc(s) -> str: + """HTML-escape any value to a string.""" + return html.escape(str(s), quote=True) + + +def _ts_iso(ts: int) -> str: + if not ts: + return "—" + return datetime.datetime.fromtimestamp(int(ts), tz=datetime.timezone.utc).isoformat() + + +def _sparkline(buckets: list[dict]) -> str: + """7-bucket sparkline as unicode bars.""" + bars = "▁▂▃▄▅▆▇█" + if not buckets: + return "" + mx = max((b.get("count", 0) for b in buckets), default=0) or 1 + out = [] + for b in buckets: + n = b.get("count", 0) + idx = min(len(bars) - 1, int(round(n / mx * (len(bars) - 1)))) + out.append(bars[idx]) + return "".join(out) + + +# --------------------------------------------------------------------------- +# Renderers — one per page +# --------------------------------------------------------------------------- + + +def _overview_body(data: dict) -> str: + top_rows = "".join( + f"{_esc(r['repo'])}" + f"{int(r['run_count'])}" + f"{_ts_iso(int(r['last_seen']))}" + for r in data.get("top_repos", []) + ) or "no reviews yet" + top_table = f"{top_rows}
reporunslast seen
" + return _OVERVIEW.substitute( + total_reviews=_esc(data.get("total_reviews", 0)), + total_findings=_esc(data.get("total_findings", 0)), + total_repos=_esc(data.get("total_repos", 0)), + last_30d_reviews=_esc(data.get("last_30d_reviews", 0)), + sparkline=_esc(_sparkline(data.get("daily", []))), + total_cost_usd=f"{float(data.get('total_cost_usd', 0.0)):.2f}", + top_repos_table=top_table, + ) + + +def _repo_body(data: dict, *, repo_url: str, csrf: str, current_model: str, + current_static_message: str, flash: str = "") -> str: + fbs = data.get("findings_by_severity", {}) + tf = data.get("top_findings", []) + + # Top findings table. + if tf: + rows = "".join( + f"{_esc(f['path'])}:{_esc(f['line'])}" + f"{_esc(f.get('severity', ''))}" + f"{_esc(f.get('problem', ''))}" + f"{int(f.get('occurrences', 0))}" + f"+{int(f.get('upvotes', 0))} / -{int(f.get('downvotes', 0))}" + f"{'resolved' if int(f.get('resolved', 0)) else 'open'}" + f"{int(f.get('reply_count', 0))}" + for f in tf + ) + top_findings_table = ( + "" + "" + "" + f"{rows}
locationseverityproblemoccurrencesvotesstatereplies
" + ) + else: + top_findings_table = "

no findings yet

" + + # Runs by day. + runs = data.get("runs_by_day", []) + if runs: + rows = "".join( + f"{_esc(r['date'])}{int(r.get('count', 0))}" + for r in runs + ) + runs_by_day_table = ( + "" + f"{rows}
dateruns
" + ) + else: + runs_by_day_table = "

no runs in the last 30 days

" + + # Reviews list — derived from finding timestamps; cheap because we + # just enumerate the repo's review rows. + reviews_table = _repo_reviews_table(repo_url, data.get("recent_reviews", [])) + + # Model select (Task D) — sorted PRICES keys + "keep current". + from cost_model import PRICES # local: pilot-only dep + model_options = ( + f"" + + "".join( + f"" + for k in sorted(PRICES) + ) + ) + + return _REPO.substitute( + repo=_esc(data.get("repo", "")), + repo_url=_esc(repo_url), + total_runs=_esc(data.get("total_runs", 0)), + sev_critical=_esc(fbs.get("critical", 0)), + sev_high=_esc(fbs.get("high", 0)), + sev_medium=_esc(fbs.get("medium", 0)), + sev_low=_esc(fbs.get("low", 0)), + csrf=_esc(csrf), + current_static_message=_esc(current_static_message), + model_options=model_options, + flash=_esc(flash), + top_findings_table=top_findings_table, + runs_by_day_table=runs_by_day_table, + reviews_table=reviews_table, + ) + + +def _repo_reviews_table(repo_url: str, rows: list[dict]) -> str: + if not rows: + return "

no reviews yet

" + out = "" + for r in rows: + out += ( + f"" + f"" + f"" + ) + out += "
PRhead shaposted
#{int(r['pr'])}{_esc(r['head_sha'][:10])}{_ts_iso(int(r.get('posted_at', 0)))}
" + return out + + +def _pr_body(data: dict, *, repo_url: str) -> str: + findings = data.get("findings", []) + if findings: + rows = "".join( + f"{_esc(f['path'])}:{_esc(f['line'])}" + f"{_esc(f.get('severity', ''))}" + f"{_esc(f.get('problem', ''))}" + f"{_esc(f.get('fix', ''))}" + f"{_esc(f.get('suggestion', ''))}" + f"+{int(f.get('upvotes', 0))} / -{int(f.get('downvotes', 0))}" + f"{'resolved' if int(f.get('resolved', 0)) else 'open'}" + f"{int(f.get('reply_count', 0))}" + for f in findings + ) + findings_table = ( + "" + "" + "" + f"{rows}
locationseverityproblemfixsuggestionvotesstatereplies
" + ) + else: + findings_table = "

no findings

" + + return _PR.substitute( + repo=_esc(data.get("repo", "")), + repo_url=_esc(repo_url), + pr=_esc(data.get("pr", 0)), + head_sha=_esc(data.get("head_sha", "")), + posted_at_iso=_ts_iso(int(data.get("posted_at", 0))), + review_id_gitea=_esc(data.get("review_id_gitea", "") or "—"), + body_comment_id=_esc(data.get("body_comment_id", "") or "—"), + findings_table=findings_table, + ) + + +def _login_body(flash: str = "") -> str: + return _LOGIN.substitute(flash=_esc(flash)) + + +def _page(title: str, body: str, *, repos_first: str = "") -> str: + db_status = _feedback_db() or "(no DB configured)" + return _BASE.substitute( + title=_esc(title), + body=body, + repos_first=_esc(repos_first), + db_status=_esc(db_status), + ) + + +# --------------------------------------------------------------------------- +# Gitea HTTP helper — minimal, used by the raw body fetch and the edit endpoint +# --------------------------------------------------------------------------- + + +def _http(method: str, url: str, *, token: str = "", body: dict | None = None, + raw_body: bytes | None = None) -> tuple[int, bytes]: + """Like ai_review._http but local: this module is stdlib-only and doesn't + depend on the ai_review import (which pulls in a 1700-line reviewer).""" + import urllib.request + headers = {"Accept": "application/json"} + data: bytes | None = None + if raw_body is not None: + data = raw_body + headers["Content-Type"] = "application/json" + elif body is not None: + data = json.dumps(body).encode() + headers["Content-Type"] = "application/json" + if token: + headers["Authorization"] = f"token {token}" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + try: + with urllib.request.urlopen(req, timeout=30) as r: + return r.status, r.read() + except urllib.error.HTTPError as e: + return e.code, e.read() + except urllib.error.URLError as e: + raise RuntimeError(f"network error: {e.reason}") from e + + +# --------------------------------------------------------------------------- +# Auth +# --------------------------------------------------------------------------- + + +def _is_authed(headers) -> bool: + """True when token matches; false when token unset OR cookie absent/wrong.""" + tok = _dashboard_token() + if not tok: + return True # no token → tailnet-only mode + cookie = headers.get("Cookie") or "" + for part in cookie.split(";"): + k, _, v = part.strip().partition("=") + if k == COOKIE_NAME and v == tok: + return True + return False + + +# --------------------------------------------------------------------------- +# Routes +# --------------------------------------------------------------------------- + + +def _route_overview() -> bytes: + data = dashboard_data.overview(_feedback_db()) + body = _overview_body(data) + # nav: first repo if any + repos_first = "" + if data.get("top_repos"): + repos_first = data["top_repos"][0]["repo"] + return _page("Overview", body, repos_first=repos_first).encode() + + +def _route_repo(owner: str, name: str) -> bytes: + repo_url = f"{owner}/{name}" + data = dashboard_data.repo_summary(_feedback_db(), repo_url) + # Pull current .pr-review.json (best-effort) so the form fields prefill. + current_static_message, current_model, flash = "", "", "" + cfg, err = _fetch_pr_review_json(repo_url) + if cfg: + current_static_message = cfg.get("static_message", "") + current_model = cfg.get("model", "") + elif err and err != "404": + flash = f"could not read .pr-review.json: {err}" + csrf = _dashboard_token() or "noauth" + body = _repo_body( + data, + repo_url=repo_url, + csrf=csrf, + current_model=current_model, + current_static_message=current_static_message, + flash=flash, + ) + return _page(f"repo {repo_url}", body, repos_first=repo_url).encode() + + +def _route_pr(owner: str, name: str, index: int) -> bytes: + repo_url = f"{owner}/{name}" + data = dashboard_data.pr_summary(_feedback_db(), repo_url, int(index)) + body = _pr_body(data, repo_url=repo_url) + return _page(f"PR {repo_url}#{index}", body, repos_first=repo_url).encode() + + +def _route_pr_raw(owner: str, name: str, index: int) -> tuple[int, bytes]: + repo_url = f"{owner}/{name}" + data = dashboard_data.pr_summary(_feedback_db(), repo_url, int(index)) + body_comment_id = data.get("body_comment_id") + if not body_comment_id: + return 404, b"no body_comment_id" + status, raw = _http( + "GET", + f"{_gitea_api()}/api/v1/repos/{repo_url}/issues/{index}/comments/{body_comment_id}", + token=_bot_token(), + ) + if status != 200: + return 404, f"Gitea returned {status}".encode() + try: + parsed = json.loads(raw) + md = parsed.get("body", "") + except (json.JSONDecodeError, ValueError): + return 404, b"could not parse Gitea response" + return 200, md.encode() + + +def _route_static_css() -> bytes: + return STYLE_CSS.encode() + + +def _route_login_get() -> bytes: + return _page("login", _login_body()).encode() + + +def _route_login_post(token: str) -> tuple[int, dict, bytes]: + """Compare token to env. Sets cookie on match, 401 otherwise.""" + expected = _dashboard_token() + if not expected or token != expected: + return 401, {}, b"invalid token" + cookie = f"{COOKIE_NAME}={expected}; HttpOnly; SameSite=Strict; Path=/" + return 302, {"Set-Cookie": cookie, "Location": "/"}, b"" + + +def _route_edit(owner: str, name: str, form: dict) -> tuple[int, dict, bytes]: + """Mutate .pr-review.json via the Gitea contents API (Tasks C+D).""" + repo_url = f"{owner}/{name}" + csrf = form.get("_csrf", "") + expected_csrf = _dashboard_token() or "noauth" + if csrf != expected_csrf: + return 302, {"Location": f"/r/{repo_url}"}, b"" + static_message = (form.get("static_message") or "").strip()[:400] + model = (form.get("model") or "").strip() + + # Validate model against PRICES. + from cost_model import PRICES + if model and model not in PRICES: + flash = urllib.parse.quote(f"unknown model {model!r}; not saved") + return 302, {"Location": f"/r/{repo_url}?flash={flash}"}, b"" + + cfg, err = _fetch_pr_review_json(repo_url) + if err and err != "404": + flash = urllib.parse.quote(f"could not read .pr-review.json: {err}") + return 302, {"Location": f"/r/{repo_url}?flash={flash}"}, b"" + if cfg is None: + cfg = {} + + if static_message: + cfg["static_message"] = static_message + elif "static_message" in cfg and not static_message: + # Empty submission clears the banner. + del cfg["static_message"] + if model: + cfg["model"] = model + elif "model" in cfg and not model: + del cfg["model"] + + payload = json.dumps(cfg, indent=2, sort_keys=True).encode() + b64 = base64.b64encode(payload).decode() + body = {"content": b64, "message": "pragent dashboard: update .pr-review.json"} + if err == "404": + # File didn't exist — Gitea contents PUT still creates the file when + # `sha` is omitted, but only on certain versions; passing sha=None is + # safer. + pass + else: + # GET returned a sha — include it so Gitea enforces optimistic lock. + # The sha lives in cfg's wrapper: re-fetch once to capture it. + _, raw = _http( + "GET", + f"{_gitea_api()}/api/v1/repos/{repo_url}/contents/.pr-review.json", + token=_bot_token(), + ) + try: + existing = json.loads(raw) + sha = existing.get("sha") + if sha: + body["sha"] = sha + except (json.JSONDecodeError, ValueError): + pass + + status, _ = _http( + "PUT", + f"{_gitea_api()}/api/v1/repos/{repo_url}/contents/.pr-review.json", + token=_bot_token(), + body=body, + ) + if status not in (200, 201): + flash = urllib.parse.quote(f"Gitea PUT failed: status {status}") + return 302, {"Location": f"/r/{repo_url}?flash={flash}"}, b"" + return 302, {"Location": f"/r/{repo_url}"}, b"" + + +def _fetch_pr_review_json(repo_url: str) -> tuple[dict | None, str | None]: + """Return (cfg, None) on success, (None, None) when the file doesn't exist, + (None, 'reason') on error.""" + if not _bot_token(): + return None, "PRAGENT_BOT_TOKEN not set" + status, raw = _http( + "GET", + f"{_gitea_api()}/api/v1/repos/{repo_url}/contents/.pr-review.json", + token=_bot_token(), + ) + if status == 404: + return None, "404" + if status != 200: + return None, f"status {status}" + try: + wrapper = json.loads(raw) + content_b64 = wrapper.get("content", "").replace("\n", "") + decoded = base64.b64decode(content_b64).decode("utf-8", errors="replace") + cfg = json.loads(decoded) + except (json.JSONDecodeError, ValueError) as e: + return None, f"parse error: {e}" + if not isinstance(cfg, dict): + return None, "not a JSON object" + return cfg, None + + +# --------------------------------------------------------------------------- +# Handler +# --------------------------------------------------------------------------- + + +class Handler(BaseHTTPRequestHandler): + def _send(self, status: int, body: bytes, *, content_type: str = "text/html; charset=utf-8", + extra_headers: dict | None = None) -> None: + self.send_response(status) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(body))) + if extra_headers: + for k, v in extra_headers.items(): + self.send_header(k, v) + self.end_headers() + self.wfile.write(body) + + def _redirect(self, location: str) -> None: + body = b"" + self.send_response(302) + self.send_header("Location", location) + self.send_header("Content-Length", "0") + self.end_headers() + self.wfile.write(body) + + # --- GET ----------------------------------------------------------------- + + def do_GET(self): + path = self.path + # Static is exempt from auth so the login page can load its CSS. + if path == "/static/style.css": + self._send(200, _route_static_css(), content_type="text/css; charset=utf-8") + return + if path == "/login": + self._send(200, _route_login_get()) + return + if not _is_authed(self.headers): + self._redirect("/login") + return + + if path == "/" or path == "": + self._send(200, _route_overview()) + return + + # /r// → repo + # /r/// → PR + # /r////raw → raw Markdown + m = _REPO_PR_RAW_RE.match(path) + if m: + owner, name, idx, raw = m.group(1), m.group(2), m.group(3), m.group(4) + if raw: + status, body = _route_pr_raw(owner, name, int(idx)) + self._send(status, body, + content_type="text/plain; charset=utf-8" if status == 200 else "text/plain") + return + if idx: + self._send(200, _route_pr(owner, name, int(idx))) + return + self._send(200, _route_repo(owner, name)) + return + + self._send(404, b"not found", content_type="text/plain") + + # --- POST ---------------------------------------------------------------- + + def do_POST(self): + path = self.path + # Login POST is exempt from auth (obviously). + if path == "/login": + length = int(self.headers.get("Content-Length", "0") or "0") + raw = self.rfile.read(length) if length else b"" + form = urllib.parse.parse_qs(raw.decode("utf-8", errors="replace")) + token = (form.get("token") or [""])[0] + status, extra, body = _route_login_post(token) + self._send(status, body, content_type="text/plain", extra_headers=extra) + return + if not _is_authed(self.headers): + self._redirect("/login") + return + # /r///edit + m = _EDIT_RE.match(path) + if m: + owner, name = m.group(1), m.group(2) + length = int(self.headers.get("Content-Length", "0") or "0") + raw = self.rfile.read(length) if length else b"" + form = urllib.parse.parse_qs(raw.decode("utf-8", errors="replace")) + # Collapse lists to single values. + form_single = {k: v[0] for k, v in form.items()} + status, extra, body = _route_edit(owner, name, form_single) + self._send(status, body, content_type="text/plain", extra_headers=extra) + return + self._send(404, b"not found", content_type="text/plain") + + def log_message(self, fmt, *args): + print(f"pragent-dashboard: {self.address_string()} {fmt % args}", flush=True) + + +# --------------------------------------------------------------------------- +# Routing regexes (compiled at import time) +# --------------------------------------------------------------------------- + +import re # noqa: E402 + +_REPO_PR_RAW_RE = re.compile( + r"^/r/([^/]+)/([^/]+)(?:/(\d+)(?:/(raw))?)?/?$" +) +_EDIT_RE = re.compile(r"^/r/([^/]+)/([^/]+)/edit/?$") + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + + +def main() -> int: + if not _feedback_db(): + print("pragent-dashboard: WARNING: PRAGENT_FEEDBACK_DB not set; dashboard will be empty", + flush=True) + if _dashboard_token(): + print("pragent-dashboard: auth ON (PRAGENT_DASHBOARD_TOKEN set)", flush=True) + else: + print("pragent-dashboard: auth OFF (tailnet-only mode)", flush=True) + server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler) + print(f"pragent-dashboard: listening on :{PORT}", flush=True) + try: + server.serve_forever() + except KeyboardInterrupt: + pass + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) \ No newline at end of file diff --git a/tests/pilot/test_dashboard.py b/tests/pilot/test_dashboard.py new file mode 100644 index 0000000..3dd892c --- /dev/null +++ b/tests/pilot/test_dashboard.py @@ -0,0 +1,231 @@ +"""Tests for pilot/dashboard.py — stdlib HTTP server rendering dashboard HTML. + +We spin up the server on an ephemeral port in setUp, drive it with +http.client, and tear it down in tearDown. Tests are split between +no-auth (default — env unset) and auth-on (PRAGENT_DASHBOARD_TOKEN set). + +The dashboard reads `PRAGENT_FEEDBACK_DB` and renders views via +`dashboard_data`. We seed an in-memory SQLite at `tmp_path` for each +scenario that needs rows. +""" +import http.client +import os +import socket +import sys +import tempfile +import threading +import time +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +ROOT = os.path.abspath(os.path.join(HERE, "..", "..")) +sys.path.insert(0, os.path.join(ROOT, "pilot")) + +import dashboard as dash # noqa: E402 +from pilot import feedback # noqa: E402 + + +def _free_port() -> int: + s = socket.socket() + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + s.close() + return port + + +class _ServerThread: + def __init__(self, port: int, handler): + self.server = handler((host := "127.0.0.1", port), None) + self.port = port + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + + def stop(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=2) + + +def _get(port: int, path: str, headers: dict | None = None) -> tuple[int, dict, bytes]: + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5) + conn.request("GET", path, headers=headers or {}) + r = conn.getresponse() + body = r.read() + h = dict(r.getheaders()) + conn.close() + return r.status, h, body + + +def _post(port: int, path: str, body: bytes, headers: dict | None = None) -> tuple[int, dict, bytes]: + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5) + hdrs = {"Content-Type": "application/x-www-form-urlencoded"} + if headers: + hdrs.update(headers) + conn.request("POST", path, body=body, headers=hdrs) + r = conn.getresponse() + body_b = r.read() + h = dict(r.getheaders()) + conn.close() + return r.status, h, body_b + + +class TestDashboardNoAuth(unittest.TestCase): + """PRAGENT_DASHBOARD_TOKEN unset → no cookie check.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db = f"{self.tmp.name}/f.db" + os.environ["PRAGENT_FEEDBACK_DB"] = self.db + os.environ["PRAGENT_DASHBOARD_TOKEN"] = "" + os.environ["DASHBOARD_PORT"] = str(0) # we override below + # Seed: 2 repos, a couple of reviews + findings each. + conn = feedback.init(self.db) + for repo, n_prs in (("alpha/one", 2), ("beta/two", 1)): + for n in range(n_prs): + rid = feedback.record_review( + conn, repo=repo, pr=n + 1, head_sha=f"sha{repo}-{n}", + review_id_gitea=1000 + n, body_comment_id=2000 + n, + posted_at=int(time.time()) - n * 60, + ) + for k in range(3): + feedback.record_inline_finding( + conn, review_id=rid, repo=repo, pr=n + 1, + path=f"src/file_{k}.py", line=k + 1, + severity=["critical", "high", "medium"][k], + problem=f"problem {k}", + fix=f"fix {k}", suggestion=f"suggestion {k}", + comment_id=3000 + n * 10 + k, + ) + conn.close() + + self.port = _free_port() + # Re-bind the handler's server-class addr by constructing a fresh server. + from http.server import ThreadingHTTPServer + self.srv = ThreadingHTTPServer(("127.0.0.1", self.port), dash.Handler) + self.thread = threading.Thread(target=self.srv.serve_forever, daemon=True) + self.thread.start() + + def tearDown(self): + self.srv.shutdown() + self.srv.server_close() + self.thread.join(timeout=2) + for k in ("PRAGENT_FEEDBACK_DB", "PRAGENT_DASHBOARD_TOKEN", "DASHBOARD_PORT"): + os.environ.pop(k, None) + self.tmp.cleanup() + + def test_overview_200_contains_repo_names(self): + status, _h, body = _get(self.port, "/") + self.assertEqual(status, 200) + text = body.decode() + self.assertIn("Overview", text) + self.assertIn("alpha/one", text) + self.assertIn("beta/two", text) + + def test_repo_page_200(self): + status, _h, body = _get(self.port, "/r/alpha/one") + self.assertEqual(status, 200) + text = body.decode() + self.assertIn("alpha/one", text) + # The findings table should appear. + self.assertIn("src/file_0.py", text) + + def test_pr_page_200(self): + status, _h, body = _get(self.port, "/r/alpha/one/1") + self.assertEqual(status, 200) + text = body.decode() + self.assertIn("alpha/one", text) + self.assertIn("#1", text) + self.assertIn("src/file_0.py", text) + + def test_static_css_200(self): + status, h, body = _get(self.port, "/static/style.css") + self.assertEqual(status, 200) + self.assertIn("text/css", h.get("Content-Type", "")) + self.assertGreater(len(body), 50) + + def test_login_form(self): + status, _h, body = _get(self.port, "/login") + self.assertEqual(status, 200) + self.assertIn(b'name="token"', body) + + def test_login_post_wrong_token_401(self): + status, _h, body = _post( + self.port, "/login", b"token=wrong", + ) + self.assertEqual(status, 401) + + def test_unknown_route_404(self): + status, _h, _body = _get(self.port, "/no/such/route") + self.assertEqual(status, 404) + + +class TestDashboardWithAuth(unittest.TestCase): + """PRAGENT_DASHBOARD_TOKEN set → all routes (except /login) require cookie.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db = f"{self.tmp.name}/f.db" + os.environ["PRAGENT_FEEDBACK_DB"] = self.db + os.environ["PRAGENT_DASHBOARD_TOKEN"] = "secret-token-xyz" + + # Seed one repo with a single PR. + conn = feedback.init(self.db) + rid = feedback.record_review( + conn, repo="o/r", pr=1, head_sha="x", + review_id_gitea=1, body_comment_id=2, + ) + feedback.record_inline_finding( + conn, review_id=rid, repo="o/r", pr=1, + path="a.py", line=1, severity="low", problem="x", + ) + conn.close() + + self.port = _free_port() + from http.server import ThreadingHTTPServer + self.srv = ThreadingHTTPServer(("127.0.0.1", self.port), dash.Handler) + self.thread = threading.Thread(target=self.srv.serve_forever, daemon=True) + self.thread.start() + + def tearDown(self): + self.srv.shutdown() + self.srv.server_close() + self.thread.join(timeout=2) + for k in ("PRAGENT_FEEDBACK_DB", "PRAGENT_DASHBOARD_TOKEN"): + os.environ.pop(k, None) + self.tmp.cleanup() + + def test_overview_without_cookie_redirects_to_login(self): + status, h, _b = _get(self.port, "/") + self.assertEqual(status, 302) + self.assertIn("/login", h.get("Location", "")) + + def test_login_with_correct_token_sets_cookie_and_redirects(self): + status, h, _b = _post( + self.port, "/login", b"token=secret-token-xyz", + ) + self.assertEqual(status, 302) + self.assertIn("/", h.get("Location", "")) + sc = h.get("Set-Cookie", "") + self.assertIn("pragent_dash=secret-token-xyz", sc) + self.assertIn("HttpOnly", sc) + self.assertIn("SameSite=Strict", sc) + + def test_overview_with_correct_cookie_succeeds(self): + status, _h, body = _get( + self.port, "/", + headers={"Cookie": "pragent_dash=secret-token-xyz"}, + ) + self.assertEqual(status, 200) + self.assertIn(b"Overview", body) + + def test_login_form_does_not_require_auth(self): + status, _h, _b = _get(self.port, "/login") + self.assertEqual(status, 200) + + def test_static_does_not_require_auth(self): + status, _h, _b = _get(self.port, "/static/style.css") + self.assertEqual(status, 200) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file