Claude
bb9b6aa12d
Three pure functions — overview / repo_summary / pr_summary — that open the feedback SQLite via feedback.init, run their queries, and return plain dicts/lists. All three tolerate a missing or empty DB by returning a zero-shaped dict. Cost is hardcoded 0.0: per-review usage:cost isn't stored, only the raw review rows are. Surfacing a rolled-up dollar figure without telemetry would be guessing, so we don't. 17 new tests under tests/pilot/test_dashboard_data.py.
249 lines
9.7 KiB
Python
249 lines
9.7 KiB
Python
"""Tests for pilot/dashboard_data.py — read-only query layer over the feedback SQLite.
|
||
|
||
Covers: empty-DB fallbacks (no crash on missing/empty DB), overview rollups,
|
||
per-repo drill-down (findings by severity, top findings, runs by day), and
|
||
the per-PR view. The dashboard never writes — only reads.
|
||
"""
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import unittest
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
ROOT = os.path.abspath(os.path.join(HERE, "..", ".."))
|
||
sys.path.insert(0, os.path.join(HERE, "..", "..")) # so `from pilot import …` works
|
||
|
||
from pilot import dashboard_data, feedback
|
||
|
||
|
||
def _seed_repo(conn, *, repo: str, prs: int, findings_per_pr: int, day_offset: int = 0):
|
||
"""Seed one repo with `prs` PRs each with `findings_per_pr` findings.
|
||
|
||
All timestamps cluster on (now - day_offset days). Returns list of review ids.
|
||
"""
|
||
base = int(time.time()) - day_offset * 86400
|
||
rids = []
|
||
for n in range(prs):
|
||
rid = feedback.record_review(
|
||
conn, repo=repo, pr=n + 1, head_sha=f"sha{n}",
|
||
review_id_gitea=1000 + n, body_comment_id=2000 + n,
|
||
posted_at=base + n * 60,
|
||
)
|
||
rids.append(rid)
|
||
for k in range(findings_per_pr):
|
||
feedback.record_inline_finding(
|
||
conn, review_id=rid, repo=repo, pr=n + 1,
|
||
path=f"src/file_{k}.py", line=k + 1,
|
||
severity=["critical", "high", "medium", "low"][k % 4],
|
||
problem=f"problem {k}",
|
||
fix=f"fix {k}", suggestion=f"suggestion {k}",
|
||
comment_id=3000 + n * 10 + k,
|
||
posted_at=base + n * 60,
|
||
)
|
||
return rids
|
||
|
||
|
||
class TestEmptyDB(unittest.TestCase):
|
||
def test_missing_file_returns_zero_dict(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
missing = f"{d}/nope.db"
|
||
data = dashboard_data.overview(missing)
|
||
self.assertEqual(data["total_reviews"], 0)
|
||
self.assertEqual(data["total_findings"], 0)
|
||
self.assertEqual(data["total_repos"], 0)
|
||
self.assertEqual(data["last_30d_reviews"], 0)
|
||
self.assertEqual(len(data["daily"]), 7)
|
||
self.assertEqual(data["top_repos"], [])
|
||
self.assertEqual(data["total_cost_usd"], 0.0)
|
||
|
||
def test_missing_file_repo_summary_safe(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
data = dashboard_data.repo_summary(f"{d}/nope.db", "o/r")
|
||
self.assertEqual(data["repo"], "o/r")
|
||
self.assertEqual(data["total_runs"], 0)
|
||
self.assertEqual(data["runs_by_day"], [])
|
||
for sev in ("critical", "high", "medium", "low"):
|
||
self.assertEqual(data["findings_by_severity"][sev], 0)
|
||
self.assertEqual(data["top_findings"], [])
|
||
self.assertEqual(data["models_used"], [])
|
||
|
||
def test_missing_file_pr_summary_safe(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
data = dashboard_data.pr_summary(f"{d}/nope.db", "o/r", 1)
|
||
self.assertEqual(data["repo"], "o/r")
|
||
self.assertEqual(data["pr"], 1)
|
||
self.assertEqual(data["findings"], [])
|
||
self.assertEqual(data["usage"], {})
|
||
|
||
|
||
class TestEmptyButExistingDB(unittest.TestCase):
|
||
"""`init` creates the schema — DB exists but has no rows."""
|
||
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db = f"{self.tmp.name}/f.db"
|
||
feedback.init(self.db)
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def test_overview_is_zero(self):
|
||
data = dashboard_data.overview(self.db)
|
||
self.assertEqual(data["total_reviews"], 0)
|
||
self.assertEqual(data["total_findings"], 0)
|
||
self.assertEqual(data["total_repos"], 0)
|
||
|
||
def test_repo_summary_is_zero(self):
|
||
data = dashboard_data.repo_summary(self.db, "o/r")
|
||
self.assertEqual(data["total_runs"], 0)
|
||
self.assertEqual(data["findings_by_severity"], {"critical": 0, "high": 0, "medium": 0, "low": 0})
|
||
|
||
def test_pr_summary_is_zero(self):
|
||
data = dashboard_data.pr_summary(self.db, "o/r", 1)
|
||
self.assertEqual(data["findings"], [])
|
||
|
||
|
||
class TestOverview(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db = f"{self.tmp.name}/f.db"
|
||
self.conn = feedback.init(self.db)
|
||
_seed_repo(self.conn, repo="alpha/one", prs=3, findings_per_pr=2)
|
||
_seed_repo(self.conn, repo="beta/two", prs=1, findings_per_pr=4)
|
||
self.conn.close()
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def test_totals(self):
|
||
data = dashboard_data.overview(self.db)
|
||
self.assertEqual(data["total_reviews"], 4)
|
||
self.assertEqual(data["total_findings"], 6 + 4) # 3*2 + 1*4 = 10
|
||
self.assertEqual(data["total_repos"], 2)
|
||
self.assertEqual(data["total_cost_usd"], 0.0)
|
||
|
||
def test_top_repos_sorted_by_run_count(self):
|
||
data = dashboard_data.overview(self.db)
|
||
repos = [r["repo"] for r in data["top_repos"]]
|
||
# alpha/one has 3 runs, beta/two has 1.
|
||
self.assertEqual(repos[0], "alpha/one")
|
||
self.assertEqual(data["top_repos"][0]["run_count"], 3)
|
||
self.assertEqual(data["top_repos"][1]["run_count"], 1)
|
||
# last_seen is a unix timestamp int.
|
||
for r in data["top_repos"]:
|
||
self.assertIsInstance(r["last_seen"], int)
|
||
|
||
def test_daily_buckets_are_7(self):
|
||
data = dashboard_data.overview(self.db)
|
||
self.assertEqual(len(data["daily"]), 7)
|
||
for b in data["daily"]:
|
||
self.assertIn("date", b)
|
||
self.assertIn("count", b)
|
||
|
||
def test_last_30d_reviews(self):
|
||
data = dashboard_data.overview(self.db)
|
||
self.assertEqual(data["last_30d_reviews"], 4)
|
||
|
||
|
||
class TestRepoSummary(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db = f"{self.tmp.name}/f.db"
|
||
self.conn = feedback.init(self.db)
|
||
# 4 PRs with 2 findings each → 8 findings, severity cycle [c,h,m,l,c,h,m,l]
|
||
_seed_repo(self.conn, repo="o/r", prs=4, findings_per_pr=2)
|
||
# Add some reactions so top_findings has signal.
|
||
rows = self.conn.execute(
|
||
"SELECT id, comment_id FROM inline_finding WHERE repo=? ORDER BY id LIMIT 3",
|
||
("o/r",),
|
||
).fetchall()
|
||
for r in rows:
|
||
feedback.record_reaction(self.conn, comment_id=r["comment_id"], user="u", content="+1")
|
||
self.conn.close()
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def test_basic_shape(self):
|
||
data = dashboard_data.repo_summary(self.db, "o/r")
|
||
self.assertEqual(data["repo"], "o/r")
|
||
self.assertEqual(data["total_runs"], 4)
|
||
self.assertIsInstance(data["last_run_ts"], int)
|
||
|
||
def test_findings_by_severity(self):
|
||
data = dashboard_data.repo_summary(self.db, "o/r")
|
||
fbs = data["findings_by_severity"]
|
||
# 4 PRs × 2 findings; per-PR severities are [critical, high].
|
||
# (k in range(2) → k=0 critical, k=1 high for every PR.)
|
||
self.assertEqual(fbs["critical"], 4)
|
||
self.assertEqual(fbs["high"], 4)
|
||
self.assertEqual(fbs["medium"], 0)
|
||
self.assertEqual(fbs["low"], 0)
|
||
|
||
def test_runs_by_day_is_list(self):
|
||
data = dashboard_data.repo_summary(self.db, "o/r")
|
||
self.assertIsInstance(data["runs_by_day"], list)
|
||
for r in data["runs_by_day"]:
|
||
self.assertIn("date", r)
|
||
self.assertIn("count", r)
|
||
|
||
def test_top_findings_structure(self):
|
||
data = dashboard_data.repo_summary(self.db, "o/r")
|
||
self.assertGreater(len(data["top_findings"]), 0)
|
||
first = data["top_findings"][0]
|
||
for k in ("path", "line", "severity", "problem", "occurrences", "upvotes", "downvotes", "resolved", "reply_count"):
|
||
self.assertIn(k, first)
|
||
|
||
def test_models_used_is_empty_list_with_note(self):
|
||
# The schema has no `model` column on review — the dashboard can't show
|
||
# model usage from this DB today. We document that via an empty list.
|
||
data = dashboard_data.repo_summary(self.db, "o/r")
|
||
self.assertEqual(data["models_used"], [])
|
||
|
||
|
||
class TestPRSummary(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db = f"{self.tmp.name}/f.db"
|
||
self.conn = feedback.init(self.db)
|
||
rid = feedback.record_review(
|
||
self.conn, repo="o/r", pr=42, head_sha="abc",
|
||
review_id_gitea=9001, body_comment_id=8001,
|
||
posted_at=1700000000,
|
||
)
|
||
for k in range(3):
|
||
feedback.record_inline_finding(
|
||
self.conn, review_id=rid, repo="o/r", pr=42,
|
||
path=f"src/x_{k}.py", line=k + 10,
|
||
severity=["critical", "high", "low"][k],
|
||
problem=f"p{k}", fix=f"f{k}", suggestion=f"s{k}",
|
||
comment_id=7000 + k,
|
||
)
|
||
self.conn.close()
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def test_meta(self):
|
||
data = dashboard_data.pr_summary(self.db, "o/r", 42)
|
||
self.assertEqual(data["repo"], "o/r")
|
||
self.assertEqual(data["pr"], 42)
|
||
self.assertEqual(data["head_sha"], "abc")
|
||
self.assertEqual(data["review_id_gitea"], 9001)
|
||
self.assertEqual(data["body_comment_id"], 8001)
|
||
self.assertEqual(data["posted_at"], 1700000000)
|
||
# usage is empty because the schema has no usage column.
|
||
self.assertEqual(data["usage"], {})
|
||
|
||
def test_findings(self):
|
||
data = dashboard_data.pr_summary(self.db, "o/r", 42)
|
||
self.assertEqual(len(data["findings"]), 3)
|
||
for f in data["findings"]:
|
||
for k in ("path", "line", "severity", "problem", "fix", "suggestion",
|
||
"upvotes", "downvotes", "resolved", "reply_count"):
|
||
self.assertIn(k, f)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main() |