lhl/code-review-graph
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

chore: sync local changes, add Chinese docs and opencode config

Browse Source
This commit is contained in:
AuraK Developer
2026-08-31 11:37:03 +08:00
parent 307d2fd471
commit ecc55158c1
81 changed files with 7645 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_coverage.py
+421
View File
@@ -0,0 +1,421 @@
"""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"]
tests/test_line_coverage.py
+332
View File
@@ -0,0 +1,332 @@
"""Tests for the V2.1 line/unit coverage gate (gate="both+line").
Covers: per-file line coverage thresholds, unit-completeness (gap-free),
giant-file exemption, the four-state gate enum, and the v2 coverage index
round-trip (nodes.file_hash based, ranges persisted, v1 fallback).
"""
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.scoring import compute_coverage # noqa: E402
from code_review_graph.tools.scoring_tools import ( # noqa: E402
_coverage_index_path,
save_coverage_index_func,
)
class _Row:
def __init__(self, d):
self._d = d
def __getitem__(self, k):
return self._d[k]
class _Cursor:
def __init__(self, rows):
self._rows = rows
def fetchall(self):
return [_Row(r) for r in self._rows]
def fetchone(self):
return _Row(self._rows[0]) if self._rows else None
class _Conn:
def __init__(self, units):
self.units = units
def execute(self, sql, params=None):
if "kind IN ('Function','Class','Test')" in sql:
# returns dict-like rows with .keys() for dict(row) conversion
return _Cursor([dict(u) for u in self.units])
if "kind='File'" in sql:
return _Cursor([])
return _Cursor([])
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 _Store:
def __init__(self, files, units=None):
self._files = list(files)
self._conn = _Conn(units or [])
self._nodes = [
_Node("D:/repo/" + f, f) for f in self._files
]
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 []
@pytest.fixture
def repo(tmp_path: Path) -> Path:
Path(tmp_path, "src").mkdir(exist_ok=True)
Path(tmp_path, ".git").mkdir(exist_ok=True) # satisfy _get_store repo-root check
Path(tmp_path, "src", "a.py").write_text(
"\n".join(f"# line {i}" for i in range(1, 41)),
encoding="utf-8",
)
Path(tmp_path, "src", "b.py").write_text(
"def b():\n return 2\n",
encoding="utf-8",
)
return tmp_path
def _mk_store_units(rel, units):
"""Graph semantic units for a file: {name: [s,e]}."""
return [
{"kind": "Function", "name": name, "line_start": s, "line_end": e}
for name, (s, e) in units.items()
]
def test_line_coverage_four_tiers(repo):
files = ["src/a.py", "src/b.py"]
a_units = _mk_store_units("src/a.py", {"a": [1, 40]})
store = _Store(files, a_units)
a_rel = "src/a.py"
def cov(ranges):
return compute_coverage(
store, repo, deep_read_files=["src/a.py"],
include_churn=False, gate="both+line",
file_read_ranges={a_rel: ranges},
file_semantic_units={a_rel: [{"range": [1, 40], "kind": "Function", "name": "a"}]},
)
r = cov([[1, 20]]) # 50% of 40 lines
assert r["line_gap_files"][0]["coverage_pct"] == 50.0
assert r["target_reached"] is False
r = cov([[1, 40]]) # 100%
assert r["line_gap_files"] == []
assert r["line_coverage_pct"] == 100.0
def test_unit_gap_detected(repo):
files = ["src/a.py"]
units = _mk_store_units("src/a.py", {"a": [1, 20], "b": [21, 40]})
store = _Store(files, units)
a_rel = "src/a.py" # sub-agents report relative paths
# report only unit a, missing unit b
r = compute_coverage(
store, repo, deep_read_files=["src/a.py"], include_churn=False,
gate="both+line",
file_read_ranges={a_rel: [[1, 40]]},
file_semantic_units={a_rel: [{"range": [1, 20], "kind": "Function", "name": "a"}]},
)
assert len(r["unit_gap_files"]) == 1
assert r["unit_gap_files"][0]["covered_units"] == 1
assert r["unit_gap_files"][0]["total_units"] == 2
def test_unit_exempt_giant_file(repo):
files = ["src/a.py"]
units = _mk_store_units("src/a.py", {"big": [1, 40]})
store = _Store(files, units)
a_rel = "src/a.py" # relative path as sub-agents report
r = compute_coverage(
store, repo, deep_read_files=["src/a.py"], include_churn=False,
gate="both+line",
file_read_ranges={a_rel: [[1, 2]]},
file_semantic_units={a_rel: [{"range": [1, 2], "kind": "Function", "name": "big"}]},
)
assert len(r["unit_exempt_files"]) == 1 # single unit, span>80% -> exempt
assert r["unit_gap_files"] == []
def test_gate_both_line_requires_quality(repo):
files = ["src/a.py", "src/b.py"]
units = _mk_store_units("src/a.py", {"a": [1, 40]})
store = _Store(files, units)
a_rel = "src/a.py"
b_rel = "src/b.py"
# full file-count coverage but only 50% line coverage on a.py
r = compute_coverage(
store, repo,
deep_read_files=["src/a.py", "src/b.py"], include_churn=False,
gate="both+line",
file_read_ranges={a_rel: [[1, 20]], b_rel: [[1, 2]]},
file_semantic_units={
a_rel: [{"range": [1, 40], "kind": "Function", "name": "a"}],
b_rel: [],
},
)
assert r["coverage_pct"] == 100.0
assert r["line_coverage_pct"] < 100.0
assert r["target_reached"] is False # quality gate fails despite 100% file coverage
def test_gate_both_unchanged_back_compat(repo):
files = ["src/a.py", "src/b.py"]
store = _Store(files)
r = compute_coverage(
store, repo, deep_read_files=["src/a.py", "src/b.py"],
include_churn=False, gate="both",
)
assert r["gate"] == "both"
assert "line_gap_files" in r # present but empty (only filled for both+line)
assert r["line_gap_files"] == []
def test_fail_closed_missing_read_ranges(repo):
"""Deep-read file with NO read_ranges must fail (not silently pass).
Guards the v2.5.2 fail-closed fix: previously a deep-read file without
file_read_ranges was neither counted as covered nor recorded as a gap,
so gate="both+line" returned target_reached=true while line_coverage_pct
stayed 0 (silent green). Now it must be a line gap + missing_data.
"""
files = ["src/a.py"]
units = _mk_store_units("src/a.py", {"a": [1, 40]})
store = _Store(files, units)
r = compute_coverage(
store, repo, deep_read_files=["src/a.py"], include_churn=False,
gate="both+line",
# NO file_read_ranges / file_semantic_units passed at all
)
assert r["line_coverage_pct"] == 0.0
assert len(r["line_gap_files"]) == 1
assert r["line_gap_files"][0]["reason"] == "missing read_ranges"
assert any(m["field"] == "file_read_ranges" for m in r["missing_data_files"])
assert r["target_reached"] is False
def test_fail_closed_missing_semantic_units(repo):
"""Graph has units but file_semantic_units is absent -> unit gap."""
files = ["src/a.py"]
units = _mk_store_units("src/a.py", {"a": [1, 40]})
store = _Store(files, units)
r = compute_coverage(
store, repo, deep_read_files=["src/a.py"], include_churn=False,
gate="both+line",
file_read_ranges={"src/a.py": [[1, 40]]},
# file_semantic_units omitted
)
assert r["unit_coverage_pct"] == 0.0
assert len(r["unit_gap_files"]) == 1
assert r["unit_gap_files"][0]["reason"] == "missing semantic_units"
assert any(m["field"] == "file_semantic_units" for m in r["missing_data_files"])
assert r["target_reached"] is False
def test_fail_closed_full_data_passes(repo):
"""With complete three-piece data the file is verified (no false gap)."""
files = ["src/a.py"]
units = _mk_store_units("src/a.py", {"a": [1, 40]})
store = _Store(files, units)
r = compute_coverage(
store, repo, deep_read_files=["src/a.py"], include_churn=False,
gate="both+line",
file_read_ranges={"src/a.py": [[1, 40]]},
file_semantic_units={"src/a.py": [{"range": [1, 40], "kind": "Function", "name": "a"}]},
)
assert r["line_coverage_pct"] == 100.0
assert r["line_gap_files"] == []
assert r["unit_gap_files"] == []
assert r["missing_data_files"] == []
def test_gate_line_unit_only_checks_line_and_unit(repo):
"""gate=\"line+unit\" (feature) checks ONLY line + unit completeness.
File-count coverage (1 of 2 files = 50% < 85%) and high-risk coverage
are NOT part of the gate: coverage_pct/high_risk_coverage_pct are None
and target_reached reflects the line/unit quality gate alone.
"""
files = ["src/a.py", "src/b.py"]
units = _mk_store_units("src/a.py", {"a": [1, 40]})
store = _Store(files, units)
a_rel = "src/a.py"
r = compute_coverage(
store, repo,
deep_read_files=["src/a.py"], include_churn=False,
gate="line+unit",
file_read_ranges={a_rel: [[1, 40]]},
file_semantic_units={a_rel: [{"range": [1, 40], "kind": "Function", "name": "a"}]},
)
assert r["gate"] == "line+unit"
assert r["coverage_pct"] is None
assert r["high_risk_coverage_pct"] is None
assert r["line_coverage_pct"] == 100.0
assert r["unit_coverage_pct"] == 100.0
assert r["target_reached"] is True
assert r["grade"] == "good"
def test_gate_line_unit_fails_on_line_gap(repo):
"""A line gap below target still fails gate=\"line+unit\"."""
files = ["src/a.py"]
units = _mk_store_units("src/a.py", {"a": [1, 40]})
store = _Store(files, units)
a_rel = "src/a.py"
r = compute_coverage(
store, repo,
deep_read_files=["src/a.py"], include_churn=False,
gate="line+unit",
file_read_ranges={a_rel: [[1, 20]]}, # 50% of 40 lines
file_semantic_units={a_rel: [{"range": [1, 40], "kind": "Function", "name": "a"}]},
)
assert r["line_coverage_pct"] == 0.0 # 0 of 1 files pass the >=95% line gate
assert r["line_gap_files"][0]["coverage_pct"] == 50.0
assert r["target_reached"] is False
assert r["grade"] == "fail"
def test_gate_line_unit_fails_on_unit_gap(repo):
"""A missing semantic unit fails gate=\"line+unit\"."""
files = ["src/a.py"]
units = _mk_store_units("src/a.py", {"a": [1, 20], "b": [21, 40]})
store = _Store(files, units)
a_rel = "src/a.py"
r = compute_coverage(
store, repo,
deep_read_files=["src/a.py"], include_churn=False,
gate="line+unit",
file_read_ranges={a_rel: [[1, 40]]},
file_semantic_units={a_rel: [{"range": [1, 20], "kind": "Function", "name": "a"}]},
)
assert len(r["unit_gap_files"]) == 1
assert r["target_reached"] is False
def test_coverage_index_v2_roundtrip(repo, monkeypatch):
files = ["src/a.py"]
a_rel = "src/a.py"
r = save_coverage_index_func(
deep_read_files=[a_rel],
repo_root=str(repo),
file_read_ranges={a_rel: [[1, 2]]},
file_semantic_units={a_rel: [{"range": [1, 2], "kind": "Function", "name": "a"}]},
)
assert r["status"] == "ok"
index_path = _coverage_index_path(repo)
assert index_path.is_file()
payload = json.loads(index_path.read_text(encoding="utf-8"))
assert payload["version"] == 2
entry = payload["entries"][a_rel]
assert entry["ranges"] == [[1, 2]]
assert entry["units"]
tests/test_main.py
+26 -8
View File
@@ -207,9 +207,17 @@ class TestLongRunningToolsAreAsync:
)
def test_heavy_tool_source_uses_to_thread(self):
"""Defense in depth: the source of every heavy tool wrapper must
literally call asyncio.to_thread so we don't accidentally turn
a tool async without offloading the blocking work."""
"""Defense in depth: every heavy tool wrapper must offload its
blocking work via ``asyncio.to_thread`` — either directly or through
the shared ``_run_with_progress`` helper — so we don't accidentally
turn a tool async without offloading the work; that would hang the
stdio event loop on Windows. See #46, #136."""
helper = inspect.getsource(crg_main._run_with_progress)
assert "asyncio.to_thread" in helper, (
"_run_with_progress must call asyncio.to_thread to offload "
"blocking work; otherwise Windows MCP clients will hang. "
"See #46, #136."
)
for tool_name in self.HEAVY_TOOLS:
fn = getattr(crg_main, tool_name, None)
assert fn is not None, f"{tool_name} not found on module"
@@ -217,10 +225,12 @@ class TestLongRunningToolsAreAsync:
# through the wrapper to find the underlying source.
underlying = getattr(fn, "fn", None) or fn
source = inspect.getsource(underlying)
assert "asyncio.to_thread" in source, (
f"{tool_name} must call asyncio.to_thread to offload its "
f"blocking work; otherwise Windows MCP clients will hang. "
f"See #46, #136."
assert (
"asyncio.to_thread" in source or "_run_with_progress" in source
), (
f"{tool_name} must offload its blocking work via "
f"asyncio.to_thread or _run_with_progress; otherwise Windows "
f"MCP clients will hang. See #46, #136."
)
@pytest.mark.parametrize("tool_name,impl_name", HEAVY_TOOL_IMPLS.items())
@@ -361,11 +371,19 @@ class TestGraphBackedToolProvenanceCoverage:
self, category, tool_names,
):
assert tool_names, f"{category} must name at least one tool"
helper = inspect.getsource(crg_main._run_with_progress)
assert "with_provenance" in helper, (
"_run_with_progress must attach graph provenance for tools that "
"route through it"
)
for tool_name in tool_names:
tool = getattr(crg_main, tool_name, None)
assert tool is not None, f"{category}: missing {tool_name}"
underlying = getattr(tool, "fn", None) or tool
assert "with_provenance" in inspect.getsource(underlying), (
source = inspect.getsource(underlying)
assert (
"with_provenance" in source or "_run_with_progress" in source
), (
f"{category}: {tool_name} does not attach graph provenance"
)
tests/test_progress_bridge.py
+223
View File
@@ -0,0 +1,223 @@
"""Tests for the MCP progress bridge (_run_with_progress) and engine progress_cb.
Covers:
- Engine functions accept and invoke ``progress_cb`` (compute_coverage /
deep_read_plan / score_review / compute_file_churn).
- The event-loop heartbeat helper ``_run_with_progress`` emits MCP progress
notifications and relays real progress from the worker thread.
- ``CRG_TOOL_TIMEOUT`` server-side backstop returns a readable error dict.
"""
from __future__ import annotations
import asyncio
import sys
import time
from pathlib import Path
from typing import Any, Callable, Optional
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent))
from code_review_graph.scoring import ( # noqa: E402
compute_coverage,
deep_read_plan,
score_review,
)
class _FakeContext:
"""Minimal stand-in for fastmcp Context with a recording report_progress."""
def __init__(self) -> None:
self.calls: list[tuple[float, Optional[float], Optional[str]]] = []
async def report_progress(
self, progress: float, total: Optional[float] = None, message: Optional[str] = None
) -> None:
self.calls.append((progress, total, message))
class _FakeStore:
"""Minimal GraphStore stub covering what coverage/score use."""
def __init__(self, files: list[str]) -> None:
self._files = list(files)
def get_all_files(self) -> list[str]:
return list(self._files)
def get_nodes_by_file(self, file_path: str):
return []
def get_edges_by_target(self, qualified_name: str):
return []
def get_community_ids_by_qualified_names(self, qualified_names):
return {}
def get_edges_by_source(self, qualified_name: str):
return []
def get_communities(self, limit=None):
return []
def get_edges(self, kind=None, limit=None):
return []
@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 _record_progress(records: list[tuple[float, Optional[str]]]) -> Callable[[float, Optional[str]], None]:
def cb(fraction: float, message: Optional[str]) -> None:
records.append((fraction, message))
return cb
# ---------------------------------------------------------------------------
# Engine progress_cb
# ---------------------------------------------------------------------------
def test_compute_coverage_invokes_progress_cb(risk_repo):
files = ["src/a.py", "src/b.py", "src/c.py", "src/d.py"]
store = _FakeStore(files)
records: list[tuple[float, Optional[str]]] = []
result = compute_coverage(
store, risk_repo,
deep_read_files=["src/a.py"],
include_churn=False,
progress_cb=_record_progress(records),
)
assert result["status"] == "ok"
# Weights report every 50 files (>=1 call) plus a final "done".
assert len(records) >= 1
assert records[-1][0] == 1.0
assert "done" in (records[-1][1] or "").lower()
def test_deep_read_plan_invokes_progress_cb(risk_repo):
files = ["src/a.py", "src/b.py", "src/c.py", "src/d.py"]
store = _FakeStore(files)
records: list[tuple[float, Optional[str]]] = []
plan = deep_read_plan(
store, risk_repo,
deep_read_files=["src/a.py"],
target_coverage=85.0,
include_churn=False,
progress_cb=_record_progress(records),
)
assert plan["status"] == "ok"
assert len(records) >= 1
assert records[-1][0] == 1.0
def test_score_review_invokes_progress_cb(risk_repo):
store = _FakeStore([])
records: list[tuple[float, Optional[str]]] = []
result = score_review(
store, risk_repo,
changed_files=["src/a.py", "src/b.py", "src/c.py", "src/d.py"],
include_churn=False,
progress_cb=_record_progress(records),
)
assert result["status"] == "ok"
# One report per metric (5) + final done.
assert len(records) >= 5
assert records[-1][0] == 1.0
# ---------------------------------------------------------------------------
# _run_with_progress heartbeat helper
# ---------------------------------------------------------------------------
def test_run_with_progress_emits_heartbeat_and_real_progress(risk_repo):
from code_review_graph.main import _run_with_progress
ctx = _FakeContext()
def slow_coverage(deep_read_files, repo_root, progress_cb=None, **kw):
for i in range(3):
time.sleep(0.05)
if progress_cb:
progress_cb(i / 3.0, f"step {i}")
return compute_coverage(
_FakeStore(["src/a.py", "src/b.py"]),
repo_root,
deep_read_files=deep_read_files,
include_churn=False,
progress_cb=progress_cb,
)
result = asyncio.run(
_run_with_progress(
ctx, slow_coverage,
deep_read_files=["src/a.py"], repo_root=risk_repo,
heartbeat=0.02, tool_timeout=0,
)
)
assert result["status"] == "ok"
# Heartbeat + engine progress: at least one notification, and a real
# (non-"processing...") message from the worker surfaced through.
assert len(ctx.calls) >= 1
messages = [m for (_, _, m) in ctx.calls if m]
assert any("step" in m for m in messages), f"real progress not relayed: {messages}"
def test_run_with_progress_timeout_returns_error(risk_repo):
from code_review_graph.main import _run_with_progress
ctx = _FakeContext()
def forever(**kw):
time.sleep(5)
return {"status": "ok"}
result = asyncio.run(
_run_with_progress(
ctx, forever,
heartbeat=0.01, tool_timeout=1,
)
)
assert result["status"] == "error"
assert "timeout" in (result.get("error") or "").lower()
def test_run_with_progress_relays_when_no_progress_cb_param():
from code_review_graph.main import _run_with_progress
ctx = _FakeContext()
def plain(**kw):
time.sleep(0.2)
return {"status": "ok", "value": 42}
result = asyncio.run(
_run_with_progress(
ctx, plain,
heartbeat=0.02, tool_timeout=0,
)
)
assert result["status"] == "ok" and result["value"] == 42
# Pure heartbeat notifications (no real progress) still fire to keep the
# client timeout reset.
assert len(ctx.calls) >= 1
tests/test_prompts.py
+3 -4
View File
@@ -186,13 +186,12 @@ class TestUnifiedReviewPrompt:
result = unified_review_prompt()
text = _text(result[0])
assert "HEAD~1" in text
assert "standard tier" in text
assert "Standard tier" in text
def test_custom_base_and_tier(self):
result = unified_review_prompt(base="develop", tier="strict")
def test_custom_base(self):
result = unified_review_prompt(base="develop")
text = _text(result[0])
assert "base=develop" in text
assert "strict tier" in text
def test_mentions_score_review(self):
result = unified_review_prompt()
tests/test_report.py
+135
View File
@@ -81,6 +81,61 @@ class TestBuildReportData:
assert data["verdict"] == "\u274c FAIL"
assert data["issues"] == []
def test_reviewed_files_carried(self):
rd = _review_data()
rd["reviewed_files"] = ["src/a.py", "src/b.py"]
data = build_report_data(rd)
assert data["reviewed_files"] == ["src/a.py", "src/b.py"]
def test_reviewed_files_default_empty(self):
data = build_report_data(_review_data())
assert data["reviewed_files"] == []
def test_files_list_normalised_to_string(self):
rd = _review_data()
rd["files"] = ["src/a.py", "src/b.py"]
data = build_report_data(rd)
assert data["files"] == "src/a.py, src/b.py"
def test_reviewed_files_fallback_from_files_list(self):
rd = _review_data()
rd["files"] = ["src/a.py", "src/b.py"]
data = build_report_data(rd)
assert data["reviewed_files"] == ["src/a.py", "src/b.py"]
# explicit reviewed_files wins over the files-list fallback
rd["reviewed_files"] = ["src/a.py"]
assert build_report_data(rd)["reviewed_files"] == ["src/a.py"]
def test_reviewed_files_string_split(self):
"""A comma-separated STRING reviewed_files must be split into an array
(agents pass both formats). Guards against iterating the string
char-by-char (which rendered 's','e','r'... as fake files)."""
rd = _review_data()
rd["reviewed_files"] = "src/a.rs, src/b.rs, "
data = build_report_data(rd)
assert data["reviewed_files"] == ["src/a.rs", "src/b.rs"]
md = render_markdown_report(rd)
assert "src/a.rs" in md
assert "src/b.rs" in md
# no single-char file entries
assert "- `s`" not in md
assert "265" not in md
def test_metrics_filters_non_objective_keys(self):
rd = _review_data()
rd["metrics"]["blast_radius"] = {
"direct_changed_nodes": 101,
"impacted_nodes": 13,
}
rd["metrics"]["objective_grade"] = "fail"
data = build_report_data(rd)
assert "blast_radius" not in data["metrics"]
assert "objective_grade" not in data["metrics"]
assert set(data["metrics"]) == {
"sql_risk",
"exception_coverage",
}
class TestLoadTemplate:
def test_loads_package_asset(self):
@@ -123,6 +178,45 @@ class TestRenderMarkdownReport:
assert "# 代码审查报告" in md
assert "未发现问题" in md
def test_line_unit_coverage_skips_file_count_rows(self):
"""gate=\"line+unit\" (feature) must NOT render the 全库/高风险 rows."""
rd = {
"scope": "feature",
"tier": "standard",
"reviewed_files": ["src/a.py", "src/b.py"],
"coverage": {
"coverage_pct": None,
"high_risk_coverage_pct": None,
"grade": "good",
"target_reached": True,
"gate": "line+unit",
"line_coverage_pct": 100.0,
"unit_coverage_pct": 100.0,
"line_gap_files": [],
"unit_gap_files": [],
},
}
md = render_markdown_report(rd)
assert "## 覆盖度" in md
assert "**状态**" in md
assert "100.0%" in md
assert "覆盖度（全库）" not in md
assert "覆盖度（高风险）" not in md
def test_reviewed_files_collapsible(self):
rd = {
"scope": "feature",
"tier": "standard",
"reviewed_files": ["src/a.py", "src/b.py"],
}
md = render_markdown_report(rd)
assert "审查文件（2 个）" in md
assert "<details>" in md
assert "点击展开 / 收起" in md
assert "src/a.py" in md
# flat "文件" meta line is skipped when reviewed_files is present
assert "- **文件**：" not in md
class TestGenerateReport:
def test_both_writes_html_and_md(self, tmp_path):
@@ -147,6 +241,47 @@ class TestGenerateReport:
assert "N+1" in html_text
assert "# 代码审查报告" in md_text
def test_script_closing_sequence_is_escaped(self, tmp_path):
"""Literal "</script>" in finding text must not break the HTML.
A literal ``</script>`` inside the injected JSON data would close the
surrounding <script> tag early, corrupting the report. The JSON payload
must escape every "<" as "\\u003c" so the browser parses it as a plain
string while the JS template re-renders it as "<".
"""
root = _mkroot(tmp_path)
rd = _review_data()
rd["findings"] = [
{
"severity": "major",
"category": "security",
"message": 'literal </script> opener',
"path": "src/app.py",
"line": 42,
"confidence": 8,
"fix": "escape </script> before embedding",
},
]
result = generate_report_func(
rd,
output_path="script-safe",
repo_root=str(root),
)
assert result["status"] == "ok"
html_text = (root / "script-safe.html").read_text(encoding="utf-8")
# The injected JSON payload must not contain a bare closing tag.
data_block = html_text.split("const data = ", 1)[1].split(";", 1)[0]
assert "</script" not in data_block
# It must keep the JSON escape so the value round-trips as "<".
assert "\\u003c/script" in data_block
# The real template closing tag must still be present exactly once.
assert html_text.count("</script>") == 1
# Finding text is intact after JSON decoding.
import json as _json
decoded = _json.loads(data_block)
assert "literal </script> opener" in decoded["issues"][0]["message"]
def test_markdown_only(self, tmp_path):
root = _mkroot(tmp_path)
result = generate_report_func(
tests/test_scoring.py
+13
View File
@@ -139,6 +139,19 @@ class TestVulnerabilityHeuristic:
assert result["value"] >= 1
assert result["grade"] in ("warn", "fail")
def test_note_is_chinese(self, tmp_path):
"""The metric note must follow the report language (Chinese), not
English. Technical proper nouns (OWASP, npm audit) may remain."""
Path(tmp_path, "s.py").write_text(
"def f():\n return 1\n",
encoding="utf-8",
)
result = compute_vulnerability_heuristic(["s.py"], tmp_path)
assert result["note"]
assert "启发式" in result["note"]
# the body is Chinese, not the old English sentence
assert "Heuristic OWASP/secret-pattern scan" not in result["note"]
# ---------------------------------------------------------------------------
# dedupe_findings