"""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()