Merge dashboard agent: data + HTTP server + mutation endpoints
This commit is contained in:
Claude
@@ -0,0 +1,792 @@
|
||||
#!/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/<owner>/<name> repo summary + edit form
|
||||
GET /r/<owner>/<name>/<index> one PR's findings
|
||||
GET /r/<owner>/<name>/<index>/raw raw Markdown body (via Gitea)
|
||||
GET /static/style.css CSS
|
||||
GET /login login form
|
||||
POST /login check token, set cookie
|
||||
POST /r/<owner>/<name>/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=<token>`.
|
||||
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.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
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("""<!doctype html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<title>${title}</title>
|
||||
<link rel="stylesheet" href="/static/style.css">
|
||||
</head>
|
||||
<body>
|
||||
<header>
|
||||
<h1>pragent dashboard</h1>
|
||||
<nav>
|
||||
<a href="/">Home</a>
|
||||
<a href="/r/${repos_first}">repos</a>
|
||||
</nav>
|
||||
<span class="muted" style="margin-left:auto">${db_status}</span>
|
||||
</header>
|
||||
<main>
|
||||
${body}
|
||||
</main>
|
||||
</body>
|
||||
</html>""")
|
||||
|
||||
|
||||
_OVERVIEW = string.Template("""<h2>Overview</h2>
|
||||
<div class="metric-row">
|
||||
<div class="metric"><div class="v">${total_reviews}</div><div class="l">reviews</div></div>
|
||||
<div class="metric"><div class="v">${total_findings}</div><div class="l">findings</div></div>
|
||||
<div class="metric"><div class="v">${total_repos}</div><div class="l">repos</div></div>
|
||||
<div class="metric"><div class="v">${last_30d_reviews}</div><div class="l">last 30d</div></div>
|
||||
</div>
|
||||
|
||||
<h2>Last 7 days</h2>
|
||||
<div class="sparkline">${sparkline}</div>
|
||||
<div class="muted">total cost: $${total_cost_usd} — no per-review cost logged</div>
|
||||
|
||||
<h2>Top repos</h2>
|
||||
${top_repos_table}
|
||||
""")
|
||||
|
||||
|
||||
_REPO = string.Template("""<h2>Repo: <code>${repo}</code></h2>
|
||||
<div class="metric-row">
|
||||
<div class="metric"><div class="v">${total_runs}</div><div class="l">runs</div></div>
|
||||
<div class="metric"><div class="v">${sev_critical}</div><div class="l sev-critical">critical</div></div>
|
||||
<div class="metric"><div class="v">${sev_high}</div><div class="l sev-high">high</div></div>
|
||||
<div class="metric"><div class="v">${sev_medium}</div><div class="l sev-medium">medium</div></div>
|
||||
<div class="metric"><div class="v">${sev_low}</div><div class="l sev-low">low</div></div>
|
||||
</div>
|
||||
|
||||
<h2>Edit .pr-review.json</h2>
|
||||
${flash}
|
||||
<form method="post" action="/r/${repo_url}/edit">
|
||||
<input type="hidden" name="_csrf" value="${csrf}">
|
||||
<label for="static_message">Static banner message (max 400 chars)</label>
|
||||
<textarea id="static_message" name="static_message" maxlength="400">${current_static_message}</textarea>
|
||||
<label for="model">Model (PRICES keys)</label>
|
||||
<select id="model" name="model">${model_options}</select>
|
||||
<div class="row">
|
||||
<button type="submit">Save</button>
|
||||
<span class="muted">posted via the bot identity; one commit on the base branch</span>
|
||||
</div>
|
||||
</form>
|
||||
|
||||
<h2>Top findings (by occurrence)</h2>
|
||||
${top_findings_table}
|
||||
|
||||
<h2>Runs by day (last 30d)</h2>
|
||||
${runs_by_day_table}
|
||||
|
||||
<h2>Reviews</h2>
|
||||
${reviews_table}
|
||||
""")
|
||||
|
||||
|
||||
_PR = string.Template("""<h2>PR <code>${repo}</code> #${pr}</h2>
|
||||
<div class="muted">head sha: <code>${head_sha}</code></div>
|
||||
<div class="muted">posted_at: ${posted_at_iso}</div>
|
||||
<div class="muted">review_id_gitea: ${review_id_gitea} · body_comment_id: ${body_comment_id}</div>
|
||||
|
||||
<h2>Findings</h2>
|
||||
${findings_table}
|
||||
|
||||
<p><a href="/r/${repo_url}/${pr}/raw">raw review body (Markdown)</a></p>
|
||||
""")
|
||||
|
||||
|
||||
_LOGIN = string.Template("""<h2>Login</h2>
|
||||
${flash}
|
||||
<form method="post" action="/login">
|
||||
<label for="token">Dashboard token</label>
|
||||
<input type="text" id="token" name="token" autofocus>
|
||||
<div class="row">
|
||||
<button type="submit">Sign in</button>
|
||||
</div>
|
||||
</form>
|
||||
""")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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"<tr><td><a href=\"/r/{_esc(r['repo'])}\">{_esc(r['repo'])}</a></td>"
|
||||
f"<td>{int(r['run_count'])}</td>"
|
||||
f"<td class=\"muted\">{_ts_iso(int(r['last_seen']))}</td></tr>"
|
||||
for r in data.get("top_repos", [])
|
||||
) or "<tr><td class=\"muted\">no reviews yet</td></tr>"
|
||||
top_table = f"<table><thead><tr><th>repo</th><th>runs</th><th>last seen</th></tr></thead><tbody>{top_rows}</tbody></table>"
|
||||
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"<tr><td><code>{_esc(f['path'])}:{_esc(f['line'])}</code></td>"
|
||||
f"<td class=\"sev-{_esc(f.get('severity', 'low').lower())}\">{_esc(f.get('severity', ''))}</td>"
|
||||
f"<td>{_esc(f.get('problem', ''))}</td>"
|
||||
f"<td>{int(f.get('occurrences', 0))}</td>"
|
||||
f"<td>+{int(f.get('upvotes', 0))} / -{int(f.get('downvotes', 0))}</td>"
|
||||
f"<td>{'resolved' if int(f.get('resolved', 0)) else 'open'}</td>"
|
||||
f"<td>{int(f.get('reply_count', 0))}</td></tr>"
|
||||
for f in tf
|
||||
)
|
||||
top_findings_table = (
|
||||
"<table><thead><tr><th>location</th><th>severity</th>"
|
||||
"<th>problem</th><th>occurrences</th><th>votes</th>"
|
||||
"<th>state</th><th>replies</th></tr></thead><tbody>"
|
||||
f"{rows}</tbody></table>"
|
||||
)
|
||||
else:
|
||||
top_findings_table = "<p class=\"muted\">no findings yet</p>"
|
||||
|
||||
# Runs by day.
|
||||
runs = data.get("runs_by_day", [])
|
||||
if runs:
|
||||
rows = "".join(
|
||||
f"<tr><td>{_esc(r['date'])}</td><td>{int(r.get('count', 0))}</td></tr>"
|
||||
for r in runs
|
||||
)
|
||||
runs_by_day_table = (
|
||||
"<table><thead><tr><th>date</th><th>runs</th></tr></thead>"
|
||||
f"<tbody>{rows}</tbody></table>"
|
||||
)
|
||||
else:
|
||||
runs_by_day_table = "<p class=\"muted\">no runs in the last 30 days</p>"
|
||||
|
||||
# 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"<option value=\"\">— keep current ({_esc(current_model or 'unset')}) —</option>"
|
||||
+ "".join(
|
||||
f"<option value=\"{_esc(k)}\" {'selected' if k == current_model else ''}>{_esc(k)}</option>"
|
||||
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 "<p class=\"muted\">no reviews yet</p>"
|
||||
out = "<table><thead><tr><th>PR</th><th>head sha</th><th>posted</th></tr></thead><tbody>"
|
||||
for r in rows:
|
||||
out += (
|
||||
f"<tr><td><a href=\"/r/{_esc(repo_url)}/{int(r['pr'])}\">#{int(r['pr'])}</a></td>"
|
||||
f"<td><code>{_esc(r['head_sha'][:10])}</code></td>"
|
||||
f"<td class=\"muted\">{_ts_iso(int(r.get('posted_at', 0)))}</td></tr>"
|
||||
)
|
||||
out += "</tbody></table>"
|
||||
return out
|
||||
|
||||
|
||||
def _pr_body(data: dict, *, repo_url: str) -> str:
|
||||
findings = data.get("findings", [])
|
||||
if findings:
|
||||
rows = "".join(
|
||||
f"<tr><td><code>{_esc(f['path'])}:{_esc(f['line'])}</code></td>"
|
||||
f"<td class=\"sev-{_esc(f.get('severity', 'low').lower())}\">{_esc(f.get('severity', ''))}</td>"
|
||||
f"<td>{_esc(f.get('problem', ''))}</td>"
|
||||
f"<td>{_esc(f.get('fix', ''))}</td>"
|
||||
f"<td>{_esc(f.get('suggestion', ''))}</td>"
|
||||
f"<td>+{int(f.get('upvotes', 0))} / -{int(f.get('downvotes', 0))}</td>"
|
||||
f"<td>{'resolved' if int(f.get('resolved', 0)) else 'open'}</td>"
|
||||
f"<td>{int(f.get('reply_count', 0))}</td></tr>"
|
||||
for f in findings
|
||||
)
|
||||
findings_table = (
|
||||
"<table><thead><tr><th>location</th><th>severity</th>"
|
||||
"<th>problem</th><th>fix</th><th>suggestion</th>"
|
||||
"<th>votes</th><th>state</th><th>replies</th></tr></thead>"
|
||||
f"<tbody>{rows}</tbody></table>"
|
||||
)
|
||||
else:
|
||||
findings_table = "<p class=\"muted\">no findings</p>"
|
||||
|
||||
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)."""
|
||||
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/<owner>/<name> → repo
|
||||
# /r/<owner>/<name>/<index> → PR
|
||||
# /r/<owner>/<name>/<index>/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/<owner>/<name>/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())
|
||||
@@ -0,0 +1,302 @@
|
||||
"""pragent pilot — dashboard read-only query layer.
|
||||
|
||||
Three functions: overview / repo_summary / pr_summary. Each opens the SQLite
|
||||
feedback DB via `feedback.init`, runs the queries it needs, and returns plain
|
||||
dicts/lists. NEVER writes — that's the dashboard_server's job (via the Gitea
|
||||
contents API). This module is what the dashboard_server's templates render.
|
||||
|
||||
All three functions are tolerant of a missing or empty DB: they return the
|
||||
shaped dict with zeros/empty lists rather than crashing. The dashboard is a
|
||||
read-only view; the pilot can boot with no feedback DB and the dashboard
|
||||
should still load.
|
||||
|
||||
Cost note: `total_cost_usd` is hardcoded to 0.0. Per-review `usage:cost` is
|
||||
not in the feedback SQLite — only the raw `review` / `inline_finding` rows
|
||||
are stored there. The equivalent-cost calc lives in `ai_review._render_collapsible_usage`
|
||||
and only knows about the latest review's tokens. Surfacing a rolled-up dollar
|
||||
figure without per-row telemetry would be guessing, so we don't.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import os
|
||||
import sqlite3
|
||||
|
||||
from pilot import feedback
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _empty_overview() -> dict:
|
||||
return {
|
||||
"total_reviews": 0,
|
||||
"total_findings": 0,
|
||||
"total_repos": 0,
|
||||
"last_30d_reviews": 0,
|
||||
"daily": [{"date": _iso_date(i), "count": 0} for i in range(7)],
|
||||
"top_repos": [],
|
||||
"total_cost_usd": 0.0,
|
||||
}
|
||||
|
||||
|
||||
def _empty_repo_summary(repo: str) -> dict:
|
||||
return {
|
||||
"repo": repo,
|
||||
"total_runs": 0,
|
||||
"last_run_ts": 0,
|
||||
"runs_by_day": [],
|
||||
"findings_by_severity": {"critical": 0, "high": 0, "medium": 0, "low": 0},
|
||||
"top_findings": [],
|
||||
# NOTE: review rows don't carry a `model` column in the schema today,
|
||||
# so we have nothing to aggregate. When that lands, replace this
|
||||
# empty list with a `SELECT model, COUNT(*) …` over `review`.
|
||||
"models_used": [],
|
||||
}
|
||||
|
||||
|
||||
def _empty_pr_summary(repo: str, pr: int) -> dict:
|
||||
return {
|
||||
"repo": repo,
|
||||
"pr": pr,
|
||||
"head_sha": "",
|
||||
"posted_at": 0,
|
||||
"review_id_gitea": None,
|
||||
"body_comment_id": None,
|
||||
"findings": [],
|
||||
# usage isn't on the review row today; ai_review.py renders it
|
||||
# in-memory at review time. Leave empty.
|
||||
"usage": {},
|
||||
}
|
||||
|
||||
|
||||
def _iso_date(days_ago: int) -> str:
|
||||
"""Return YYYY-MM-DD for `days_ago` days before today (UTC)."""
|
||||
d = datetime.datetime.now(datetime.timezone.utc).date() - datetime.timedelta(days=days_ago)
|
||||
return d.isoformat()
|
||||
|
||||
|
||||
def _open_or_none(db_path: str) -> sqlite3.Connection | None:
|
||||
"""Open the DB if it exists and looks like a feedback DB. Else None.
|
||||
|
||||
Tolerates missing files (fresh container) and a schema-less file (the
|
||||
operator dropped a stray DB at the path). Returns a connection with
|
||||
Row factory set so callers can use `row["col"]`.
|
||||
"""
|
||||
if not db_path or not os.path.exists(db_path):
|
||||
return None
|
||||
try:
|
||||
conn = feedback.init(db_path)
|
||||
except sqlite3.DatabaseError:
|
||||
return None
|
||||
return conn
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public API
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def overview(db_path: str) -> dict:
|
||||
"""Top-of-page summary: totals + 7-bucket daily sparkline + top 5 repos."""
|
||||
conn = _open_or_none(db_path)
|
||||
if conn is None:
|
||||
return _empty_overview()
|
||||
try:
|
||||
cur = conn.execute("SELECT COUNT(*) FROM review")
|
||||
total_reviews = cur.fetchone()[0]
|
||||
cur = conn.execute("SELECT COUNT(*) FROM inline_finding")
|
||||
total_findings = cur.fetchone()[0]
|
||||
cur = conn.execute("SELECT COUNT(DISTINCT repo) FROM review")
|
||||
total_repos = cur.fetchone()[0]
|
||||
|
||||
# Last 30d window — reviews AND findings posted within the window.
|
||||
ts_30d_ago = int(datetime.datetime.now(datetime.timezone.utc).timestamp()) - 30 * 86400
|
||||
cur = conn.execute("SELECT COUNT(*) FROM review WHERE posted_at >= ?", (ts_30d_ago,))
|
||||
last_30d_reviews = cur.fetchone()[0]
|
||||
|
||||
# 7-bucket daily sparkline, oldest first. Bucket key is UTC date.
|
||||
cur = conn.execute(
|
||||
"SELECT posted_at FROM review WHERE posted_at >= ?",
|
||||
(int(datetime.datetime.now(datetime.timezone.utc).timestamp()) - 7 * 86400,),
|
||||
)
|
||||
buckets: dict[str, int] = {_iso_date(i): 0 for i in range(7)}
|
||||
for (ts,) in cur.fetchall():
|
||||
d = datetime.datetime.fromtimestamp(ts, tz=datetime.timezone.utc).date().isoformat()
|
||||
if d in buckets:
|
||||
buckets[d] += 1
|
||||
daily = [{"date": _iso_date(i), "count": buckets[_iso_date(i)]} for i in range(7)]
|
||||
|
||||
# Top 5 repos by run count, descending. last_seen is the most recent
|
||||
# review timestamp on that repo.
|
||||
cur = conn.execute(
|
||||
"SELECT repo, COUNT(*) AS runs, MAX(posted_at) AS last_seen "
|
||||
"FROM review GROUP BY repo ORDER BY runs DESC, last_seen DESC LIMIT 5"
|
||||
)
|
||||
top_repos = [
|
||||
{"repo": row[0], "run_count": row[1], "last_seen": int(row[2])}
|
||||
for row in cur.fetchall()
|
||||
]
|
||||
|
||||
return {
|
||||
"total_reviews": total_reviews,
|
||||
"total_findings": total_findings,
|
||||
"total_repos": total_repos,
|
||||
"last_30d_reviews": last_30d_reviews,
|
||||
"daily": daily,
|
||||
"top_repos": top_repos,
|
||||
"total_cost_usd": 0.0,
|
||||
}
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def repo_summary(db_path: str, repo: str) -> dict:
|
||||
"""Per-repo drill-down: runs by day, severity histogram, top findings."""
|
||||
conn = _open_or_none(db_path)
|
||||
if conn is None:
|
||||
return _empty_repo_summary(repo)
|
||||
try:
|
||||
cur = conn.execute(
|
||||
"SELECT COUNT(*), MAX(posted_at) FROM review WHERE repo = ?", (repo,)
|
||||
)
|
||||
row = cur.fetchone()
|
||||
total_runs = row[0] or 0
|
||||
last_run_ts = int(row[1]) if row[1] else 0
|
||||
|
||||
# runs_by_day for the last 30 days, oldest first; zero-buckets included.
|
||||
cur = conn.execute(
|
||||
"SELECT posted_at FROM review WHERE repo = ? AND posted_at >= ?",
|
||||
(repo, int(datetime.datetime.now(datetime.timezone.utc).timestamp()) - 30 * 86400),
|
||||
)
|
||||
buckets: dict[str, int] = {}
|
||||
for d in range(30):
|
||||
buckets[_iso_date(d)] = 0 # newest-day mapped to 0; we'll iterate
|
||||
# Re-key: build oldest-first, days_ago goes 29..0
|
||||
oldest_first = {}
|
||||
for d in range(30):
|
||||
oldest_first[_iso_date(29 - d)] = 0
|
||||
for (ts,) in cur.fetchall():
|
||||
d = datetime.datetime.fromtimestamp(ts, tz=datetime.timezone.utc).date().isoformat()
|
||||
if d in oldest_first:
|
||||
oldest_first[d] += 1
|
||||
runs_by_day = [{"date": k, "count": v} for k, v in oldest_first.items()]
|
||||
|
||||
# findings_by_severity — case-insensitive match; bucket unknown as 'low'.
|
||||
cur = conn.execute(
|
||||
"SELECT severity, COUNT(*) FROM inline_finding WHERE repo = ? GROUP BY severity",
|
||||
(repo,),
|
||||
)
|
||||
fbs = {"critical": 0, "high": 0, "medium": 0, "low": 0}
|
||||
for sev, n in cur.fetchall():
|
||||
k = (sev or "").strip().lower()
|
||||
if k not in fbs:
|
||||
k = "low"
|
||||
fbs[k] += n
|
||||
|
||||
# top_findings — top 5 posthashes by occurrence count, joined with
|
||||
# vote rollups via feedback.findings_with_votes.
|
||||
cur = conn.execute(
|
||||
"SELECT f.path, f.line, MAX(f.severity) AS severity, MAX(f.problem) AS problem, "
|
||||
"COUNT(*) AS occurrences, "
|
||||
"COALESCE(SUM(CASE WHEN rct.content = '+1' THEN 1 ELSE 0 END), 0) AS upvotes, "
|
||||
"COALESCE(SUM(CASE WHEN rct.content = '-1' THEN 1 ELSE 0 END), 0) AS downvotes, "
|
||||
"MAX(ts.resolved) AS resolved, "
|
||||
"COALESCE((SELECT COUNT(*) FROM reply WHERE finding_id IN "
|
||||
" (SELECT id FROM inline_finding WHERE posthash = f.posthash AND repo = f.repo AND path = f.path AND line = f.line)), 0) AS reply_count "
|
||||
"FROM inline_finding f "
|
||||
"LEFT JOIN reaction rct ON rct.comment_id = f.comment_id "
|
||||
"LEFT JOIN thread_state ts ON ts.finding_id = f.id "
|
||||
"WHERE f.repo = ? "
|
||||
"GROUP BY f.posthash, f.repo, f.path, f.line "
|
||||
"ORDER BY occurrences DESC, upvotes DESC LIMIT 5",
|
||||
(repo,),
|
||||
)
|
||||
top_findings = [
|
||||
{
|
||||
"path": r[0],
|
||||
"line": r[1],
|
||||
"severity": r[2],
|
||||
"problem": r[3],
|
||||
"occurrences": r[4],
|
||||
"upvotes": int(r[5] or 0),
|
||||
"downvotes": int(r[6] or 0),
|
||||
"resolved": int(r[7] or 0),
|
||||
"reply_count": int(r[8] or 0),
|
||||
}
|
||||
for r in cur.fetchall()
|
||||
]
|
||||
|
||||
return {
|
||||
"repo": repo,
|
||||
"total_runs": total_runs,
|
||||
"last_run_ts": last_run_ts,
|
||||
"runs_by_day": runs_by_day,
|
||||
"findings_by_severity": fbs,
|
||||
"top_findings": top_findings,
|
||||
"models_used": [], # see _empty_repo_summary NOTE
|
||||
}
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def pr_summary(db_path: str, repo: str, pr: int) -> dict:
|
||||
"""Per-PR view: meta + every finding the bot ever posted on that PR."""
|
||||
conn = _open_or_none(db_path)
|
||||
if conn is None:
|
||||
return _empty_pr_summary(repo, pr)
|
||||
try:
|
||||
cur = conn.execute(
|
||||
"SELECT head_sha, posted_at, review_id_gitea, body_comment_id "
|
||||
"FROM review WHERE repo = ? AND pr = ? ORDER BY posted_at DESC LIMIT 1",
|
||||
(repo, pr),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
if row is None:
|
||||
return _empty_pr_summary(repo, pr)
|
||||
head_sha, posted_at, review_id_gitea, body_comment_id = row
|
||||
|
||||
cur = conn.execute(
|
||||
"SELECT f.path, f.line, f.severity, f.problem, f.fix, f.suggestion, "
|
||||
"COALESCE(SUM(CASE WHEN rct.content = '+1' THEN 1 ELSE 0 END), 0) AS upvotes, "
|
||||
"COALESCE(SUM(CASE WHEN rct.content = '-1' THEN 1 ELSE 0 END), 0) AS downvotes, "
|
||||
"MAX(ts.resolved) AS resolved, "
|
||||
"COALESCE((SELECT COUNT(*) FROM reply WHERE finding_id = f.id), 0) AS reply_count "
|
||||
"FROM inline_finding f "
|
||||
"LEFT JOIN reaction rct ON rct.comment_id = f.comment_id "
|
||||
"LEFT JOIN thread_state ts ON ts.finding_id = f.id "
|
||||
"WHERE f.repo = ? AND f.pr = ? "
|
||||
"GROUP BY f.id "
|
||||
"ORDER BY f.path, f.line",
|
||||
(repo, pr),
|
||||
)
|
||||
findings = [
|
||||
{
|
||||
"path": r[0],
|
||||
"line": r[1],
|
||||
"severity": r[2],
|
||||
"problem": r[3],
|
||||
"fix": r[4],
|
||||
"suggestion": r[5],
|
||||
"upvotes": int(r[6] or 0),
|
||||
"downvotes": int(r[7] or 0),
|
||||
"resolved": int(r[8] or 0),
|
||||
"reply_count": int(r[9] or 0),
|
||||
}
|
||||
for r in cur.fetchall()
|
||||
]
|
||||
|
||||
return {
|
||||
"repo": repo,
|
||||
"pr": pr,
|
||||
"head_sha": head_sha,
|
||||
"posted_at": int(posted_at),
|
||||
"review_id_gitea": review_id_gitea,
|
||||
"body_comment_id": body_comment_id,
|
||||
"findings": findings,
|
||||
"usage": {},
|
||||
}
|
||||
finally:
|
||||
conn.close()
|
||||
@@ -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()
|
||||
@@ -0,0 +1,249 @@
|
||||
"""Tests for pilot/dashboard_data.py — read-only query layer over the feedback SQLite.
|
||||
|
||||
Covers: empty-DB fallbacks (no crash on missing/empty DB), overview rollups,
|
||||
per-repo drill-down (findings by severity, top findings, runs by day), and
|
||||
the per-PR view. The dashboard never writes — only reads.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
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(HERE, "..", "..")) # so `from pilot import …` works
|
||||
|
||||
from pilot import dashboard_data, feedback
|
||||
|
||||
|
||||
def _seed_repo(conn, *, repo: str, prs: int, findings_per_pr: int, day_offset: int = 0):
|
||||
"""Seed one repo with `prs` PRs each with `findings_per_pr` findings.
|
||||
|
||||
All timestamps cluster on (now - day_offset days). Returns list of review ids.
|
||||
"""
|
||||
base = int(time.time()) - day_offset * 86400
|
||||
rids = []
|
||||
for n in range(prs):
|
||||
rid = feedback.record_review(
|
||||
conn, repo=repo, pr=n + 1, head_sha=f"sha{n}",
|
||||
review_id_gitea=1000 + n, body_comment_id=2000 + n,
|
||||
posted_at=base + n * 60,
|
||||
)
|
||||
rids.append(rid)
|
||||
for k in range(findings_per_pr):
|
||||
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", "low"][k % 4],
|
||||
problem=f"problem {k}",
|
||||
fix=f"fix {k}", suggestion=f"suggestion {k}",
|
||||
comment_id=3000 + n * 10 + k,
|
||||
posted_at=base + n * 60,
|
||||
)
|
||||
return rids
|
||||
|
||||
|
||||
class TestEmptyDB(unittest.TestCase):
|
||||
def test_missing_file_returns_zero_dict(self):
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
missing = f"{d}/nope.db"
|
||||
data = dashboard_data.overview(missing)
|
||||
self.assertEqual(data["total_reviews"], 0)
|
||||
self.assertEqual(data["total_findings"], 0)
|
||||
self.assertEqual(data["total_repos"], 0)
|
||||
self.assertEqual(data["last_30d_reviews"], 0)
|
||||
self.assertEqual(len(data["daily"]), 7)
|
||||
self.assertEqual(data["top_repos"], [])
|
||||
self.assertEqual(data["total_cost_usd"], 0.0)
|
||||
|
||||
def test_missing_file_repo_summary_safe(self):
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
data = dashboard_data.repo_summary(f"{d}/nope.db", "o/r")
|
||||
self.assertEqual(data["repo"], "o/r")
|
||||
self.assertEqual(data["total_runs"], 0)
|
||||
self.assertEqual(data["runs_by_day"], [])
|
||||
for sev in ("critical", "high", "medium", "low"):
|
||||
self.assertEqual(data["findings_by_severity"][sev], 0)
|
||||
self.assertEqual(data["top_findings"], [])
|
||||
self.assertEqual(data["models_used"], [])
|
||||
|
||||
def test_missing_file_pr_summary_safe(self):
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
data = dashboard_data.pr_summary(f"{d}/nope.db", "o/r", 1)
|
||||
self.assertEqual(data["repo"], "o/r")
|
||||
self.assertEqual(data["pr"], 1)
|
||||
self.assertEqual(data["findings"], [])
|
||||
self.assertEqual(data["usage"], {})
|
||||
|
||||
|
||||
class TestEmptyButExistingDB(unittest.TestCase):
|
||||
"""`init` creates the schema — DB exists but has no rows."""
|
||||
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db = f"{self.tmp.name}/f.db"
|
||||
feedback.init(self.db)
|
||||
|
||||
def tearDown(self):
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_overview_is_zero(self):
|
||||
data = dashboard_data.overview(self.db)
|
||||
self.assertEqual(data["total_reviews"], 0)
|
||||
self.assertEqual(data["total_findings"], 0)
|
||||
self.assertEqual(data["total_repos"], 0)
|
||||
|
||||
def test_repo_summary_is_zero(self):
|
||||
data = dashboard_data.repo_summary(self.db, "o/r")
|
||||
self.assertEqual(data["total_runs"], 0)
|
||||
self.assertEqual(data["findings_by_severity"], {"critical": 0, "high": 0, "medium": 0, "low": 0})
|
||||
|
||||
def test_pr_summary_is_zero(self):
|
||||
data = dashboard_data.pr_summary(self.db, "o/r", 1)
|
||||
self.assertEqual(data["findings"], [])
|
||||
|
||||
|
||||
class TestOverview(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db = f"{self.tmp.name}/f.db"
|
||||
self.conn = feedback.init(self.db)
|
||||
_seed_repo(self.conn, repo="alpha/one", prs=3, findings_per_pr=2)
|
||||
_seed_repo(self.conn, repo="beta/two", prs=1, findings_per_pr=4)
|
||||
self.conn.close()
|
||||
|
||||
def tearDown(self):
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_totals(self):
|
||||
data = dashboard_data.overview(self.db)
|
||||
self.assertEqual(data["total_reviews"], 4)
|
||||
self.assertEqual(data["total_findings"], 6 + 4) # 3*2 + 1*4 = 10
|
||||
self.assertEqual(data["total_repos"], 2)
|
||||
self.assertEqual(data["total_cost_usd"], 0.0)
|
||||
|
||||
def test_top_repos_sorted_by_run_count(self):
|
||||
data = dashboard_data.overview(self.db)
|
||||
repos = [r["repo"] for r in data["top_repos"]]
|
||||
# alpha/one has 3 runs, beta/two has 1.
|
||||
self.assertEqual(repos[0], "alpha/one")
|
||||
self.assertEqual(data["top_repos"][0]["run_count"], 3)
|
||||
self.assertEqual(data["top_repos"][1]["run_count"], 1)
|
||||
# last_seen is a unix timestamp int.
|
||||
for r in data["top_repos"]:
|
||||
self.assertIsInstance(r["last_seen"], int)
|
||||
|
||||
def test_daily_buckets_are_7(self):
|
||||
data = dashboard_data.overview(self.db)
|
||||
self.assertEqual(len(data["daily"]), 7)
|
||||
for b in data["daily"]:
|
||||
self.assertIn("date", b)
|
||||
self.assertIn("count", b)
|
||||
|
||||
def test_last_30d_reviews(self):
|
||||
data = dashboard_data.overview(self.db)
|
||||
self.assertEqual(data["last_30d_reviews"], 4)
|
||||
|
||||
|
||||
class TestRepoSummary(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db = f"{self.tmp.name}/f.db"
|
||||
self.conn = feedback.init(self.db)
|
||||
# 4 PRs with 2 findings each → 8 findings, severity cycle [c,h,m,l,c,h,m,l]
|
||||
_seed_repo(self.conn, repo="o/r", prs=4, findings_per_pr=2)
|
||||
# Add some reactions so top_findings has signal.
|
||||
rows = self.conn.execute(
|
||||
"SELECT id, comment_id FROM inline_finding WHERE repo=? ORDER BY id LIMIT 3",
|
||||
("o/r",),
|
||||
).fetchall()
|
||||
for r in rows:
|
||||
feedback.record_reaction(self.conn, comment_id=r["comment_id"], user="u", content="+1")
|
||||
self.conn.close()
|
||||
|
||||
def tearDown(self):
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_basic_shape(self):
|
||||
data = dashboard_data.repo_summary(self.db, "o/r")
|
||||
self.assertEqual(data["repo"], "o/r")
|
||||
self.assertEqual(data["total_runs"], 4)
|
||||
self.assertIsInstance(data["last_run_ts"], int)
|
||||
|
||||
def test_findings_by_severity(self):
|
||||
data = dashboard_data.repo_summary(self.db, "o/r")
|
||||
fbs = data["findings_by_severity"]
|
||||
# 4 PRs × 2 findings; per-PR severities are [critical, high].
|
||||
# (k in range(2) → k=0 critical, k=1 high for every PR.)
|
||||
self.assertEqual(fbs["critical"], 4)
|
||||
self.assertEqual(fbs["high"], 4)
|
||||
self.assertEqual(fbs["medium"], 0)
|
||||
self.assertEqual(fbs["low"], 0)
|
||||
|
||||
def test_runs_by_day_is_list(self):
|
||||
data = dashboard_data.repo_summary(self.db, "o/r")
|
||||
self.assertIsInstance(data["runs_by_day"], list)
|
||||
for r in data["runs_by_day"]:
|
||||
self.assertIn("date", r)
|
||||
self.assertIn("count", r)
|
||||
|
||||
def test_top_findings_structure(self):
|
||||
data = dashboard_data.repo_summary(self.db, "o/r")
|
||||
self.assertGreater(len(data["top_findings"]), 0)
|
||||
first = data["top_findings"][0]
|
||||
for k in ("path", "line", "severity", "problem", "occurrences", "upvotes", "downvotes", "resolved", "reply_count"):
|
||||
self.assertIn(k, first)
|
||||
|
||||
def test_models_used_is_empty_list_with_note(self):
|
||||
# The schema has no `model` column on review — the dashboard can't show
|
||||
# model usage from this DB today. We document that via an empty list.
|
||||
data = dashboard_data.repo_summary(self.db, "o/r")
|
||||
self.assertEqual(data["models_used"], [])
|
||||
|
||||
|
||||
class TestPRSummary(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db = f"{self.tmp.name}/f.db"
|
||||
self.conn = feedback.init(self.db)
|
||||
rid = feedback.record_review(
|
||||
self.conn, repo="o/r", pr=42, head_sha="abc",
|
||||
review_id_gitea=9001, body_comment_id=8001,
|
||||
posted_at=1700000000,
|
||||
)
|
||||
for k in range(3):
|
||||
feedback.record_inline_finding(
|
||||
self.conn, review_id=rid, repo="o/r", pr=42,
|
||||
path=f"src/x_{k}.py", line=k + 10,
|
||||
severity=["critical", "high", "low"][k],
|
||||
problem=f"p{k}", fix=f"f{k}", suggestion=f"s{k}",
|
||||
comment_id=7000 + k,
|
||||
)
|
||||
self.conn.close()
|
||||
|
||||
def tearDown(self):
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_meta(self):
|
||||
data = dashboard_data.pr_summary(self.db, "o/r", 42)
|
||||
self.assertEqual(data["repo"], "o/r")
|
||||
self.assertEqual(data["pr"], 42)
|
||||
self.assertEqual(data["head_sha"], "abc")
|
||||
self.assertEqual(data["review_id_gitea"], 9001)
|
||||
self.assertEqual(data["body_comment_id"], 8001)
|
||||
self.assertEqual(data["posted_at"], 1700000000)
|
||||
# usage is empty because the schema has no usage column.
|
||||
self.assertEqual(data["usage"], {})
|
||||
|
||||
def test_findings(self):
|
||||
data = dashboard_data.pr_summary(self.db, "o/r", 42)
|
||||
self.assertEqual(len(data["findings"]), 3)
|
||||
for f in data["findings"]:
|
||||
for k in ("path", "line", "severity", "problem", "fix", "suggestion",
|
||||
"upvotes", "downvotes", "resolved", "reply_count"):
|
||||
self.assertIn(k, f)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,205 @@
|
||||
"""Tests for the edit endpoint — POST /r/<owner>/<name>/edit (Task C).
|
||||
|
||||
We mock the Gitea HTTP layer (urllib.request.urlopen) so the test never
|
||||
touches the network. The dashboard handler is responsible for:
|
||||
* auth + CSRF
|
||||
* read .pr-review.json via GET (404 → start from {})
|
||||
* validate model against cost_model.PRICES
|
||||
* PUT the updated file back, with sha + base64 content
|
||||
* redirect to /r/<owner>/<name> on success
|
||||
"""
|
||||
import base64
|
||||
import http.client
|
||||
import json
|
||||
import os
|
||||
import socket
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
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
|
||||
|
||||
|
||||
def _post(port: int, path: str, body: bytes, *, cookie: str | None = None) -> tuple[int, dict, bytes]:
|
||||
conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
|
||||
headers = {"Content-Type": "application/x-www-form-urlencoded"}
|
||||
if cookie:
|
||||
headers["Cookie"] = cookie
|
||||
conn.request("POST", path, body=body, headers=headers)
|
||||
r = conn.getresponse()
|
||||
body_b = r.read()
|
||||
h = dict(r.getheaders())
|
||||
conn.close()
|
||||
return r.status, h, body_b
|
||||
|
||||
|
||||
class _FakeResp:
|
||||
def __init__(self, status: int, body: bytes):
|
||||
self.status = status
|
||||
self._body = body
|
||||
|
||||
def read(self):
|
||||
return self._body
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *a):
|
||||
return False
|
||||
|
||||
|
||||
class TestDashboardEdit(unittest.TestCase):
|
||||
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-edit-token"
|
||||
os.environ["PRAGENT_BOT_TOKEN"] = "bot-token"
|
||||
# Seed a row so the repo page is meaningful.
|
||||
conn = feedback.init(self.db)
|
||||
feedback.record_review(
|
||||
conn, repo="o/r", pr=1, head_sha="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()
|
||||
self.cookie = "pragent_dash=secret-edit-token"
|
||||
|
||||
# Records of HTTP calls made by the handler.
|
||||
self.calls: list[tuple[str, str, dict | None, bytes | None]] = []
|
||||
|
||||
def tearDown(self):
|
||||
self.srv.shutdown()
|
||||
self.srv.server_close()
|
||||
self.thread.join(timeout=2)
|
||||
for k in ("PRAGENT_FEEDBACK_DB", "PRAGENT_DASHBOARD_TOKEN", "PRAGENT_BOT_TOKEN"):
|
||||
os.environ.pop(k, None)
|
||||
self.tmp.cleanup()
|
||||
|
||||
def _urlopen(self, req, timeout=30):
|
||||
"""Replacement for urllib.request.urlopen that the handler uses."""
|
||||
url = req.full_url if hasattr(req, "full_url") else req
|
||||
method = getattr(req, "method", None) or "GET"
|
||||
body = getattr(req, "data", None)
|
||||
headers = dict(getattr(req, "headers", {}) or {})
|
||||
self.calls.append((method, url, headers, body))
|
||||
# Route based on URL: GET contents/.../raw vs PUT contents/.pr-review.json
|
||||
if method == "GET" and ".pr-review.json" in url:
|
||||
return _FakeResp(200, json.dumps({
|
||||
"content": base64.b64encode(b'{"focus":["x"],"model":"claude-haiku-4-5"}').decode(),
|
||||
"sha": "deadbeef",
|
||||
}).encode())
|
||||
if method == "PUT" and ".pr-review.json" in url:
|
||||
return _FakeResp(200, b'{}')
|
||||
return _FakeResp(404, b'{"message":"not found"}')
|
||||
|
||||
def test_edit_updates_static_message_and_model(self):
|
||||
form = (
|
||||
b"_csrf=secret-edit-token"
|
||||
b"&static_message=Hello%20world"
|
||||
b"&model=claude-sonnet-5"
|
||||
)
|
||||
with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
|
||||
status, h, _b = _post(self.port, "/r/o/r/edit", form, cookie=self.cookie)
|
||||
self.assertEqual(status, 302)
|
||||
self.assertEqual(h.get("Location"), "/r/o/r")
|
||||
|
||||
# Find the PUT call.
|
||||
put_calls = [c for c in self.calls if c[0] == "PUT"]
|
||||
self.assertEqual(len(put_calls), 1, self.calls)
|
||||
method, url, _hdrs, body = put_calls[0]
|
||||
self.assertIn(".pr-review.json", url)
|
||||
payload = json.loads(body)
|
||||
self.assertIn("content", payload)
|
||||
self.assertEqual(payload["sha"], "deadbeef")
|
||||
decoded = base64.b64decode(payload["content"]).decode()
|
||||
cfg = json.loads(decoded)
|
||||
self.assertEqual(cfg.get("static_message"), "Hello world")
|
||||
self.assertEqual(cfg.get("model"), "claude-sonnet-5")
|
||||
|
||||
def test_edit_strips_static_message_to_400(self):
|
||||
long_msg = "x" * 600
|
||||
form = (
|
||||
f"_csrf=secret-edit-token"
|
||||
f"&static_message={long_msg}"
|
||||
f"&model=claude-haiku-4-5"
|
||||
).encode()
|
||||
with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
|
||||
_post(self.port, "/r/o/r/edit", form, cookie=self.cookie)
|
||||
put = next(c for c in self.calls if c[0] == "PUT")
|
||||
cfg = json.loads(base64.b64decode(json.loads(put[3])["content"]))
|
||||
self.assertEqual(len(cfg["static_message"]), 400)
|
||||
|
||||
def test_edit_rejects_unknown_model_with_flash(self):
|
||||
form = (
|
||||
b"_csrf=secret-edit-token"
|
||||
b"&static_message=hi"
|
||||
b"&model=does-not-exist"
|
||||
)
|
||||
with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
|
||||
status, h, _b = _post(self.port, "/r/o/r/edit", form, cookie=self.cookie)
|
||||
self.assertEqual(status, 302)
|
||||
self.assertIn("flash=", h.get("Location", ""))
|
||||
# No PUT should have been issued.
|
||||
put_calls = [c for c in self.calls if c[0] == "PUT"]
|
||||
self.assertEqual(put_calls, [])
|
||||
|
||||
def test_edit_requires_auth(self):
|
||||
form = b"_csrf=secret-edit-token&static_message=x&model=claude-haiku-4-5"
|
||||
with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
|
||||
status, h, _b = _post(self.port, "/r/o/r/edit", form)
|
||||
self.assertEqual(status, 302)
|
||||
self.assertEqual(h.get("Location"), "/login")
|
||||
|
||||
def test_edit_csrf_mismatch_redirects_without_save(self):
|
||||
form = b"_csrf=wrong&static_message=x&model=claude-haiku-4-5"
|
||||
with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
|
||||
status, h, _b = _post(self.port, "/r/o/r/edit", form, cookie=self.cookie)
|
||||
self.assertEqual(status, 302)
|
||||
self.assertEqual(h.get("Location"), "/r/o/r")
|
||||
put_calls = [c for c in self.calls if c[0] == "PUT"]
|
||||
self.assertEqual(put_calls, [])
|
||||
|
||||
def test_edit_creates_file_when_missing(self):
|
||||
"""When GET returns 404, the PUT must still happen (no sha)."""
|
||||
def _route(req, timeout=30):
|
||||
url = req.full_url
|
||||
method = getattr(req, "method", None) or "GET"
|
||||
body = getattr(req, "data", None)
|
||||
self.calls.append((method, url, {}, body))
|
||||
if method == "GET" and ".pr-review.json" in url:
|
||||
return _FakeResp(404, b'{"message":"not found"}')
|
||||
if method == "PUT" and ".pr-review.json" in url:
|
||||
return _FakeResp(201, b"{}")
|
||||
return _FakeResp(404, b"")
|
||||
|
||||
form = b"_csrf=secret-edit-token&static_message=hi&model=claude-haiku-4-5"
|
||||
with patch.object(dash.urllib.request, "urlopen", side_effect=_route):
|
||||
status, h, _b = _post(self.port, "/r/o/r/edit", form, cookie=self.cookie)
|
||||
self.assertEqual(status, 302)
|
||||
put = next(c for c in self.calls if c[0] == "PUT")
|
||||
payload = json.loads(put[3])
|
||||
self.assertNotIn("sha", payload, "missing-file PUT should omit sha")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,74 @@
|
||||
"""Tests for the model <select> in the repo edit form (Task D)."""
|
||||
import http.client
|
||||
import os
|
||||
import socket
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
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 cost_model, 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
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
class TestRepoEditSelect(unittest.TestCase):
|
||||
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["PRAGENT_BOT_TOKEN"] = ""
|
||||
conn = feedback.init(self.db)
|
||||
feedback.record_review(conn, repo="o/r", pr=1, head_sha="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", "PRAGENT_BOT_TOKEN"):
|
||||
os.environ.pop(k, None)
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_repo_page_renders_select_with_one_option_per_price(self):
|
||||
status, _h, body = _get(self.port, "/r/o/r")
|
||||
self.assertEqual(status, 200)
|
||||
text = body.decode()
|
||||
self.assertIn('<select id="model" name="model">', text)
|
||||
# Every PRICES key should appear as an <option value="…">.
|
||||
for k in sorted(cost_model.PRICES):
|
||||
self.assertIn(f'<option value="{k}"', text, f"missing {k} in select")
|
||||
# Plus the "keep current" placeholder.
|
||||
self.assertIn("— keep current", text)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user