diff --git a/codegraph/analysis/__init__.py b/codegraph/analysis/__init__.py index debaef0..350cf75 100644 --- a/codegraph/analysis/__init__.py +++ b/codegraph/analysis/__init__.py @@ -6,7 +6,7 @@ from codegraph.analysis.hotspots import Hotspot, find_hotspots from codegraph.analysis.metrics import GraphMetrics, compute_metrics from codegraph.analysis.roles import classify_roles -from codegraph.analysis.untested import UntestedNode, find_untested +from codegraph.analysis.untested import UntestedNode, find_untested, rank_untested __all__ = [ "BlastRadiusResult", @@ -26,5 +26,6 @@ "find_hotspots", "find_untested", "match_route", + "rank_untested", "trace", ] diff --git a/codegraph/analysis/untested.py b/codegraph/analysis/untested.py index f74e04c..3dc56da 100644 --- a/codegraph/analysis/untested.py +++ b/codegraph/analysis/untested.py @@ -1,7 +1,7 @@ """Untested-symbol detection.""" from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field import networkx as nx @@ -27,6 +27,29 @@ class UntestedNode: file: str line_start: int incoming_calls: int + hotspot_score: int = field(default=0) + blast_radius_size: int = field(default=0) + blast_files: int = field(default=0) + + +def _compute_hotspot_score(graph: nx.MultiDiGraph, nid: str) -> int: + """Compute hotspot score for a node: fan_in*2 + fan_out + loc//50. + + Uses the same formula as :func:`codegraph.analysis.hotspots.Hotspot.score`. + """ + attrs = graph.nodes.get(nid) or {} + fan_in = 0 + fan_out = 0 + for _src, _dst, key in graph.in_edges(nid, keys=True): + if key == EdgeKind.CALLS.value: + fan_in += 1 + for _src, _dst, key in graph.out_edges(nid, keys=True): + if key == EdgeKind.CALLS.value: + fan_out += 1 + line_start = int(attrs.get("line_start") or 0) + line_end = int(attrs.get("line_end") or 0) + loc = max(0, line_end - line_start + 1) if line_end else 0 + return fan_in * 2 + fan_out + loc // 50 def find_untested(graph: nx.MultiDiGraph) -> list[UntestedNode]: @@ -77,3 +100,42 @@ def find_untested(graph: nx.MultiDiGraph) -> list[UntestedNode]: ) out.sort(key=lambda u: (-u.incoming_calls, u.file, u.line_start)) return out + + +def rank_untested( + graph: nx.MultiDiGraph, + *, + blast_depth: int = 3, + top: int | None = None, +) -> list[UntestedNode]: + """Return untested nodes ranked by risk: hotspot score x blast radius. + + Enriches each :class:`UntestedNode` with ``hotspot_score``, + ``blast_radius_size``, and ``blast_files``, then sorts by + ``(-hotspot_score, -blast_radius_size, file, line_start)``. + + Blast-radius computation is skipped for nodes with no callers (perf). + + Args: + graph: The project call graph. + blast_depth: Maximum hop depth for blast-radius traversal. + top: If given, truncate result to this many entries. + + Returns: + Sorted list of :class:`UntestedNode` with risk fields populated. + """ + from codegraph.analysis.blast_radius import blast_radius + + nodes = find_untested(graph) + for node in nodes: + node.hotspot_score = _compute_hotspot_score(graph, node.id) + if node.incoming_calls > 0: + br = blast_radius(graph, node.id, depth=blast_depth) + node.blast_radius_size = br.size + node.blast_files = len(br.files) + nodes.sort( + key=lambda u: (-u.hotspot_score, -u.blast_radius_size, u.file, u.line_start) + ) + if top is not None: + nodes = nodes[:top] + return nodes diff --git a/codegraph/cli.py b/codegraph/cli.py index 1a792ed..cfe8985 100644 --- a/codegraph/cli.py +++ b/codegraph/cli.py @@ -1359,21 +1359,32 @@ def query_subgraph( @query_app.command("untested") def query_untested(limit: int = typer.Option(50, "--limit")) -> None: - """List functions/methods with no test-side caller.""" - from codegraph.analysis import find_untested + """List functions/methods with no test-side caller, ranked by risk.""" + from codegraph.analysis import rank_untested graph = _open_graph(Path.cwd()) if graph is None: raise typer.Exit(1) - rows = find_untested(graph) + rows = rank_untested(graph) console.print(f"[bold]{len(rows)} untested[/bold]") table = Table() table.add_column("qualname") table.add_column("file") table.add_column("line", justify="right") table.add_column("callers", justify="right") + table.add_column("hotspot", justify="right") + table.add_column("blast_r", justify="right") + table.add_column("blast_f", justify="right") for u in rows[:limit]: - table.add_row(u.qualname, u.file, str(u.line_start), str(u.incoming_calls)) + table.add_row( + u.qualname, + u.file, + str(u.line_start), + str(u.incoming_calls), + str(u.hotspot_score), + str(u.blast_radius_size), + str(u.blast_files), + ) console.print(table) diff --git a/codegraph/mcp_server/server.py b/codegraph/mcp_server/server.py index 378f420..b690fd1 100644 --- a/codegraph/mcp_server/server.py +++ b/codegraph/mcp_server/server.py @@ -331,10 +331,10 @@ def tool_untested( graph: nx.MultiDiGraph, limit: int = 50, ) -> list[dict[str, Any]]: - """Return untested functions/methods.""" - from codegraph.analysis.untested import find_untested + """Return untested functions/methods ranked by risk.""" + from codegraph.analysis.untested import rank_untested - items = find_untested(graph) + items = rank_untested(graph) return [ { "qualname": u.qualname, @@ -342,6 +342,9 @@ def tool_untested( "file": u.file, "line": u.line_start, "incoming_calls": u.incoming_calls, + "hotspot_score": u.hotspot_score, + "blast_radius_size": u.blast_radius_size, + "blast_files": u.blast_files, } for u in items[:limit] ] diff --git a/codegraph/review/rules.py b/codegraph/review/rules.py index 1efce21..c34e2bb 100644 --- a/codegraph/review/rules.py +++ b/codegraph/review/rules.py @@ -10,6 +10,8 @@ import networkx as nx import yaml +from codegraph.analysis._common import in_test_module +from codegraph.graph.schema import EdgeKind, NodeKind from codegraph.review.differ import GraphDiff, NodeChange from codegraph.review.risk import ( _count_callers, @@ -87,6 +89,13 @@ class Finding: severity="med", message="Modified node signature change", ), + Rule( + id="new-untested-hotspot", + when="new_untested_hotspot", + severity="high", + message="New function with {fan_in} callers and no test coverage", + threshold=5, + ), ] @@ -98,6 +107,7 @@ class Finding: "introduces_cycle", "high_fan_in", "new_dead_code", + "new_untested_hotspot", } @@ -316,6 +326,39 @@ def evaluate_rules( new_graph=new_graph, old_graph=old_graph, extra=extra, ) ) + elif when == "new_untested_hotspot": + threshold = rule.threshold or 5 + for change in diff.added_nodes: + if not _node_matches(rule, change): + continue + if change.kind not in ( + NodeKind.FUNCTION.value, + NodeKind.METHOD.value, + ): + continue + new_id = _find_node_id(change.qualname, change.kind, new_graph) + if new_id is None: + continue + fan_in = _count_callers(new_id, new_graph) + if fan_in < threshold: + continue + # Check whether any caller is from a test module. + has_test_caller = False + for src, _dst, key in new_graph.in_edges(new_id, keys=True): + if key != EdgeKind.CALLS.value: + continue + if in_test_module(new_graph, src): + has_test_caller = True + break + if has_test_caller: + continue + findings.append( + _make_finding( + rule, change, + new_graph=new_graph, old_graph=old_graph, extra=extra, + fmt_kwargs={"fan_in": fan_in}, + ) + ) elif when == "introduces_cycle": if not cycle_introduced: continue diff --git a/tests/fixtures/python_hotspot_v1/core.py b/tests/fixtures/python_hotspot_v1/core.py new file mode 100644 index 0000000..d4c7d1e --- /dev/null +++ b/tests/fixtures/python_hotspot_v1/core.py @@ -0,0 +1,12 @@ +"""Minimal v1 module — no hot_helper yet.""" +from __future__ import annotations + + +def helper_a(x: int) -> int: + """Simple helper.""" + return x + 1 + + +def helper_b(x: int) -> int: + """Another helper.""" + return x + 2 diff --git a/tests/fixtures/python_hotspot_v1/test_core.py b/tests/fixtures/python_hotspot_v1/test_core.py new file mode 100644 index 0000000..16820f9 --- /dev/null +++ b/tests/fixtures/python_hotspot_v1/test_core.py @@ -0,0 +1,12 @@ +"""Tests for core (v1).""" +from __future__ import annotations + +from core import helper_a, helper_b + + +def test_helper_a() -> None: + assert helper_a(1) == 2 + + +def test_helper_b() -> None: + assert helper_b(1) == 3 diff --git a/tests/fixtures/python_hotspot_v2/core.py b/tests/fixtures/python_hotspot_v2/core.py new file mode 100644 index 0000000..21d0443 --- /dev/null +++ b/tests/fixtures/python_hotspot_v2/core.py @@ -0,0 +1,32 @@ +"""Minimal v2 module — hot_helper added with many callers but no test.""" +from __future__ import annotations + + +def hot_helper(x: int) -> int: + """New critical helper used everywhere.""" + return x * 10 + + +def helper_a(x: int) -> int: + """Simple helper.""" + return hot_helper(x) + 1 + + +def helper_b(x: int) -> int: + """Another helper.""" + return hot_helper(x) + 2 + + +def helper_c(x: int) -> int: + """Third helper.""" + return hot_helper(x) + 3 + + +def helper_d(x: int) -> int: + """Fourth helper.""" + return hot_helper(x) + 4 + + +def helper_e(x: int) -> int: + """Fifth helper.""" + return hot_helper(x) + 5 diff --git a/tests/fixtures/python_hotspot_v2/test_core.py b/tests/fixtures/python_hotspot_v2/test_core.py new file mode 100644 index 0000000..1b189af --- /dev/null +++ b/tests/fixtures/python_hotspot_v2/test_core.py @@ -0,0 +1,12 @@ +"""Tests for core (v2) — intentionally does NOT test hot_helper directly.""" +from __future__ import annotations + +from core import helper_a, helper_b + + +def test_helper_a() -> None: + assert helper_a(1) == 11 + + +def test_helper_b() -> None: + assert helper_b(1) == 12 diff --git a/tests/fixtures/python_hotspot_v2_tested/core.py b/tests/fixtures/python_hotspot_v2_tested/core.py new file mode 100644 index 0000000..21d0443 --- /dev/null +++ b/tests/fixtures/python_hotspot_v2_tested/core.py @@ -0,0 +1,32 @@ +"""Minimal v2 module — hot_helper added with many callers but no test.""" +from __future__ import annotations + + +def hot_helper(x: int) -> int: + """New critical helper used everywhere.""" + return x * 10 + + +def helper_a(x: int) -> int: + """Simple helper.""" + return hot_helper(x) + 1 + + +def helper_b(x: int) -> int: + """Another helper.""" + return hot_helper(x) + 2 + + +def helper_c(x: int) -> int: + """Third helper.""" + return hot_helper(x) + 3 + + +def helper_d(x: int) -> int: + """Fourth helper.""" + return hot_helper(x) + 4 + + +def helper_e(x: int) -> int: + """Fifth helper.""" + return hot_helper(x) + 5 diff --git a/tests/fixtures/python_hotspot_v2_tested/test_core.py b/tests/fixtures/python_hotspot_v2_tested/test_core.py new file mode 100644 index 0000000..8e570f9 --- /dev/null +++ b/tests/fixtures/python_hotspot_v2_tested/test_core.py @@ -0,0 +1,16 @@ +"""Tests for core (v2) — this variant DOES test hot_helper directly.""" +from __future__ import annotations + +from core import helper_a, helper_b, hot_helper + + +def test_hot_helper() -> None: + assert hot_helper(2) == 20 + + +def test_helper_a() -> None: + assert helper_a(1) == 11 + + +def test_helper_b() -> None: + assert helper_b(1) == 12 diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index b2849f2..0e4ac0d 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -222,6 +222,9 @@ def test_untested_returns_list(fixture_graph: nx.MultiDiGraph) -> None: for r in results: assert "qualname" in r assert "kind" in r + assert "hotspot_score" in r + assert "blast_radius_size" in r + assert "blast_files" in r def test_hotspots_returns_list(fixture_graph: nx.MultiDiGraph) -> None: diff --git a/tests/test_rank_untested.py b/tests/test_rank_untested.py new file mode 100644 index 0000000..b3fedcb --- /dev/null +++ b/tests/test_rank_untested.py @@ -0,0 +1,138 @@ +"""Tests for rank_untested and the enriched UntestedNode fields.""" +from __future__ import annotations + +import networkx as nx + +from codegraph.analysis.untested import find_untested, rank_untested +from codegraph.graph.schema import EdgeKind, NodeKind + + +def _make_graph() -> nx.MultiDiGraph: + """Build a minimal graph with two untested functions: + + - ``hot_func``: 3 incoming CALLS (high fan-in, high hotspot) + - ``cold_func``: 0 incoming CALLS (zero fan-in, low hotspot) + + Neither has any caller from a test module, so both are untested. + """ + g: nx.MultiDiGraph = nx.MultiDiGraph() + + # The two untested functions + g.add_node( + "hot_func", + kind=NodeKind.FUNCTION.value, + name="hot_func", + qualname="pkg.hot_func", + file="pkg/core.py", + line_start=10, + line_end=30, + metadata={}, + ) + g.add_node( + "cold_func", + kind=NodeKind.FUNCTION.value, + name="cold_func", + qualname="pkg.cold_func", + file="pkg/core.py", + line_start=35, + line_end=40, + metadata={}, + ) + + # Three non-test callers of hot_func + for i in range(3): + caller_id = f"caller_{i}" + g.add_node( + caller_id, + kind=NodeKind.FUNCTION.value, + name=f"caller_{i}", + qualname=f"pkg.caller_{i}", + file="pkg/callers.py", + line_start=i * 10 + 1, + line_end=i * 10 + 5, + metadata={}, + ) + g.add_edge(caller_id, "hot_func", key=EdgeKind.CALLS.value, kind=EdgeKind.CALLS.value) + + return g + + +def _make_graph_with_test_caller() -> nx.MultiDiGraph: + """Same as _make_graph but adds a test-module caller for hot_func.""" + g = _make_graph() + g.add_node( + "test_hot_func", + kind=NodeKind.FUNCTION.value, + name="test_hot_func", + qualname="tests.test_core.test_hot_func", + file="tests/test_core.py", + line_start=1, + line_end=5, + metadata={"is_test": True}, + ) + g.add_edge( + "test_hot_func", "hot_func", + key=EdgeKind.CALLS.value, kind=EdgeKind.CALLS.value, + ) + return g + + +class TestRankUntested: + def test_hot_func_ranks_above_cold_func(self) -> None: + g = _make_graph() + ranked = rank_untested(g) + qualnames = [u.qualname for u in ranked] + assert "pkg.hot_func" in qualnames + assert "pkg.cold_func" in qualnames + hot_idx = qualnames.index("pkg.hot_func") + cold_idx = qualnames.index("pkg.cold_func") + assert hot_idx < cold_idx, "hot_func must rank above cold_func" + + def test_hotspot_score_populated(self) -> None: + g = _make_graph() + ranked = rank_untested(g) + hot = next(u for u in ranked if u.qualname == "pkg.hot_func") + # fan_in=3, fan_out=0, loc=(30-10+1)=21 -> score = 3*2 + 0 + 21//50 = 6 + assert hot.hotspot_score == 6 + cold = next(u for u in ranked if u.qualname == "pkg.cold_func") + # fan_in=0, fan_out=0, loc=(40-35+1)=6 -> score = 0 + assert cold.hotspot_score == 0 + + def test_blast_radius_populated_for_hot_func(self) -> None: + g = _make_graph() + ranked = rank_untested(g) + hot = next(u for u in ranked if u.qualname == "pkg.hot_func") + # 3 direct callers should be in blast radius + assert hot.blast_radius_size >= 3 + assert hot.blast_files >= 1 + + def test_blast_radius_zero_for_cold_func(self) -> None: + g = _make_graph() + ranked = rank_untested(g) + cold = next(u for u in ranked if u.qualname == "pkg.cold_func") + # cold_func has no callers, blast radius skipped -> 0 + assert cold.blast_radius_size == 0 + assert cold.blast_files == 0 + + def test_top_parameter_truncates(self) -> None: + g = _make_graph() + ranked = rank_untested(g, top=1) + assert len(ranked) == 1 + assert ranked[0].qualname == "pkg.hot_func" + + def test_find_untested_not_affected(self) -> None: + """find_untested still works and new fields default to 0.""" + g = _make_graph() + rows = find_untested(g) + for row in rows: + assert row.hotspot_score == 0 + assert row.blast_radius_size == 0 + assert row.blast_files == 0 + + def test_test_caller_removes_from_untested(self) -> None: + """When a test module calls hot_func it is no longer untested.""" + g = _make_graph_with_test_caller() + ranked = rank_untested(g) + qualnames = [u.qualname for u in ranked] + assert "pkg.hot_func" not in qualnames + assert "pkg.cold_func" in qualnames diff --git a/tests/test_review_rules_untested_hotspot.py b/tests/test_review_rules_untested_hotspot.py new file mode 100644 index 0000000..1710d29 --- /dev/null +++ b/tests/test_review_rules_untested_hotspot.py @@ -0,0 +1,126 @@ +"""Tests for the new_untested_hotspot review rule.""" +from __future__ import annotations + +import shutil +from pathlib import Path + +import networkx as nx +import pytest + +from codegraph.graph.builder import GraphBuilder +from codegraph.graph.store_networkx import to_digraph +from codegraph.graph.store_sqlite import SQLiteGraphStore +from codegraph.review.differ import diff_graphs +from codegraph.review.rules import DEFAULT_RULES, Rule, evaluate_rules + +FIXTURES = Path(__file__).parent / "fixtures" + + +def _build_graph(repo: Path, db_path: Path) -> nx.MultiDiGraph: + store = SQLiteGraphStore(db_path) + GraphBuilder(repo, store).build(incremental=False) + g = to_digraph(store) + store.close() + return g + + +@pytest.fixture +def hotspot_graphs( + tmp_path: Path, +) -> tuple[nx.MultiDiGraph, nx.MultiDiGraph]: + """v1 → v2: v2 adds hot_helper with 5 callers but no test coverage.""" + old_repo = tmp_path / "old" + new_repo = tmp_path / "new" + old_repo.mkdir() + new_repo.mkdir() + shutil.copytree(FIXTURES / "python_hotspot_v1", old_repo / "pkg") + shutil.copytree(FIXTURES / "python_hotspot_v2", new_repo / "pkg") + old_g = _build_graph(old_repo, tmp_path / "old.db") + new_g = _build_graph(new_repo, tmp_path / "new.db") + return old_g, new_g + + +@pytest.fixture +def hotspot_tested_graphs( + tmp_path: Path, +) -> tuple[nx.MultiDiGraph, nx.MultiDiGraph]: + """v1 → v2_tested: same as above but v2 test suite does call hot_helper.""" + old_repo = tmp_path / "old" + new_repo = tmp_path / "new" + old_repo.mkdir() + new_repo.mkdir() + shutil.copytree(FIXTURES / "python_hotspot_v1", old_repo / "pkg") + shutil.copytree(FIXTURES / "python_hotspot_v2_tested", new_repo / "pkg") + old_g = _build_graph(old_repo, tmp_path / "old.db") + new_g = _build_graph(new_repo, tmp_path / "new.db") + return old_g, new_g + + +class TestNewUntestedHotspotRule: + def test_rule_fires_for_high_fan_in_new_function( + self, + hotspot_graphs: tuple[nx.MultiDiGraph, nx.MultiDiGraph], + ) -> None: + """Rule fires when a new function has >= 5 callers and no test coverage.""" + old_g, new_g = hotspot_graphs + diff = diff_graphs(old_g, new_g) + rules = [ + Rule( + id="new-untested-hotspot", + when="new_untested_hotspot", + severity="high", + message="New function with {fan_in} callers and no test coverage", + threshold=5, + ) + ] + findings = evaluate_rules(diff, new_graph=new_g, old_graph=old_g, rules=rules) + matching = [f for f in findings if f.rule_id == "new-untested-hotspot"] + assert matching, "Expected the rule to fire for hot_helper" + assert any("hot_helper" in f.qualname for f in matching) + + def test_rule_suppressed_by_test_caller( + self, + hotspot_tested_graphs: tuple[nx.MultiDiGraph, nx.MultiDiGraph], + ) -> None: + """Rule does not fire when the test suite already calls hot_helper.""" + old_g, new_g = hotspot_tested_graphs + diff = diff_graphs(old_g, new_g) + rules = [ + Rule( + id="new-untested-hotspot", + when="new_untested_hotspot", + severity="high", + message="New function with {fan_in} callers and no test coverage", + threshold=5, + ) + ] + findings = evaluate_rules(diff, new_graph=new_g, old_graph=old_g, rules=rules) + matching = [f for f in findings if f.rule_id == "new-untested-hotspot"] + assert not any( + "hot_helper" in f.qualname for f in matching + ), "Rule should be suppressed when test module calls hot_helper" + + def test_rule_in_default_rules(self) -> None: + """Confirm new-untested-hotspot is in DEFAULT_RULES.""" + ids = [r.id for r in DEFAULT_RULES] + assert "new-untested-hotspot" in ids + + def test_threshold_below_five_does_not_fire( + self, + hotspot_graphs: tuple[nx.MultiDiGraph, nx.MultiDiGraph], + ) -> None: + """With threshold=10 the rule should not fire (only 5 callers).""" + old_g, new_g = hotspot_graphs + diff = diff_graphs(old_g, new_g) + rules = [ + Rule( + id="new-untested-hotspot", + when="new_untested_hotspot", + severity="high", + message="New function with {fan_in} callers and no test coverage", + threshold=10, + ) + ] + findings = evaluate_rules(diff, new_graph=new_g, old_graph=old_g, rules=rules) + matching = [f for f in findings if f.rule_id == "new-untested-hotspot"] + assert not matching, "With threshold=10, 5 callers should not trigger rule"