Claude
be24ef245c
Mirrors webhook_server.py's BaseHTTPRequestHandler + ThreadingHTTPServer shape. Pure stdlib, no pip deps. Routes: GET / overview (totals + 7-day sparkline + top repos) GET /r/<owner>/<name> repo summary (severity histogram + top findings) GET /r/<owner>/<name>/<index> one PR's findings GET /r/<owner>/<name>/<index>/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/<owner>/<name>/edit (Tasks C+D) Auth: when PRAGENT_DASHBOARD_TOKEN is set, every route except /login and /static/* requires Cookie: pragent_dash=<token>. 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.
231 lines
8.1 KiB
Python
231 lines
8.1 KiB
Python
"""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() |