97 lines
3.6 KiB
Python
97 lines
3.6 KiB
Python
"""Tests for pilot/feedback_post.py — report delivery to Gitea.
|
|
|
|
Mock `ai_review.gitea_get` + `gitea_post` so we exercise the find-or-create
|
|
+ comment-post flow without hitting the real API.
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, os.path.join(HERE, "..", "..", "pilot"))
|
|
|
|
import feedback # noqa: E402
|
|
import feedback_analyze # noqa: E402
|
|
import feedback_post # noqa: E402
|
|
|
|
|
|
def _make_fake(method_routes: dict):
|
|
"""`method_routes` maps HTTP path substring → (status, body, method).
|
|
|
|
For our purposes both gitea_get and gitea_post share the same fake —
|
|
gitea_get is GET, gitea_post is POST, and the post helper also has a
|
|
body param. The fake returns whatever the route's body says.
|
|
"""
|
|
def fake_get(api, repo, path, token, accept="application/json"):
|
|
for needle in sorted(method_routes.keys(), key=len, reverse=True):
|
|
status, body, _m = method_routes[needle]
|
|
if needle in path:
|
|
return status, json.dumps(body).encode()
|
|
return 404, b'{"message":"not found"}'
|
|
|
|
def fake_post(api, repo, path, token, body):
|
|
for needle in sorted(method_routes.keys(), key=len, reverse=True):
|
|
status, resp_body, _m = method_routes[needle]
|
|
if needle in path:
|
|
return status, json.dumps(resp_body).encode()
|
|
return 404, b'{"message":"not found"}'
|
|
|
|
return fake_get, fake_post
|
|
|
|
|
|
class TestDeliver(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db = f"{self.tmp.name}/f.db"
|
|
conn = feedback.init(self.db)
|
|
rid = feedback.record_review(conn, repo="o/r", pr=1, head_sha="x")
|
|
feedback.record_inline_finding(
|
|
conn, review_id=rid, repo="o/r", pr=1,
|
|
path="a.ts", line=1, severity="HIGH",
|
|
problem="x", comment_id=99,
|
|
)
|
|
conn.close()
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def test_creates_issue_then_posts_comment(self):
|
|
routes = {
|
|
"issues?state=open": (200, [], "GET"), # no existing issue
|
|
"issues": (201, {"id": 42, "number": 7, "title": "..."}, "POST"),
|
|
"issues/7/comments": (201, {"id": 777}, "POST"),
|
|
}
|
|
fake_get, fake_post = _make_fake(routes)
|
|
with patch("ai_review.gitea_get", side_effect=fake_get), \
|
|
patch("ai_review.gitea_post", side_effect=fake_post):
|
|
stats = feedback_post.deliver(
|
|
api="http://x", token="t", db_path=self.db,
|
|
repo="gitea_admin/pragent", title="pragent feedback roll-up",
|
|
)
|
|
self.assertEqual(stats["issue_id"], 7)
|
|
self.assertEqual(stats["comment_id"], 777)
|
|
|
|
def test_reuses_existing_issue(self):
|
|
routes = {
|
|
"issues?state=open": (200, [
|
|
{"id": 99, "number": 9, "title": "pragent feedback roll-up"},
|
|
{"id": 100, "number": 10, "title": "something else"},
|
|
], "GET"),
|
|
"issues/9/comments": (201, {"id": 888}, "POST"),
|
|
}
|
|
fake_get, fake_post = _make_fake(routes)
|
|
with patch("ai_review.gitea_get", side_effect=fake_get), \
|
|
patch("ai_review.gitea_post", side_effect=fake_post):
|
|
stats = feedback_post.deliver(
|
|
api="http://x", token="t", db_path=self.db,
|
|
repo="gitea_admin/pragent", title="pragent feedback roll-up",
|
|
)
|
|
self.assertEqual(stats["issue_id"], 9)
|
|
self.assertEqual(stats["comment_id"], 888)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |