Claude
210 lines
8.2 KiB
Python
210 lines
8.2 KiB
Python
"""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()
|