Marcos
243 lines
9.5 KiB
Python
243 lines
9.5 KiB
Python
"""Tests for pilot/feedback_harvest.py.
|
|
|
|
Mock `gitea_get` so we exercise the harvester's flow against canned Gitea
|
|
responses. Verify:
|
|
- bot-authored reviews only are processed
|
|
- reactions + thread state + replies all get recorded
|
|
- best-effort failures don't raise (one bad endpoint shouldn't kill the
|
|
whole harvest)
|
|
- posthash dedup: harvesting the same PR twice does NOT double-count
|
|
reactions.
|
|
"""
|
|
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 ai_review # noqa: E402
|
|
import feedback # noqa: E402
|
|
import feedback_harvest # noqa: E402
|
|
|
|
|
|
def _make_fake_gitea(routes: dict):
|
|
"""Build a stand-in for `ai_review.gitea_get` that returns canned bodies.
|
|
|
|
`routes` maps relative path (substring) → (status, json_body). Sort
|
|
keys longest-first so e.g. `pulls/5/reviews/100/comments` matches
|
|
before `pulls/5/reviews` (which is also a substring of the longer
|
|
path).
|
|
"""
|
|
def fake(api, repo, path, token, accept="application/json"):
|
|
for needle in sorted(routes.keys(), key=len, reverse=True):
|
|
if needle in path:
|
|
status, body = routes[needle]
|
|
return status, json.dumps(body).encode()
|
|
return 404, b'{"message":"not found"}'
|
|
return fake
|
|
|
|
|
|
def _patch(fake):
|
|
"""Apply the fake to ai_review.gitea_get and feedback_harvest's import."""
|
|
return patch("ai_review.gitea_get", side_effect=fake)
|
|
|
|
|
|
class TestHarvestForPr(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db = f"{self.tmp.name}/f.db"
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def _review_payload(self, body="", commit_id="abc123", review_id=100):
|
|
return [{
|
|
"id": review_id, "user": {"login": "pragent-bot"},
|
|
"commit_id": commit_id, "body": body,
|
|
"created_at": "2026-08-20T10:00:00Z",
|
|
}]
|
|
|
|
def _inline_payload(self, comment_id=500, body="**[LOW]** x", path="a/b.ts", position=42, resolver=""):
|
|
return [{
|
|
"id": comment_id, "path": path, "position": position,
|
|
"body": body, "resolver": resolver,
|
|
}]
|
|
|
|
def test_happy_path_records_reaction_and_thread(self):
|
|
fake = _make_fake_gitea({
|
|
"pulls/5/reviews": (200, self._review_payload()),
|
|
"pulls/5/reviews/100/comments": (200, self._inline_payload(resolver="masi")),
|
|
"issues/comments/500/reactions": (200, [
|
|
{"user": {"login": "alice"}, "content": "+1",
|
|
"created_at": "2026-08-20T11:00:00Z"},
|
|
{"user": {"login": "bob"}, "content": "-1",
|
|
"created_at": "2026-08-20T11:01:00Z"},
|
|
]),
|
|
"issues/5/comments": (200, []), # no replies
|
|
})
|
|
with _patch(fake):
|
|
stats = feedback_harvest.harvest_for_pr(
|
|
api="http://x", token="t", repo="o/r", pr_index=5,
|
|
db_path=self.db,
|
|
)
|
|
self.assertEqual(stats["reviews_seen"], 1)
|
|
self.assertEqual(stats["findings_seen"], 1)
|
|
self.assertEqual(stats["reactions_recorded"], 2)
|
|
self.assertEqual(stats["thread_states_recorded"], 1)
|
|
# DB should have 1 review, 1 finding, 2 reactions, 1 thread_state
|
|
conn = feedback.init(self.db)
|
|
self.assertEqual(
|
|
conn.execute("SELECT COUNT(*) FROM review").fetchone()[0], 1,
|
|
)
|
|
self.assertEqual(
|
|
conn.execute("SELECT COUNT(*) FROM inline_finding").fetchone()[0], 1,
|
|
)
|
|
self.assertEqual(
|
|
conn.execute("SELECT COUNT(*) FROM reaction").fetchone()[0], 2,
|
|
)
|
|
self.assertEqual(
|
|
conn.execute("SELECT resolved FROM thread_state").fetchone()[0], 1,
|
|
)
|
|
conn.close()
|
|
|
|
def test_skips_non_bot_reviews(self):
|
|
fake = _make_fake_gitea({
|
|
"pulls/5/reviews": (200, [{
|
|
"id": 999, "user": {"login": "masi"}, # not the bot
|
|
"commit_id": "x", "body": "", "created_at": "2026-08-20T10:00:00Z",
|
|
}]),
|
|
})
|
|
with _patch(fake):
|
|
stats = feedback_harvest.harvest_for_pr(
|
|
api="http://x", token="t", repo="o/r", pr_index=5,
|
|
db_path=self.db,
|
|
)
|
|
self.assertEqual(stats["reviews_seen"], 0)
|
|
conn = feedback.init(self.db)
|
|
self.assertEqual(
|
|
conn.execute("SELECT COUNT(*) FROM review").fetchone()[0], 0,
|
|
)
|
|
conn.close()
|
|
|
|
def test_review_list_failure_does_not_raise(self):
|
|
fake = _make_fake_gitea({
|
|
"pulls/5/reviews": (500, None),
|
|
})
|
|
with _patch(fake):
|
|
stats = feedback_harvest.harvest_for_pr(
|
|
api="http://x", token="t", repo="o/r", pr_index=5,
|
|
db_path=self.db,
|
|
)
|
|
self.assertEqual(stats["reviews_seen"], 0)
|
|
self.assertGreaterEqual(stats["errors"], 1)
|
|
|
|
def test_reactions_endpoint_returns_null_is_tolerated(self):
|
|
# Some Gitea endpoints return JSON `null` for empty lists. We must
|
|
# not crash — treat it as "no reactions".
|
|
fake = _make_fake_gitea({
|
|
"pulls/5/reviews": (200, self._review_payload()),
|
|
"pulls/5/reviews/100/comments": (200, self._inline_payload()),
|
|
"issues/comments/500/reactions": (200, None),
|
|
"issues/5/comments": (200, []),
|
|
})
|
|
with _patch(fake):
|
|
stats = feedback_harvest.harvest_for_pr(
|
|
api="http://x", token="t", repo="o/r", pr_index=5,
|
|
db_path=self.db,
|
|
)
|
|
self.assertEqual(stats["reactions_recorded"], 0)
|
|
|
|
def test_reactions_dedup_via_pk_across_harvests(self):
|
|
# Two harvests of the same PR — both produce an inline_finding row,
|
|
# but reactions are PK-deduped on (comment_id, user, content) so
|
|
# the SECOND harvest does NOT double-record them.
|
|
fake = _make_fake_gitea({
|
|
"pulls/5/reviews": (200, self._review_payload()),
|
|
"pulls/5/reviews/100/comments": (200, self._inline_payload()),
|
|
"issues/comments/500/reactions": (200, [
|
|
{"user": {"login": "alice"}, "content": "+1",
|
|
"created_at": "2026-08-20T11:00:00Z"},
|
|
]),
|
|
"issues/5/comments": (200, []),
|
|
})
|
|
with _patch(fake):
|
|
feedback_harvest.harvest_for_pr(
|
|
api="http://x", token="t", repo="o/r", pr_index=5,
|
|
db_path=self.db,
|
|
)
|
|
feedback_harvest.harvest_for_pr(
|
|
api="http://x", token="t", repo="o/r", pr_index=5,
|
|
db_path=self.db,
|
|
)
|
|
conn = feedback.init(self.db)
|
|
# Two findings (no DB-level posthash UNIQUE), one reaction.
|
|
self.assertEqual(
|
|
conn.execute("SELECT COUNT(*) FROM inline_finding").fetchone()[0], 2,
|
|
)
|
|
self.assertEqual(
|
|
conn.execute("SELECT COUNT(*) FROM reaction").fetchone()[0], 1,
|
|
)
|
|
conn.close()
|
|
|
|
def test_replies_with_review_comment_id_recorded(self):
|
|
fake = _make_fake_gitea({
|
|
"pulls/5/reviews": (200, self._review_payload()),
|
|
"pulls/5/reviews/100/comments": (200, self._inline_payload(comment_id=500)),
|
|
"issues/comments/500/reactions": (200, []),
|
|
"issues/5/comments": (200, [{
|
|
"id": 900, "review_comment_id": 500,
|
|
"user": {"login": "alice"},
|
|
"body": "False positive — this is intentional",
|
|
"created_at": "2026-08-20T12:00:00Z",
|
|
}]),
|
|
})
|
|
with _patch(fake):
|
|
stats = feedback_harvest.harvest_for_pr(
|
|
api="http://x", token="t", repo="o/r", pr_index=5,
|
|
db_path=self.db,
|
|
)
|
|
self.assertEqual(stats["replies_recorded"], 1)
|
|
conn = feedback.init(self.db)
|
|
self.assertEqual(
|
|
conn.execute("SELECT COUNT(*) FROM reply").fetchone()[0], 1,
|
|
)
|
|
conn.close()
|
|
|
|
|
|
class TestParseHelpers(unittest.TestCase):
|
|
def test_severity_extracted(self):
|
|
self.assertEqual(
|
|
feedback_harvest._parse_severity("**[HIGH]** race in foo"),
|
|
"HIGH",
|
|
)
|
|
|
|
def test_severity_defaults_to_info(self):
|
|
self.assertEqual(feedback_harvest._parse_severity("plain text"), "INFO")
|
|
|
|
def test_path_line_extracted(self):
|
|
p, l = feedback_harvest._parse_path_line("see `src/foo.ts:42` here")
|
|
self.assertEqual(p, "src/foo.ts")
|
|
self.assertEqual(l, 42)
|
|
|
|
def test_negation_phrases_caught(self):
|
|
self.assertTrue(feedback_harvest._is_negation_reply("This is intentional."))
|
|
self.assertTrue(feedback_harvest._is_negation_reply("false positive — see X"))
|
|
self.assertFalse(feedback_harvest._is_negation_reply("thanks for catching this!"))
|
|
# Empty / None safe
|
|
self.assertFalse(feedback_harvest._is_negation_reply(""))
|
|
self.assertFalse(feedback_harvest._is_negation_reply(None))
|
|
|
|
def test_classify_reaction(self):
|
|
self.assertEqual(feedback_harvest.classify_reaction("+1"), "positive")
|
|
self.assertEqual(feedback_harvest.classify_reaction("-1"), "negative")
|
|
self.assertEqual(feedback_harvest.classify_reaction("rocket"), "positive")
|
|
self.assertEqual(feedback_harvest.classify_reaction("confused"), "negative")
|
|
self.assertEqual(feedback_harvest.classify_reaction("eyes"), "neutral")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |