feat(dashboard): mutate .pr-review.json via Gitea contents API

POST /r/<owner>/<name>/edit reads the current .pr-review.json via the
Gitea contents API (404 → start from {}), updates static_message and
model, then PUTs the file back as the bot identity. Static message is
stripped and capped at 400 chars; model is validated against
cost_model.PRICES (unknown → flash + redirect, no save). CSRF token is
the dashboard auth token (same value, simplest binding).

Helpers (_http) moved to urllib.request at module top so tests can patch
the entry point; the helper returns the Gitea status code so 404 (file
missing) cleanly omits sha on the create-PUT path.

6 tests under tests/pilot/test_dashboard_edit.py covering the happy
path, the 400-char cap, the unknown-model rejection, auth + CSRF gates,
and the file-missing creation path.

pilot/dashboard.py

@@ -34,7 +34,9 @@ import html
 import json
 import os
 import string
+import urllib.error
 import urllib.parse
+import urllib.request
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 from pilot import dashboard_data
@@ -439,7 +441,6 @@ def _http(method: str, url: str, *, token: str = "", body: dict | None = None,
           raw_body: bytes | None = None) -> tuple[int, bytes]:
     """Like ai_review._http but local: this module is stdlib-only and doesn't
     depend on the ai_review import (which pulls in a 1700-line reviewer)."""
-    import urllib.request
     headers = {"Accept": "application/json"}
     data: bytes | None = None
     if raw_body is not None:

tests/pilot/test_dashboard_edit.py

@@ -0,0 +1,205 @@
+"""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 + 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, *, cookie: str | None = None) -> tuple[int, dict, bytes]:
+    conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
+    headers = {"Content-Type": "application/x-www-form-urlencoded"}
+    if cookie:
+        headers["Cookie"] = cookie
+    conn.request("POST", path, body=body, headers=headers)
+    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_DASHBOARD_TOKEN"] = "secret-edit-token"
+        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()
+        self.cookie = "pragent_dash=secret-edit-token"
+
+        # 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_DASHBOARD_TOKEN", "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 = (
+            b"_csrf=secret-edit-token"
+            b"&static_message=Hello%20world"
+            b"&model=claude-sonnet-5"
+        )
+        with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
+            status, h, _b = _post(self.port, "/r/o/r/edit", form, cookie=self.cookie)
+        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=secret-edit-token"
+            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, cookie=self.cookie)
+        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 = (
+            b"_csrf=secret-edit-token"
+            b"&static_message=hi"
+            b"&model=does-not-exist"
+        )
+        with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
+            status, h, _b = _post(self.port, "/r/o/r/edit", form, cookie=self.cookie)
+        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 = b"_csrf=secret-edit-token&static_message=x&model=claude-haiku-4-5"
+        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, 302)
+        self.assertEqual(h.get("Location"), "/login")
+
+    def test_edit_csrf_mismatch_redirects_without_save(self):
+        form = b"_csrf=wrong&static_message=x&model=claude-haiku-4-5"
+        with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
+            status, h, _b = _post(self.port, "/r/o/r/edit", form, cookie=self.cookie)
+        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 = b"_csrf=secret-edit-token&static_message=hi&model=claude-haiku-4-5"
+        with patch.object(dash.urllib.request, "urlopen", side_effect=_route):
+            status, h, _b = _post(self.port, "/r/o/r/edit", form, cookie=self.cookie)
+        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()