Marcos
231 lines
8.5 KiB
Python
231 lines
8.5 KiB
Python
"""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() |