"""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. Auth is now performed by oauth2-proxy: the dashboard trusts `X-Forwarded-User` set by the proxy and returns 401 (with a Basic challenge) when the header is missing. 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 TestDashboardAuth(unittest.TestCase): """Auth gate: require X-Forwarded-User (set by oauth2-proxy). When the header is missing every non-static route returns 401 with a Basic challenge, which lets oauth2-proxy redirect the browser to Logto. Static is exempt so the unauthenticated probe traffic doesn't loop the proxy through the auth flow. """ def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db = f"{self.tmp.name}/f.db" os.environ["PRAGENT_FEEDBACK_DB"] = self.db os.environ["DASHBOARD_PORT"] = str(0) # we override below 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", "DASHBOARD_PORT"): os.environ.pop(k, None) self.tmp.cleanup() def test_anonymous_overview_returns_401_with_basic_challenge(self): status, h, body = _get(self.port, "/") self.assertEqual(status, 401) self.assertEqual(h.get("WWW-Authenticate"), 'Basic realm="pragent-dashboard"') self.assertEqual(body, b"unauthorized\n") def test_anonymous_repo_returns_401(self): status, _h, _body = _get(self.port, "/r/alpha/one") self.assertEqual(status, 401) def test_anonymous_post_returns_401(self): status, _h, _body = _post(self.port, "/r/alpha/one/edit", b"x=1") self.assertEqual(status, 401) def test_authenticated_overview_succeeds(self): status, h, body = _get(self.port, "/", headers={"X-Forwarded-User": "marcos@example.com"}) self.assertEqual(status, 200) self.assertIn(b"Overview", body) def test_static_does_not_require_auth(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_empty_x_forwarded_user_treated_as_anonymous(self): status, _h, _body = _get(self.port, "/", headers={"X-Forwarded-User": " "}) self.assertEqual(status, 401) class TestDashboardRender(unittest.TestCase): """Render-only tests — X-Forwarded-User set, real seeded data.""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db = f"{self.tmp.name}/f.db" os.environ["PRAGENT_FEEDBACK_DB"] = self.db # 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() 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.auth_hdr = {"X-Forwarded-User": "marcos@example.com"} def tearDown(self): self.srv.shutdown() self.srv.server_close() self.thread.join(timeout=2) os.environ.pop("PRAGENT_FEEDBACK_DB", None) self.tmp.cleanup() def test_overview_200_contains_repo_names(self): status, _h, body = _get(self.port, "/", headers=self.auth_hdr) 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", headers=self.auth_hdr) 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", headers=self.auth_hdr) 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_unknown_route_404(self): status, _h, _body = _get(self.port, "/no/such/route", headers=self.auth_hdr) self.assertEqual(status, 404) if __name__ == "__main__": unittest.main()