pragent/tests/pilot/test_feedback.py

"""Tests for pilot/feedback.py — SQLite storage for review feedback signals.

Covers: schema bootstrap, posthash stability, dedup-on-insert, reaction /
thread-state / reply upserts, the analyzer-side `findings_with_votes` join,
and graceful failure on bad inputs.
"""
import sqlite3
import tempfile
import unittest

from pilot import feedback


class TestPosthash(unittest.TestCase):
    def test_stable_across_calls(self):
        a = feedback.posthash("src/api/foo.ts", 42, "HIGH", "Race condition in handler")
        b = feedback.posthash("src/api/foo.ts", 42, "HIGH", "Race condition in handler")
        self.assertEqual(a, b)

    def test_length_is_short(self):
        h = feedback.posthash("a", 1, "low", "x")
        self.assertEqual(len(h), 16)

    def test_different_path_different_hash(self):
        self.assertNotEqual(
            feedback.posthash("a", 1, "LOW", "x"),
            feedback.posthash("b", 1, "LOW", "x"),
        )

    def test_different_line_different_hash(self):
        self.assertNotEqual(
            feedback.posthash("a", 1, "LOW", "x"),
            feedback.posthash("a", 2, "LOW", "x"),
        )

    def test_different_severity_different_hash(self):
        # Same line, same problem, different severity → different signal.
        self.assertNotEqual(
            feedback.posthash("a", 1, "LOW", "x"),
            feedback.posthash("a", 1, "CRITICAL", "x"),
        )

    def test_problem_prefix_used_only(self):
        # First 80 chars participate; rest is ignored.
        self.assertEqual(
            feedback.posthash("a", 1, "LOW", "x" * 80 + "tail"),
            feedback.posthash("a", 1, "LOW", "x" * 80),
        )

    def test_case_and_whitespace_normalized_in_problem(self):
        # Lowercased + stripped → same hash.
        self.assertEqual(
            feedback.posthash("a", 1, "LOW", "  Same Finding  "),
            feedback.posthash("a", 1, "LOW", "same finding"),
        )


class TestInit(unittest.TestCase):
    def test_init_creates_db(self):
        with tempfile.TemporaryDirectory() as d:
            db = f"{d}/f.db"
            conn = feedback.init(db)
            # Application tables exist (sqlite_sequence is a bookkeeping table
            # created by AUTOINCREMENT — not part of the contract).
            tables = {r[0] for r in conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table'"
            ).fetchall()}
            self.assertTrue(
                {"review", "inline_finding", "reaction", "thread_state", "reply"}.issubset(tables),
                f"missing tables: got {tables}",
            )
            conn.close()

    def test_init_is_idempotent(self):
        with tempfile.TemporaryDirectory() as d:
            db = f"{d}/f.db"
            feedback.init(db)
            # Second call must not raise.
            feedback.init(db)


class TestRecordReview(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.conn = feedback.init(f"{self.tmp.name}/f.db")

    def tearDown(self):
        self.conn.close(); self.tmp.cleanup()

    def test_returns_id_and_row(self):
        rid = feedback.record_review(
            self.conn, repo="o/r", pr=1, head_sha="abc",
            review_id_gitea=99, body_comment_id=42,
        )
        self.assertIsNotNone(rid)
        row = self.conn.execute("SELECT * FROM review WHERE id = ?", (rid,)).fetchone()
        self.assertEqual(row[1], "o/r")
        self.assertEqual(row[4], 99)
        self.assertEqual(row[5], 42)


class TestRecordInlineFinding(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.conn = feedback.init(f"{self.tmp.name}/f.db")

    def tearDown(self):
        self.conn.close(); self.tmp.cleanup()

    def _new_review(self):
        return feedback.record_review(
            self.conn, repo="o/r", pr=1, head_sha="x",
        )

    def test_insert_returns_id(self):
        rid = self._new_review()
        fid = feedback.record_inline_finding(
            self.conn, review_id=rid, repo="o/r", pr=1,
            path="a/b.ts", line=10, severity="HIGH",
            problem="bug", fix="patch", suggestion="code",
            comment_id=555,
        )
        self.assertIsNotNone(fid)

    def test_dedup_on_posthash(self):
        # Two reviews of the SAME finding on different PRs insert two
        # rows — deduplication by posthash is the *analyzer's* job
        # (findings_with_votes GROUP BY posthash). Storing one row per
        # review preserves per-comment reactions across PRs.
        rid1 = self._new_review()
        feedback.record_inline_finding(
            self.conn, review_id=rid1, repo="o/r", pr=1,
            path="a/b.ts", line=10, severity="HIGH", problem="race",
            comment_id=100,
        )
        rid2 = feedback.record_review(self.conn, repo="o/r", pr=2, head_sha="y")
        feedback.record_inline_finding(
            self.conn, review_id=rid2, repo="o/r", pr=2,
            path="a/b.ts", line=10, severity="HIGH", problem="race",
            comment_id=200,
        )
        rows = self.conn.execute(
            "SELECT id, comment_id FROM inline_finding WHERE path='a/b.ts' AND line=10 ORDER BY id"
        ).fetchall()
        self.assertEqual(len(rows), 2)
        # Both comment_ids preserved (PK dedup is the *reaction* table's job).
        self.assertEqual([r[1] for r in rows], [100, 200])

    def test_posthash_set(self):
        rid = self._new_review()
        feedback.record_inline_finding(
            self.conn, review_id=rid, repo="o/r", pr=1,
            path="a", line=1, severity="LOW", problem="nit",
        )
        ph = self.conn.execute(
            "SELECT posthash FROM inline_finding LIMIT 1"
        ).fetchone()[0]
        expected = feedback.posthash("a", 1, "LOW", "nit")
        self.assertEqual(ph, expected)


class TestRecordReaction(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.conn = feedback.init(f"{self.tmp.name}/f.db")

    def tearDown(self):
        self.conn.close(); self.tmp.cleanup()

    def test_insert_upsert(self):
        ok = feedback.record_reaction(
            self.conn, comment_id=10, user="alice", content="+1",
        )
        self.assertTrue(ok)
        n = self.conn.execute("SELECT COUNT(*) FROM reaction").fetchone()[0]
        self.assertEqual(n, 1)
        # Re-insert same PK → no duplicate.
        feedback.record_reaction(self.conn, comment_id=10, user="alice", content="+1")
        n = self.conn.execute("SELECT COUNT(*) FROM reaction").fetchone()[0]
        self.assertEqual(n, 1)

    def test_distinct_users_can_react(self):
        feedback.record_reaction(self.conn, comment_id=10, user="a", content="+1")
        feedback.record_reaction(self.conn, comment_id=10, user="b", content="-1")
        n = self.conn.execute("SELECT COUNT(*) FROM reaction").fetchone()[0]
        self.assertEqual(n, 2)


class TestRecordThreadState(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.conn = feedback.init(f"{self.tmp.name}/f.db")
        rid = feedback.record_review(self.conn, repo="o/r", pr=1, head_sha="x")
        self.fid = feedback.record_inline_finding(
            self.conn, review_id=rid, repo="o/r", pr=1,
            path="a", line=1, severity="LOW", problem="x",
        )

    def tearDown(self):
        self.conn.close(); self.tmp.cleanup()

    def test_upsert_overwrites(self):
        feedback.record_thread_state(self.conn, finding_id=self.fid, resolved=True)
        row = self.conn.execute(
            "SELECT resolved FROM thread_state WHERE finding_id = ?", (self.fid,)
        ).fetchone()
        self.assertEqual(row[0], 1)
        feedback.record_thread_state(self.conn, finding_id=self.fid, resolved=False)
        row = self.conn.execute(
            "SELECT resolved FROM thread_state WHERE finding_id = ?", (self.fid,)
        ).fetchone()
        self.assertEqual(row[0], 0)


class TestRecordReply(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.conn = feedback.init(f"{self.tmp.name}/f.db")
        rid = feedback.record_review(self.conn, repo="o/r", pr=1, head_sha="x")
        self.fid = feedback.record_inline_finding(
            self.conn, review_id=rid, repo="o/r", pr=1,
            path="a", line=1, severity="LOW", problem="x",
        )

    def tearDown(self):
        self.conn.close(); self.tmp.cleanup()

    def test_insert_idempotent(self):
        feedback.record_reply(
            self.conn, finding_id=self.fid, author="a",
            body="hi", created_at=1000,
        )
        feedback.record_reply(
            self.conn, finding_id=self.fid, author="a",
            body="hi", created_at=1000,  # same PK
        )
        n = self.conn.execute("SELECT COUNT(*) FROM reply").fetchone()[0]
        self.assertEqual(n, 1)


class TestFindingsWithVotes(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.conn = feedback.init(f"{self.tmp.name}/f.db")

    def tearDown(self):
        self.conn.close(); self.tmp.cleanup()

    def _seed(self):
        rid = feedback.record_review(self.conn, repo="o/r", pr=1, head_sha="x")
        fid = feedback.record_inline_finding(
            self.conn, review_id=rid, repo="o/r", pr=1,
            path="a/b.ts", line=10, severity="HIGH",
            problem="race", comment_id=500,
        )
        feedback.record_reaction(self.conn, comment_id=500, user="u1", content="+1")
        feedback.record_reaction(self.conn, comment_id=500, user="u2", content="-1")
        feedback.record_thread_state(self.conn, finding_id=fid, resolved=True)
        feedback.record_reply(
            self.conn, finding_id=fid, author="u3",
            body="this is fine because of X", created_at=2000,
        )
        return fid

    def test_join_rolls_up_votes(self):
        self._seed()
        rows = list(feedback.findings_with_votes(self.conn))
        self.assertEqual(len(rows), 1)
        r = rows[0]
        self.assertEqual(r["upvotes"], 1)
        self.assertEqual(r["downvotes"], 1)
        self.assertEqual(r["resolved"], 1)
        self.assertEqual(r["reply_count"], 1)
        self.assertIn("this is fine", r["reply_bodies"])

    def test_repo_filter(self):
        self._seed()
        # Add a finding under a different repo.
        feedback.record_inline_finding(
            self.conn, review_id=None, repo="other/r", pr=99,
            path="x", line=1, severity="LOW", problem="y",
        )
        rows = list(feedback.findings_with_votes(self.conn, repo="o/r"))
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0]["repo"], "o/r")

    def test_findings_with_no_signals_return_zero_votes(self):
        feedback.record_inline_finding(
            self.conn, review_id=None, repo="x/y", pr=1,
            path="p", line=1, severity="LOW", problem="z",
        )
        rows = list(feedback.findings_with_votes(self.conn))
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0]["upvotes"], 0)
        self.assertEqual(rows[0]["downvotes"], 0)
        self.assertIsNone(rows[0]["resolved"])


class TestKnownPosthashes(unittest.TestCase):
    def test_returns_distinct_set(self):
        with tempfile.TemporaryDirectory() as d:
            conn = feedback.init(f"{d}/f.db")
            feedback.record_inline_finding(
                conn, review_id=None, repo="o/r", pr=1,
                path="a", line=1, severity="LOW", problem="x",
            )
            feedback.record_inline_finding(
                conn, review_id=None, repo="o/r", pr=1,
                path="a", line=2, severity="LOW", problem="y",
            )
            phs = feedback.known_posthashes_for_repo(conn, "o/r")
            self.assertEqual(len(phs), 2)
            conn.close()


if __name__ == "__main__":
    unittest.main()