"""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()