"""Tests for pilot/feedback_analyze.py. Verify: - empty DB produces a friendly empty-state report (no crash) - findings are aggregated by posthash across multiple PRs - net false-positive score weights downvotes + unresolved + negation replies; acceptance weights upvotes + resolved - restraint metric reports the right ratio - case-review queue lists every disagreement - markdown + JSON output modes both work """ import json import os import sys import tempfile import unittest 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 def _seed(conn, findings): """Helper: insert a list of (repo, pr, path, line, severity, problem, [reaction users/contents], [reply bodies], [resolved]) tuples. Each finding gets a fresh review row + a unique comment_id so the reaction-join in `findings_with_votes` matches.""" for f in findings: (repo, pr_idx, path, line, sev, problem, reacts, replies, resolved) = f rid = feedback.record_review(conn, repo=repo, pr=pr_idx, head_sha="x") cid = (hash((repo, pr_idx, path, line, sev, problem)) & 0xFFFFFFFF) or 1 fid = feedback.record_inline_finding( conn, review_id=rid, repo=repo, pr=pr_idx, path=path, line=line, severity=sev, problem=problem, comment_id=cid, ) for user, content in reacts: feedback.record_reaction( conn, comment_id=cid, user=user, content=content, ) for i, body in enumerate(replies): feedback.record_reply( conn, finding_id=fid, author="alice", body=body, created_at=1000 + i, ) if resolved is not None: feedback.record_thread_state( conn, finding_id=fid, resolved=resolved, ) class TestEmptyState(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db = f"{self.tmp.name}/f.db" def tearDown(self): self.tmp.cleanup() def test_empty_db_markdown_does_not_crash(self): report = feedback_analyze.analyze(self.db) self.assertIn("# pragent feedback report", report) self.assertIn("findings analyzed**: 0", report) self.assertIn("Restraint", report) def test_empty_db_json_has_zero_findings(self): report = feedback_analyze.analyze(self.db, as_json=True) d = json.loads(report) self.assertEqual(d["total_findings"], 0) self.assertEqual(d["restraint"]["total"], 0) class TestScoring(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db = f"{self.tmp.name}/f.db" self.conn = feedback.init(self.db) # Two PRs, three findings: # A: 👍×2, resolved=true → acceptance # B: 👎×2, unresolved, "false positive" reply → false-positive # C: no signals → ignored _seed(self.conn, [ ("o/r", 1, "a.ts", 10, "HIGH", "race in handler", [("u1", "+1"), ("u2", "+1")], [], True), ("o/r", 1, "b.ts", 20, "LOW", "missing semicolon", [("u1", "-1"), ("u2", "-1")], ["False positive — this is fine."], False), ("o/r", 1, "c.ts", 30, "INFO", "naming nit", [], [], None), ]) def tearDown(self): self.conn.close() self.tmp.cleanup() def test_accepted_ranked_above_fp(self): d = json.loads(feedback_analyze.analyze(self.db, as_json=True)) self.assertEqual(len(d["top_accepted"]), 1) self.assertEqual(d["top_accepted"][0]["path"], "a.ts") self.assertEqual(len(d["top_false_positive"]), 1) self.assertEqual(d["top_false_positive"][0]["path"], "b.ts") def test_fp_score_combines_signals(self): d = json.loads(feedback_analyze.analyze(self.db, as_json=True)) fp = d["top_false_positive"][0] # 2 downvotes + 1 unresolved + 2 (negation phrase) = 5 self.assertEqual(fp["fp_score"], 5) def test_acceptance_score(self): d = json.loads(feedback_analyze.analyze(self.db, as_json=True)) ac = d["top_accepted"][0] # 2 upvotes + 1 resolved = 3 self.assertEqual(ac["ac_score"], 3) def test_case_queue_contains_only_disagreements(self): d = json.loads(feedback_analyze.analyze(self.db, as_json=True)) queue = d["case_review_queue"] self.assertEqual(len(queue), 1) self.assertEqual(queue[0]["path"], "b.ts") def test_no_signal_finding_is_ignored(self): d = json.loads(feedback_analyze.analyze(self.db, as_json=True)) # c.ts has no votes, no replies → not in either top list. paths = {e["path"] for e in d["top_accepted"]} paths.update(e["path"] for e in d["top_false_positive"]) self.assertNotIn("c.ts", paths) class TestRestraint(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db = f"{self.tmp.name}/f.db" self.conn = feedback.init(self.db) def tearDown(self): self.conn.close() self.tmp.cleanup() def test_high_ratio_triggers_recommendation(self): # 3 reviews, all with findings → 100% "noisy". for pr_i in range(3): feedback.record_review(self.conn, repo="o/r", pr=pr_i, head_sha="x") # Distinct (path, line) per PR so posthash doesn't dedup. feedback.record_inline_finding( self.conn, review_id=None, repo="o/r", pr=pr_i, path=f"a{pr_i}.ts", line=1, severity="LOW", problem=f"x {pr_i}", ) report = feedback_analyze.analyze(self.db) self.assertIn("⚠️", report) self.assertIn("100%", report) def test_low_ratio_passes(self): # 4 reviews, 1 with findings → 25% noisy = at threshold. for pr_i in range(4): feedback.record_review(self.conn, repo="o/r", pr=pr_i, head_sha="x") feedback.record_inline_finding( self.conn, review_id=None, repo="o/r", pr=0, path="a.ts", line=1, severity="LOW", problem="x", ) report = feedback_analyze.analyze(self.db) self.assertIn("✅", report) class TestMarkdownOutput(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db = f"{self.tmp.name}/f.db" self.conn = feedback.init(self.db) _seed(self.conn, [ ("o/r", 1, "a.ts", 10, "HIGH", "race in handler", [("u1", "+1")], [], True), ]) def tearDown(self): self.conn.close() self.tmp.cleanup() def test_report_has_sections(self): r = feedback_analyze.analyze(self.db) for section in ( "# pragent feedback report", "## Restraint", "## Top", "## Case-review queue", "## Where this report goes", ): self.assertIn(section, r) def test_doordash_rule_quoted(self): r = feedback_analyze.analyze(self.db) # The "noise on clean code" sentence from the DoorDash recap. self.assertIn("noise on clean code", r) class TestPosthashAggregation(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db = f"{self.tmp.name}/f.db" self.conn = feedback.init(self.db) # Same finding on three PRs → one aggregated row. # Each PR has its own review + finding (comment_id differs but # posthash is identical, so they collapse on aggregation). for pr_i in range(3): rid = feedback.record_review(self.conn, repo="o/r", pr=pr_i, head_sha="x") feedback.record_inline_finding( self.conn, review_id=rid, repo="o/r", pr=pr_i, path="a.ts", line=10, severity="HIGH", problem="identical problem text", comment_id=1000 + pr_i, ) feedback.record_reaction( self.conn, comment_id=1000 + pr_i, user="u", content="+1", ) def tearDown(self): self.conn.close() self.tmp.cleanup() def test_three_occurrences_one_row(self): d = json.loads(feedback_analyze.analyze(self.db, as_json=True)) self.assertEqual(len(d["top_accepted"]), 1) self.assertEqual(d["top_accepted"][0]["occurrences"], 3) self.assertEqual(d["top_accepted"][0]["ac_score"], 3) if __name__ == "__main__": unittest.main()