refactor: split pilot architecture

Remove the obsolete dashboard now that Langfuse is the analytics surface.\nIntroduce focused transport, model, and configuration modules while preserving the ai_review facade, and document the current runtime architecture.

tests/pilot/test_dashboard.py

@@ -1,203 +0,0 @@
-"""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()

tests/pilot/test_dashboard_data.py

@@ -1,249 +0,0 @@
-"""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()

tests/pilot/test_dashboard_edit.py

@@ -1,209 +0,0 @@
-"""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 (X-Forwarded-User set by oauth2-proxy) + 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, *, 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 _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_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()
-        # Pull the per-process CSRF secret from the rendered repo page — the
-        # edit form embeds the same token as a hidden input.
-        self.csrf = dash._CSRF_SECRET
-        self.auth_hdr = {"X-Forwarded-User": "marcos@example.com"}
-
-        # 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_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 = (
-            f"_csrf={self.csrf}"
-            f"&static_message=Hello%20world"
-            f"&model=claude-sonnet-5"
-        ).encode()
-        with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
-            status, h, _b = _post(self.port, "/r/o/r/edit", form, headers=self.auth_hdr)
-        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={self.csrf}"
-            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, headers=self.auth_hdr)
-        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 = (
-            f"_csrf={self.csrf}"
-            f"&static_message=hi"
-            f"&model=does-not-exist"
-        ).encode()
-        with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
-            status, h, _b = _post(self.port, "/r/o/r/edit", form, headers=self.auth_hdr)
-        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 = f"_csrf={self.csrf}&static_message=x&model=claude-haiku-4-5".encode()
-        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, 401)
-        self.assertEqual(h.get("WWW-Authenticate"), 'Basic realm="pragent-dashboard"')
-        # No Gitea calls at all — auth gate fires first.
-        self.assertEqual(self.calls, [])
-
-    def test_edit_csrf_mismatch_redirects_without_save(self):
-        form = f"_csrf=wrong&static_message=x&model=claude-haiku-4-5".encode()
-        with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
-            status, h, _b = _post(self.port, "/r/o/r/edit", form, headers=self.auth_hdr)
-        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 = f"_csrf={self.csrf}&static_message=hi&model=claude-haiku-4-5".encode()
-        with patch.object(dash.urllib.request, "urlopen", side_effect=_route):
-            status, h, _b = _post(self.port, "/r/o/r/edit", form, headers=self.auth_hdr)
-        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()

tests/pilot/test_dashboard_select.py

@@ -1,74 +0,0 @@
-"""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_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()
-        self.auth_hdr = {"X-Forwarded-User": "marcos@example.com"}
-
-    def tearDown(self):
-        self.srv.shutdown()
-        self.srv.server_close()
-        self.thread.join(timeout=2)
-        for k in ("PRAGENT_FEEDBACK_DB", "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", headers=self.auth_hdr)
-        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()