422 lines
14 KiB
Python
422 lines
14 KiB
Python
"""Tests for coverage computation (compute_coverage / check_community_health)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from code_review_graph.parser import normalize_file_path
|
|
from code_review_graph.scoring import ( # noqa: E402
|
|
_is_source_file,
|
|
build_report_data,
|
|
check_community_health,
|
|
compute_coverage,
|
|
deep_read_plan,
|
|
render_markdown_report,
|
|
)
|
|
|
|
|
|
class _Node:
|
|
def __init__(self, qualified_name, file_path, is_test=False, kind="Function"):
|
|
self.qualified_name = qualified_name
|
|
self.file_path = file_path
|
|
self.is_test = is_test
|
|
self.kind = kind
|
|
|
|
|
|
class _Edge:
|
|
def __init__(self, kind, source_qualified, target_qualified):
|
|
self.kind = kind
|
|
self.source_qualified = source_qualified
|
|
self.target_qualified = target_qualified
|
|
|
|
|
|
class _Store:
|
|
"""Minimal fake GraphStore for coverage unit tests."""
|
|
|
|
def __init__(self, files, nodes, edges):
|
|
self._files = list(files)
|
|
self._nodes = nodes
|
|
self._edges = edges
|
|
|
|
def get_all_files(self):
|
|
return list(self._files)
|
|
|
|
def get_nodes_by_file(self, file_path):
|
|
return [n for n in self._nodes if n.file_path == file_path]
|
|
|
|
def get_edges_by_target(self, qualified_name):
|
|
return [e for e in self._edges if e.target_qualified == qualified_name]
|
|
|
|
|
|
@pytest.fixture
|
|
def tiny_repo(tmp_path: Path) -> Path:
|
|
"""A tiny repo with two source files + one non-source file."""
|
|
Path(tmp_path, "src").mkdir(exist_ok=True)
|
|
Path(tmp_path, "src", "a.py").write_text(
|
|
"def a():\n"
|
|
" try:\n"
|
|
" return 1\n"
|
|
" except Exception as e:\n"
|
|
" raise ValueError(str(e))\n",
|
|
encoding="utf-8",
|
|
)
|
|
Path(tmp_path, "src", "b.py").write_text(
|
|
"def b():\n return 2\n",
|
|
encoding="utf-8",
|
|
)
|
|
Path(tmp_path, "docs").mkdir(exist_ok=True)
|
|
Path(tmp_path, "docs", "README.md").write_text(
|
|
"# docs\n",
|
|
encoding="utf-8",
|
|
)
|
|
return tmp_path
|
|
|
|
|
|
def test_is_source_file_excludes_non_source():
|
|
assert _is_source_file("src/app.py") is True
|
|
assert _is_source_file("docs/README.md") is False
|
|
assert _is_source_file("tests/test_x.py") is False
|
|
assert _is_source_file("scripts/build.py") is False
|
|
assert _is_source_file("src/app.py.bak") is False
|
|
|
|
|
|
def test_compute_coverage_returns_expected_shape(tiny_repo):
|
|
files = [
|
|
"src/a.py",
|
|
"src/b.py",
|
|
]
|
|
nodes = [
|
|
_Node("D:/repo/src/a.py::a", "src/a.py"),
|
|
_Node("D:/repo/src/b.py::b", "src/b.py"),
|
|
]
|
|
store = _Store(files, nodes, [])
|
|
result = compute_coverage(
|
|
store,
|
|
tiny_repo,
|
|
deep_read_files=["src/a.py"],
|
|
include_churn=False,
|
|
)
|
|
assert result["status"] == "ok"
|
|
assert result["total_files"] == 2
|
|
assert result["deep_read_count"] == 1
|
|
assert 0.0 <= result["coverage_pct"] <= 100.0
|
|
assert 0.0 <= result["high_risk_coverage_pct"] <= 100.0
|
|
assert result["target"] == 95.0
|
|
assert result["overall_target"] == 85.0
|
|
assert result["high_risk_target"] == 95.0
|
|
assert result["target_reached"] is False # 1/2 = 50% < 95%
|
|
assert result["uncovered_files"] == ["src/b.py"]
|
|
# b.py is clean (no sql/vuln/redundancy signals, no topology, no churn)
|
|
# => w1 = good(1) < 2.0 and no signals => it IS a silent file.
|
|
assert result["silent_files"] == ["src/b.py"]
|
|
|
|
|
|
def test_compute_coverage_full_deep_read_reaches_target(tiny_repo):
|
|
files = ["src/a.py", "src/b.py"]
|
|
store = _Store(files, [], [])
|
|
result = compute_coverage(
|
|
store,
|
|
tiny_repo,
|
|
deep_read_files=["src/a.py", "src/b.py"],
|
|
include_churn=False,
|
|
)
|
|
assert result["coverage_pct"] == 100.0
|
|
# tiny_repo files carry no sql/vuln/redundancy signal => no high-risk
|
|
# files => gate falls back to overall coverage (100%) => reached.
|
|
assert result["target_reached"] is True
|
|
assert result["grade"] == "good"
|
|
|
|
|
|
def test_compute_coverage_excludes_docs_from_denominator(tiny_repo):
|
|
files = ["src/a.py", "src/b.py", "docs/README.md"]
|
|
store = _Store(files, [], [])
|
|
result = compute_coverage(
|
|
store,
|
|
tiny_repo,
|
|
deep_read_files=["src/a.py"],
|
|
include_churn=False,
|
|
)
|
|
assert result["total_files"] == 2 # docs excluded
|
|
assert "docs/README.md" not in result["uncovered_files"]
|
|
|
|
|
|
def test_check_community_health_fake_store():
|
|
class _Conn:
|
|
def execute(self, _sql):
|
|
return _Cursor()
|
|
|
|
class _Cursor:
|
|
def fetchone(self):
|
|
return (4,)
|
|
|
|
store = type("S", (), {"_conn": _Conn()})()
|
|
# Our fake always returns 4 for every count; just ensure it runs.
|
|
result = check_community_health(store)
|
|
assert "status" in result
|
|
|
|
|
|
def test_report_renders_coverage_section():
|
|
rd = {
|
|
"verdict": "PASS",
|
|
"scope": "whole-project",
|
|
"tier": "standard",
|
|
"timestamp": "2026-08-11T00:00:00Z",
|
|
"files": "src/a.py",
|
|
"metrics": {"sql_risk": {"value": 0, "grade": "good", "note": "ok"}},
|
|
"findings": [],
|
|
"coverage": {
|
|
"coverage_pct": 7.1,
|
|
"high_risk_coverage_pct": 62.0,
|
|
"grade": "good",
|
|
"deep_read_count": 24,
|
|
"total_files": 508,
|
|
"high_risk_total_files": 205,
|
|
"high_risk_deep_count": 14,
|
|
"target_reached": True,
|
|
"target": 95.0,
|
|
"uncovered_files": ["src/c.py"],
|
|
"silent_files": ["src/d.py"],
|
|
},
|
|
}
|
|
md = render_markdown_report(rd)
|
|
assert "## 覆盖度" in md
|
|
assert "7.1%" in md
|
|
assert "62.0%" in md
|
|
assert "✅ 达标" in md
|
|
data = build_report_data(rd)
|
|
assert data["coverage"]["coverage_pct"] == 7.1
|
|
assert data["coverage"]["high_risk_coverage_pct"] == 62.0
|
|
|
|
|
|
def test_report_marks_insufficient_coverage():
|
|
rd = {
|
|
"verdict": "FAIL",
|
|
"scope": "whole-project",
|
|
"tier": "standard",
|
|
"timestamp": "2026-08-11T00:00:00Z",
|
|
"coverage": {
|
|
"coverage_pct": 3.0,
|
|
"high_risk_coverage_pct": 30.0,
|
|
"grade": "fail",
|
|
"deep_read_count": 5,
|
|
"total_files": 508,
|
|
"high_risk_total_files": 205,
|
|
"high_risk_deep_count": 3,
|
|
"target_reached": False,
|
|
"target": 95.0,
|
|
"uncovered_files": [],
|
|
"silent_files": [],
|
|
},
|
|
}
|
|
md = render_markdown_report(rd)
|
|
assert "🔴 覆盖不足" in md
|
|
assert "30.0%" in md
|
|
|
|
|
|
@pytest.fixture
|
|
def risk_repo(tmp_path: Path) -> Path:
|
|
"""a.py carries an SQL-risk signal (fail w1), the rest are clean."""
|
|
Path(tmp_path, "src").mkdir(exist_ok=True)
|
|
Path(tmp_path, "src", "a.py").write_text(
|
|
"def a():\n"
|
|
" sql = 'SELECT * FROM users WHERE id=' + str(uid)\n"
|
|
" return sql\n",
|
|
encoding="utf-8",
|
|
)
|
|
for name in ("b", "c", "d"):
|
|
Path(tmp_path, "src", f"{name}.py").write_text(
|
|
f"def {name}():\n return 2\n",
|
|
encoding="utf-8",
|
|
)
|
|
return tmp_path
|
|
|
|
|
|
def test_coverage_gate_overall_vs_high_risk_vs_both(risk_repo):
|
|
files = ["src/a.py", "src/b.py", "src/c.py", "src/d.py"]
|
|
store = _Store(files, [], [])
|
|
|
|
# Deep-read only the risky file: overall file-count 25%, high-risk 100%.
|
|
base = dict(store=store, repo_root=risk_repo,
|
|
deep_read_files=["src/a.py"], include_churn=False)
|
|
|
|
overall = compute_coverage(gate="overall", **base)
|
|
assert overall["target_reached"] is False # 25% < 85%
|
|
|
|
high_risk = compute_coverage(gate="high_risk", **base)
|
|
assert high_risk["target_reached"] is True # 100% >= 95%
|
|
|
|
both = compute_coverage(gate="both", **base)
|
|
assert both["target_reached"] is False # overall fails
|
|
assert both["gate"] == "both"
|
|
|
|
|
|
def test_coverage_returns_gap_fields(risk_repo):
|
|
files = ["src/a.py", "src/b.py", "src/c.py", "src/d.py"]
|
|
store = _Store(files, [], [])
|
|
result = compute_coverage(
|
|
store, risk_repo,
|
|
deep_read_files=["src/a.py"],
|
|
include_churn=False,
|
|
gate="both",
|
|
)
|
|
# File-count gap: 4 files, 1 deep-read, overall target 85% => 4*0.85-1 = 2.4.
|
|
assert result["total_files"] == 4
|
|
assert result["remaining_files_to_target"] == 2.4
|
|
# Compatible risk-weight gap: a.py w1=2 (warn, one SQL hit), others
|
|
# w1=1 => total=5, overall target 85% => 4.25 - 2.0 = 2.25.
|
|
assert result["total_weight"] == 5.0
|
|
assert result["remaining_weight_to_target"] == 2.25
|
|
prio = result["priority_deep_read_files"]
|
|
assert [p["path"] for p in prio] == [
|
|
normalize_file_path(str(Path(risk_repo, "src", f"{n}.py")))
|
|
for n in ("b", "c", "d")
|
|
]
|
|
assert all(p["weight"] == 1.0 for p in prio)
|
|
|
|
|
|
def test_deep_read_plan_greedy_and_groups(risk_repo):
|
|
files = ["src/a.py", "src/b.py", "src/c.py", "src/d.py"]
|
|
store = _Store(files, [], [])
|
|
plan = deep_read_plan(
|
|
store, risk_repo,
|
|
deep_read_files=["src/a.py"],
|
|
target_coverage=85.0,
|
|
batch_size=40,
|
|
include_churn=False,
|
|
)
|
|
# File-count: total=4, current=1, target=3.4, remaining=2.4 -> pick
|
|
# the 3 uncovered files b, c, d (weights equal, any order is fine but
|
|
# the greedy loop fills up to the remaining count).
|
|
assert plan["current_files"] == 1
|
|
assert plan["target_files"] == 3.4
|
|
assert plan["remaining_files"] == 2.4
|
|
assert len(plan["planned_files"]) == 3
|
|
planned = [normalize_file_path(str(Path(risk_repo, "src", f"{n}.py")))
|
|
for n in ("b", "c", "d")]
|
|
assert plan["planned_files"] == planned
|
|
assert plan["estimated_batches"] == 1
|
|
assert plan["groups"][0]["name"] == "src"
|
|
assert len(plan["groups"][0]["files"]) == 3
|
|
|
|
|
|
def test_deep_read_plan_excludes_prior_covered(risk_repo):
|
|
files = ["src/a.py", "src/b.py", "src/c.py", "src/d.py"]
|
|
store = _Store(files, [], [])
|
|
prior = {normalize_file_path(str(Path(risk_repo, "src", "b.py")))}
|
|
plan = deep_read_plan(
|
|
store, risk_repo,
|
|
deep_read_files=["src/a.py"],
|
|
target_coverage=85.0,
|
|
include_churn=False,
|
|
prior_covered=prior,
|
|
)
|
|
# b already covered by a prior round => current=2 files, remaining =
|
|
# 4*0.85 - 2 = 1.4 -> pick c and d (2 files).
|
|
assert plan["current_files"] == 2
|
|
assert plan["remaining_files"] == 1.4
|
|
assert len(plan["planned_files"]) == 2
|
|
planned = [normalize_file_path(str(Path(risk_repo, "src", f"{n}.py")))
|
|
for n in ("c", "d")]
|
|
assert plan["planned_files"] == planned
|
|
|
|
|
|
def test_coverage_dual_target_overall_ok_high_risk_not(risk_repo):
|
|
"""Overall 85% reached but high-risk 95% not reached => gate=both fails."""
|
|
repo = risk_repo.parent / "dual_target_repo"
|
|
repo.mkdir(exist_ok=True)
|
|
src = repo / "src"
|
|
src.mkdir(exist_ok=True)
|
|
(src / "risky.py").write_text(
|
|
"def r():\n sql = 'SELECT * FROM users WHERE id=' + str(uid)\n return sql\n",
|
|
encoding="utf-8",
|
|
)
|
|
clean = [f"f{i}.py" for i in range(9)]
|
|
for name in clean:
|
|
(src / name).write_text(f"def {name}():\n return 1\n", encoding="utf-8")
|
|
files = ["src/risky.py"] + [f"src/{n}" for n in clean]
|
|
store = _Store(files, [], [])
|
|
deep = [f"src/{n}" for n in clean] # 9 clean files, skip risky
|
|
res = compute_coverage(
|
|
store, repo,
|
|
deep_read_files=deep,
|
|
include_churn=False,
|
|
gate="both",
|
|
)
|
|
assert res["total_files"] == 10
|
|
assert res["deep_read_count"] == 9
|
|
assert res["coverage_pct"] == 90.0 # >= 85% overall target
|
|
assert res["high_risk_coverage_pct"] == 0.0 # < 95% high-risk target
|
|
assert res["overall_target"] == 85.0
|
|
assert res["high_risk_target"] == 95.0
|
|
assert res["target_reached"] is False # both gate: high-risk fails
|
|
assert res["grade"] == "fail"
|
|
res_overall = compute_coverage(
|
|
store, repo,
|
|
deep_read_files=deep,
|
|
include_churn=False,
|
|
gate="overall",
|
|
)
|
|
assert res_overall["target_reached"] is True
|
|
|
|
|
|
def _write_coverage_index(repo: Path, entries: dict[str, str]) -> Path:
|
|
"""Write a coverage-index.json mapping rel path -> SHA."""
|
|
payload = {"version": 1, "last_updated": "2026-08-14T00:00:00", "entries": {
|
|
rel: {"sha": sha} for rel, sha in entries.items()
|
|
}}
|
|
index = repo / ".code-review-graph" / "coverage-index.json"
|
|
index.parent.mkdir(parents=True, exist_ok=True)
|
|
index.write_text(json.dumps(payload), encoding="utf-8")
|
|
return index
|
|
|
|
|
|
def _blob_sha(repo: Path, rel: str) -> str:
|
|
import subprocess
|
|
|
|
out = subprocess.run(
|
|
["git", "hash-object", rel],
|
|
capture_output=True, text=True, stdin=subprocess.DEVNULL,
|
|
cwd=str(repo), timeout=15,
|
|
)
|
|
assert out.returncode == 0, out.stderr
|
|
return out.stdout.strip()
|
|
|
|
|
|
def test_load_coverage_index_batch(tmp_path: Path) -> None:
|
|
"""Batch git hash-object path returns only SHA-stable files."""
|
|
repo = tmp_path
|
|
(repo / ".git").mkdir(parents=True)
|
|
# A git index is required for hash-object of tracked/untracked files to
|
|
# work in the same way; an empty repo is enough (hash-object works on any
|
|
# existing file path in the working tree).
|
|
(repo / "a.txt").write_text("alpha", encoding="utf-8")
|
|
(repo / "b.txt").write_text("beta", encoding="utf-8")
|
|
|
|
sha_a = _blob_sha(repo, "a.txt")
|
|
_write_coverage_index(repo, {"a.txt": sha_a, "b.txt": "0000000000000000000000000000000000000000"})
|
|
|
|
from code_review_graph.tools.scoring_tools import _load_coverage_index
|
|
|
|
covered = _load_coverage_index(repo)
|
|
assert covered == ["a.txt"] # only a.txt's SHA still matches
|
|
|
|
|
|
def test_load_coverage_index_skips_missing_file(tmp_path: Path) -> None:
|
|
"""A deleted indexed file must not truncate the batch or be returned."""
|
|
repo = tmp_path
|
|
(repo / ".git").mkdir(parents=True)
|
|
(repo / "a.txt").write_text("alpha", encoding="utf-8")
|
|
sha_a = _blob_sha(repo, "a.txt")
|
|
_write_coverage_index(repo, {"a.txt": sha_a, "gone.txt": sha_a})
|
|
|
|
from code_review_graph.tools.scoring_tools import _load_coverage_index
|
|
|
|
covered = _load_coverage_index(repo)
|
|
assert covered == ["a.txt"]
|