pragent/tests/pilot/test_dashboard.py

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