gitea_admin/pragent
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(dashboard): read-only query module over feedback SQLite

Browse Source 
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.
This commit is contained in:
Claude
2026-08-22 15:07:57 +00:00
parent fe5bebb4cf
commit bb9b6aa12d
2 changed files with 551 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/pilot/test_dashboard_data.py
+249
View File
@@ -0,0 +1,249 @@
"""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()