diff --git a/code_review_graph/constants.py b/code_review_graph/constants.py index 2985a5e1..993ca380 100644 --- a/code_review_graph/constants.py +++ b/code_review_graph/constants.py @@ -2,8 +2,33 @@ from __future__ import annotations +import math import os + +def _bounded_float_env( + name: str, + default: float, + *, + lower: float, + upper: float, +) -> float: + """Read a finite float strictly inside ``(lower, upper)``. + + Invalid environment configuration falls back to the documented default + instead of making graph traversal unbounded or failing during import. + """ + raw = os.environ.get(name) + if raw is None: + return default + try: + value = float(raw) + except (TypeError, ValueError): + return default + if not math.isfinite(value) or not lower < value < upper: + return default + return value + SECURITY_KEYWORDS: frozenset[str] = frozenset({ "auth", "login", "password", "token", "session", "crypt", "secret", "credential", "permission", "sql", "query", "execute", "connect", @@ -19,5 +44,30 @@ MAX_BFS_DEPTH = int(os.environ.get("CRG_MAX_BFS_DEPTH", "15")) MAX_SEARCH_RESULTS = int(os.environ.get("CRG_MAX_SEARCH_RESULTS", "20")) -# BFS engine: "sql" (SQLite recursive CTE) or "networkx" (Python-side BFS) +# Impact traversal engine: "sql" (bounded SQLite relaxation) or "networkx". BFS_ENGINE = os.environ.get("CRG_BFS_ENGINE", "sql") + +# --------------------------------------------------------------------------- +# Impact-radius scoring +# --------------------------------------------------------------------------- +# Each hop multiplies the best score so strongly coupled nodes rank first. +# These review-risk weights intentionally differ from community-clustering +# affinity weights. +IMPACT_EDGE_WEIGHTS: dict[str, float] = { + "CALLS": 1.0, + "INHERITS": 0.9, + "OVERRIDES": 0.9, + "IMPLEMENTS": 0.9, + "TESTED_BY": 0.7, + "REFERENCES": 0.6, + "DEPENDS_ON": 0.6, + "IMPORTS_FROM": 0.5, + "CONTAINS": 0.3, +} +IMPACT_DEFAULT_EDGE_WEIGHT = 0.5 +IMPACT_DEPTH_DECAY = _bounded_float_env( + "CRG_IMPACT_DEPTH_DECAY", 0.6, lower=0.0, upper=1.0, +) +IMPACT_SCORE_FLOOR = _bounded_float_env( + "CRG_IMPACT_SCORE_FLOOR", 0.05, lower=0.0, upper=1.0, +) diff --git a/code_review_graph/graph.py b/code_review_graph/graph.py index dc6c33e4..18dc8d66 100644 --- a/code_review_graph/graph.py +++ b/code_review_graph/graph.py @@ -19,7 +19,15 @@ import networkx as nx -from .constants import BFS_ENGINE, MAX_IMPACT_DEPTH, MAX_IMPACT_NODES +from .constants import ( + BFS_ENGINE, + IMPACT_DEFAULT_EDGE_WEIGHT, + IMPACT_DEPTH_DECAY, + IMPACT_EDGE_WEIGHTS, + IMPACT_SCORE_FLOOR, + MAX_IMPACT_DEPTH, + MAX_IMPACT_NODES, +) from .migrations import get_schema_version, run_migrations from .parser import EdgeInfo, NodeInfo @@ -626,9 +634,10 @@ def get_impact_radius( Returns dict with: - changed_nodes: nodes in changed files - - impacted_nodes: nodes reachable via edges + - impacted_nodes: reachable nodes ordered by best-path impact score - impacted_files: unique set of affected files - edges: connecting edges + - impact_scores: qualified name to best-path score """ if BFS_ENGINE == "networkx": return self._get_impact_radius_networkx( @@ -638,7 +647,7 @@ def get_impact_radius( changed_files, max_depth=max_depth, max_nodes=max_nodes, ) - # -- SQLite recursive CTE version (default) --------------------------- + # -- Bounded SQLite relaxation version (default) ---------------------- def get_impact_radius_sql( self, @@ -646,11 +655,13 @@ def get_impact_radius_sql( max_depth: int = MAX_IMPACT_DEPTH, max_nodes: int = MAX_IMPACT_NODES, ) -> dict[str, Any]: - """Impact radius via SQLite recursive CTE. + """Impact radius via bounded best-score relaxation in SQLite. Faster than NetworkX for large graphs because it avoids materialising the full graph in Python. """ + max_depth = max(0, int(max_depth)) + max_nodes = max(0, int(max_nodes)) if not changed_files: return { "changed_nodes": [], @@ -659,6 +670,7 @@ def get_impact_radius_sql( "edges": [], "truncated": False, "total_impacted": 0, + "impact_scores": {}, } # Seed qualified names @@ -676,10 +688,11 @@ def get_impact_radius_sql( "edges": [], "truncated": False, "total_impacted": 0, + "impact_scores": {}, } - # Build recursive CTE — use a temp table for the seed set to - # keep the query plan efficient and stay under variable limits. + # Use a temp table for the seed set to keep the query plan efficient + # and stay under SQLite variable limits. self._conn.execute( "CREATE TEMP TABLE IF NOT EXISTS _impact_seeds " "(qn TEXT PRIMARY KEY)" @@ -695,44 +708,120 @@ def get_impact_radius_sql( batch, ) - cte_sql = """ - WITH RECURSIVE impacted(node_qn, depth) AS ( - SELECT qn, 0 FROM _impact_seeds - UNION - SELECT e.target_qualified, i.depth + 1 - FROM impacted i - JOIN edges e ON e.source_qualified = i.node_qn - WHERE i.depth < ? - UNION - SELECT e.source_qualified, i.depth + 1 - FROM impacted i - JOIN edges e ON e.target_qualified = i.node_qn - WHERE i.depth < ? + # Keep one best score per endpoint rather than enumerating every path + # in a recursive CTE. Dense cyclic graphs can contain exponentially + # many paths; these three bounded temp tables contain at most one row + # per qualified name and each iteration scans the edge table once. + self._conn.execute( + "CREATE TEMP TABLE IF NOT EXISTS _impact_weights " + "(kind TEXT PRIMARY KEY, weight REAL NOT NULL)" + ) + self._conn.execute("DELETE FROM _impact_weights") + self._conn.executemany( + "INSERT INTO _impact_weights (kind, weight) VALUES (?, ?)", + list(IMPACT_EDGE_WEIGHTS.items()), + ) + for table in ("_impact_best", "_impact_frontier", "_impact_next"): + self._conn.execute( + f"CREATE TEMP TABLE IF NOT EXISTS {table} " # nosec B608 + "(node_qn TEXT PRIMARY KEY, score REAL NOT NULL)" + ) + self._conn.execute(f"DELETE FROM {table}") # nosec B608 + + self._conn.execute( + "INSERT INTO _impact_best (node_qn, score) " + "SELECT qn, 1.0 FROM _impact_seeds" ) - SELECT DISTINCT node_qn, MIN(depth) AS min_depth - FROM impacted + self._conn.execute( + "INSERT INTO _impact_frontier (node_qn, score) " + "SELECT qn, 1.0 FROM _impact_seeds" + ) + + candidate_sql = """ + INSERT INTO _impact_next (node_qn, score) + SELECT node_qn, MAX(score) + FROM ( + SELECT e.target_qualified AS node_qn, + f.score * COALESCE(w.weight, ?) * ? AS score + FROM _impact_frontier f + JOIN edges e ON e.source_qualified = f.node_qn + LEFT JOIN _impact_weights w ON w.kind = e.kind + UNION ALL + SELECT e.source_qualified AS node_qn, + f.score * COALESCE(w.weight, ?) * ? AS score + FROM _impact_frontier f + JOIN edges e ON e.target_qualified = f.node_qn + LEFT JOIN _impact_weights w ON w.kind = e.kind + ) candidates + WHERE score > ? GROUP BY node_qn - LIMIT ? """ + candidate_params = ( + IMPACT_DEFAULT_EDGE_WEIGHT, + IMPACT_DEPTH_DECAY, + IMPACT_DEFAULT_EDGE_WEIGHT, + IMPACT_DEPTH_DECAY, + IMPACT_SCORE_FLOOR, + ) + for _ in range(max_depth): + self._conn.execute("DELETE FROM _impact_next") + self._conn.execute(candidate_sql, candidate_params) + self._conn.execute( + "DELETE FROM _impact_next " + "WHERE score <= COALESCE((" + "SELECT score FROM _impact_best b " + "WHERE b.node_qn = _impact_next.node_qn" + "), 0.0)" + ) + if self._conn.execute( + "SELECT 1 FROM _impact_next LIMIT 1" + ).fetchone() is None: + break + self._conn.execute( + "INSERT OR REPLACE INTO _impact_best (node_qn, score) " + "SELECT node_qn, score FROM _impact_next" + ) + self._conn.execute("DELETE FROM _impact_frontier") + self._conn.execute( + "INSERT INTO _impact_frontier (node_qn, score) " + "SELECT node_qn, score FROM _impact_next" + ) + + # Fetch one sentinel beyond the public cap. Ghost endpoints remain in + # the frontier as bridges but cannot consume a result slot because the + # final selection joins the canonical nodes table. rows = self._conn.execute( - cte_sql, (max_depth, max_depth, max_nodes + len(seeds)), + "SELECT b.node_qn, b.score " + "FROM _impact_best b " + "JOIN nodes n ON n.qualified_name = b.node_qn " + "LEFT JOIN _impact_seeds s ON s.qn = b.node_qn " + "WHERE s.qn IS NULL " + "ORDER BY b.score DESC, b.node_qn " + "LIMIT ?", + (max_nodes + 1,), ).fetchall() + truncated = len(rows) > max_nodes + if truncated: + total_impacted = self._conn.execute( + "SELECT COUNT(*) " + "FROM _impact_best b " + "JOIN nodes n ON n.qualified_name = b.node_qn " + "LEFT JOIN _impact_seeds s ON s.qn = b.node_qn " + "WHERE s.qn IS NULL" + ).fetchone()[0] + else: + total_impacted = len(rows) + kept_rows = rows[:max_nodes] + score_by_qn = {row[0]: float(row[1]) for row in kept_rows} - # Split into seeds vs impacted - impacted_qns: set[str] = set() - for r in rows: - qn = r[0] - if qn not in seeds: - impacted_qns.add(qn) - - # Batch-fetch nodes changed_nodes = self._batch_get_nodes(seeds) - impacted_nodes = self._batch_get_nodes(impacted_qns) - - total_impacted = len(impacted_nodes) - truncated = total_impacted > max_nodes - if truncated: - impacted_nodes = impacted_nodes[:max_nodes] + impacted_nodes = self._batch_get_nodes(set(score_by_qn)) + impacted_nodes.sort( + key=lambda node: ( + -score_by_qn.get(node.qualified_name, 0.0), + node.qualified_name, + ) + ) impacted_files = list({n.file_path for n in impacted_nodes}) @@ -748,6 +837,12 @@ def get_impact_radius_sql( "edges": relevant_edges, "truncated": truncated, "total_impacted": total_impacted, + "impact_scores": { + node.qualified_name: round( + score_by_qn.get(node.qualified_name, 0.0), 4, + ) + for node in impacted_nodes + }, } # -- NetworkX BFS version (legacy) ------------------------------------ @@ -759,6 +854,8 @@ def _get_impact_radius_networkx( max_nodes: int = MAX_IMPACT_NODES, ) -> dict[str, Any]: """BFS via NetworkX (legacy). Used when CRG_BFS_ENGINE=networkx.""" + max_depth = max(0, int(max_depth)) + max_nodes = max(0, int(max_nodes)) nxg = self._build_networkx_graph() seeds: set[str] = set() @@ -767,34 +864,44 @@ def _get_impact_radius_networkx( for n in nodes: seeds.add(n.qualified_name) - visited: set[str] = set() - frontier = seeds.copy() - depth = 0 - impacted: set[str] = set() + best: dict[str, float] = dict.fromkeys(seeds, 1.0) + frontier = dict(best) - while frontier and depth < max_depth: - visited.update(frontier) - next_frontier: set[str] = set() - for qn in frontier: - if qn in nxg: - for neighbor in nxg.neighbors(qn): - if neighbor not in visited: - next_frontier.add(neighbor) - impacted.add(neighbor) - if qn in nxg: - for pred in nxg.predecessors(qn): - if pred not in visited: - next_frontier.add(pred) - impacted.add(pred) - next_frontier -= visited - if len(visited) + len(next_frontier) > max_nodes: + for _ in range(max_depth): + if not frontier: break + next_frontier: dict[str, float] = {} + for qn, score in frontier.items(): + if qn not in nxg: + continue + neighbors = [ + (target, data) + for _, target, data in nxg.out_edges(qn, data=True) + ] + [ + (source, data) + for source, _, data in nxg.in_edges(qn, data=True) + ] + for other_qn, data in neighbors: + weight = IMPACT_EDGE_WEIGHTS.get( + data.get("kind", ""), IMPACT_DEFAULT_EDGE_WEIGHT, + ) + new_score = score * weight * IMPACT_DEPTH_DECAY + if new_score <= IMPACT_SCORE_FLOOR: + continue + if new_score > best.get(other_qn, 0.0): + best[other_qn] = new_score + next_frontier[other_qn] = new_score frontier = next_frontier - depth += 1 changed_nodes = self._batch_get_nodes(seeds) - impacted_qns = impacted - seeds + impacted_qns = set(best) - seeds impacted_nodes = self._batch_get_nodes(impacted_qns) + impacted_nodes.sort( + key=lambda node: ( + -best.get(node.qualified_name, 0.0), + node.qualified_name, + ) + ) total_impacted = len(impacted_nodes) truncated = total_impacted > max_nodes @@ -815,6 +922,12 @@ def _get_impact_radius_networkx( "edges": relevant_edges, "truncated": truncated, "total_impacted": total_impacted, + "impact_scores": { + node.qualified_name: round( + best.get(node.qualified_name, 0.0), 4, + ) + for node in impacted_nodes + }, } def get_subgraph(self, qualified_names: list[str]) -> dict[str, Any]: @@ -1285,14 +1398,27 @@ def load_flow_adjacency(self) -> "FlowAdjacency": # --- Internal helpers --- def _build_networkx_graph(self) -> nx.DiGraph: - """Build (or return cached) in-memory NetworkX directed graph from all edges.""" + """Build a directed graph, retaining the strongest parallel edge.""" with self._cache_lock: if self._nxg_cache is not None: return self._nxg_cache g: nx.DiGraph = nx.DiGraph() rows = self._conn.execute("SELECT * FROM edges").fetchall() for r in rows: - g.add_edge(r["source_qualified"], r["target_qualified"], kind=r["kind"]) + source = r["source_qualified"] + target = r["target_qualified"] + kind = r["kind"] + if g.has_edge(source, target): + existing = g[source][target].get("kind", "") + existing_weight = IMPACT_EDGE_WEIGHTS.get( + existing, IMPACT_DEFAULT_EDGE_WEIGHT, + ) + candidate_weight = IMPACT_EDGE_WEIGHTS.get( + kind, IMPACT_DEFAULT_EDGE_WEIGHT, + ) + if candidate_weight <= existing_weight: + continue + g.add_edge(source, target, kind=kind) self._nxg_cache = g return g diff --git a/code_review_graph/tools/query.py b/code_review_graph/tools/query.py index 41646c00..3542a42b 100644 --- a/code_review_graph/tools/query.py +++ b/code_review_graph/tools/query.py @@ -80,8 +80,15 @@ def get_impact_radius( abs_files, max_depth=max_depth, max_nodes=max_results ) + impact_scores = result.get("impact_scores", {}) changed_dicts = [node_to_dict(n) for n in result["changed_nodes"]] - impacted_dicts = [node_to_dict(n) for n in result["impacted_nodes"]] + impacted_dicts = [] + for node in result["impacted_nodes"]: + node_dict = node_to_dict(node) + score = impact_scores.get(node.qualified_name) + if score is not None: + node_dict["impact_score"] = score + impacted_dicts.append(node_dict) edge_dicts = [edge_to_dict(e) for e in result["edges"]] truncated = result["truncated"] total_impacted = result["total_impacted"] diff --git a/tests/test_graph.py b/tests/test_graph.py index 4c92af39..1865ec89 100644 --- a/tests/test_graph.py +++ b/tests/test_graph.py @@ -3,8 +3,12 @@ import logging import sqlite3 import tempfile +import time from pathlib import Path +import pytest + +import code_review_graph.constants as constants_module from code_review_graph.graph import GraphStore from code_review_graph.parser import EdgeInfo, NodeInfo @@ -457,6 +461,209 @@ def test_empty_changed_files(self): assert result["total_impacted"] == 0 +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ("0.75", 0.75), + ("", 0.6), + ("not-a-number", 0.6), + ("nan", 0.6), + ("inf", 0.6), + ("-0.1", 0.6), + ("0", 0.6), + ("1", 0.6), + ("1.2", 0.6), + ], +) +def test_impact_float_configuration_is_finite_and_bounded( + monkeypatch, raw, expected, +): + monkeypatch.setenv("CRG_TEST_IMPACT_FLOAT", raw) + assert constants_module._bounded_float_env( + "CRG_TEST_IMPACT_FLOAT", 0.6, lower=0.0, upper=1.0, + ) == pytest.approx(expected) + + +class TestWeightedImpactScoring: + """Best-path scoring stays ranked, bounded, and engine-independent.""" + + def setup_method(self): + self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + self.tmp.close() + self.store = GraphStore(self.tmp.name) + + def teardown_method(self): + self.store.close() + Path(self.tmp.name).unlink(missing_ok=True) + + def _add_func(self, name: str, path: str) -> str: + self.store.upsert_node(NodeInfo( + kind="Function", name=name, file_path=path, + line_start=1, line_end=10, language="python", + )) + return f"{path}::{name}" + + def _add_edge( + self, kind: str, source: str, target: str, line: int = 1, + ) -> None: + self.store.upsert_edge(EdgeInfo( + kind=kind, source=source, target=target, + file_path="/seed.py", line=line, + )) + + @staticmethod + def _ordered_qns(result) -> list[str]: + return [node.qualified_name for node in result["impacted_nodes"]] + + def test_edge_weights_rank_best_path_and_engines_match(self): + seed = self._add_func("seed", "/seed.py") + called = self._add_func("called", "/called.py") + imported = self._add_func("imported", "/imported.py") + indirect = self._add_func("indirect", "/indirect.py") + self._add_edge("CALLS", seed, called) + self._add_edge("IMPORTS_FROM", seed, imported) + self._add_edge("CALLS", called, indirect) + self.store.commit() + + sql = self.store.get_impact_radius_sql(["/seed.py"], max_depth=2) + nx_result = self.store._get_impact_radius_networkx( + ["/seed.py"], max_depth=2, + ) + + assert sql["impact_scores"][called] == pytest.approx(0.6) + assert sql["impact_scores"][indirect] == pytest.approx(0.36) + assert sql["impact_scores"][imported] == pytest.approx(0.3) + assert self._ordered_qns(sql) == [called, indirect, imported] + assert sql["impact_scores"] == nx_result["impact_scores"] + assert self._ordered_qns(sql) == self._ordered_qns(nx_result) + + def test_deeper_strong_path_beats_shallow_weak_path(self): + seed = self._add_func("seed", "/seed.py") + middle = self._add_func("middle", "/middle.py") + target = self._add_func("target", "/target.py") + self._add_edge("CONTAINS", seed, target) + self._add_edge("CALLS", seed, middle, line=2) + self._add_edge("CALLS", middle, target, line=3) + self.store.commit() + + sql = self.store.get_impact_radius_sql(["/seed.py"], max_depth=2) + nx_result = self.store._get_impact_radius_networkx( + ["/seed.py"], max_depth=2, + ) + + assert sql["impact_scores"][target] == pytest.approx(0.36) + assert sql["impact_scores"] == nx_result["impact_scores"] + + def test_score_floor_stops_expansion_in_both_engines(self): + qns = [ + self._add_func(f"node_{index}", f"/node_{index}.py") + for index in range(8) + ] + for index, (source, target) in enumerate(zip(qns, qns[1:])): + self._add_edge("CALLS", source, target, line=index + 1) + self.store.commit() + + sql = self.store.get_impact_radius_sql( + ["/node_0.py"], max_depth=8, + ) + nx_result = self.store._get_impact_radius_networkx( + ["/node_0.py"], max_depth=8, + ) + + assert qns[5] in sql["impact_scores"] + assert qns[6] not in sql["impact_scores"] + assert sql["impact_scores"] == nx_result["impact_scores"] + + def test_unknown_edge_kind_uses_default_weight(self): + seed = self._add_func("seed", "/seed.py") + target = self._add_func("target", "/target.py") + self._add_edge("UNKNOWN_KIND", seed, target) + self.store.commit() + + sql = self.store.get_impact_radius_sql(["/seed.py"], max_depth=1) + nx_result = self.store._get_impact_radius_networkx( + ["/seed.py"], max_depth=1, + ) + + assert sql["impact_scores"][target] == pytest.approx(0.3) + assert sql["impact_scores"] == nx_result["impact_scores"] + + def test_truncation_is_exact_at_boundary_and_uses_sentinel(self): + seed = self._add_func("seed", "/seed.py") + targets = [ + self._add_func(f"target_{index}", f"/target_{index}.py") + for index in range(3) + ] + for index, target in enumerate(targets): + self._add_edge("CALLS", seed, target, line=index + 1) + self.store.commit() + + exact = self.store.get_impact_radius_sql( + ["/seed.py"], max_depth=1, max_nodes=3, + ) + capped = self.store.get_impact_radius_sql( + ["/seed.py"], max_depth=1, max_nodes=2, + ) + + assert exact["truncated"] is False + assert exact["total_impacted"] == 3 + assert capped["truncated"] is True + assert capped["total_impacted"] == 3 + assert len(capped["impacted_nodes"]) == 2 + + def test_ghost_endpoint_bridges_without_consuming_limit(self): + seed = self._add_func("seed", "/seed.py") + target = self._add_func("target", "/target.py") + ghost = "external.package::ghost" + self._add_edge("CALLS", seed, ghost) + self._add_edge("CALLS", ghost, target, line=2) + self.store.commit() + + result = self.store.get_impact_radius_sql( + ["/seed.py"], max_depth=2, max_nodes=1, + ) + + assert self._ordered_qns(result) == [target] + assert ghost not in result["impact_scores"] + assert result["truncated"] is False + + def test_parallel_edges_use_strongest_weight_in_both_engines(self): + seed = self._add_func("seed", "/seed.py") + target = self._add_func("target", "/target.py") + self._add_edge("CALLS", seed, target, line=1) + self._add_edge("CONTAINS", seed, target, line=2) + self.store.commit() + + sql = self.store.get_impact_radius_sql(["/seed.py"], max_depth=1) + nx_result = self.store._get_impact_radius_networkx( + ["/seed.py"], max_depth=1, + ) + assert sql["impact_scores"][target] == pytest.approx(0.6) + assert sql["impact_scores"] == nx_result["impact_scores"] + + def test_dense_mixed_cycle_is_bounded(self): + qns = [self._add_func(f"node_{i}", f"/node_{i}.py") for i in range(12)] + line = 1 + for source_index, source in enumerate(qns): + for target_index, target in enumerate(qns): + if source_index == target_index: + continue + kind = "CALLS" if (source_index + target_index) % 2 else "IMPORTS_FROM" + self._add_edge(kind, source, target, line=line) + line += 1 + self.store.commit() + + started = time.monotonic() + result = self.store.get_impact_radius_sql( + ["/node_0.py"], max_depth=25, max_nodes=20, + ) + elapsed = time.monotonic() - started + + assert len(result["impacted_nodes"]) == 11 + assert result["truncated"] is False + assert elapsed < 5.0 + + class TestGetTransitiveTestsFrontierCap: """Regression tests for O(N*M) query explosion in get_transitive_tests.""" diff --git a/tests/test_tools.py b/tests/test_tools.py index b81f5bec..bbb6294d 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -11,6 +11,7 @@ import code_review_graph.tools._common as common_module import code_review_graph.tools.analysis_tools as analysis_module import code_review_graph.tools.docs as docs_module +import code_review_graph.tools.query as query_module from code_review_graph.graph import GraphStore, _sanitize_name, node_to_dict from code_review_graph.parser import EdgeInfo, NodeInfo from code_review_graph.tools import ( @@ -1945,3 +1946,48 @@ def test_registered_sync_tool_preserves_existing_fields(self, tmp_path): assert result == expected assert envelope["updated_at"] == "2000-01-02T03:04:05" assert envelope["built_on_branch"] == "main" + + +def test_impact_radius_tool_exposes_best_first_scores(monkeypatch, tmp_path): + """The public tool adds scores without changing the stored node schema.""" + store = GraphStore(tmp_path / "impact.db") + seed = "/seed.py::seed" + called = "/called.py::called" + imported = "/imported.py::imported" + for name, path in ( + ("seed", "/seed.py"), + ("called", "/called.py"), + ("imported", "/imported.py"), + ): + store.upsert_node(NodeInfo( + kind="Function", name=name, file_path=path, + line_start=1, line_end=3, language="python", + )) + store.upsert_edge(EdgeInfo( + kind="CALLS", source=seed, target=called, + file_path="/seed.py", line=1, + )) + store.upsert_edge(EdgeInfo( + kind="IMPORTS_FROM", source=seed, target=imported, + file_path="/seed.py", line=2, + )) + store.commit() + + monkeypatch.setattr( + query_module, "_get_store", lambda _repo_root: (store, tmp_path), + ) + monkeypatch.setattr( + query_module, + "_resolve_graph_file_paths", + lambda _store, _root, _files: ["/seed.py"], + ) + + result = query_module.get_impact_radius( + changed_files=["seed.py"], repo_root=str(tmp_path), + ) + + assert [node["name"] for node in result["impacted_nodes"]] == [ + "called", "imported", + ] + scores = [node["impact_score"] for node in result["impacted_nodes"]] + assert scores == sorted(scores, reverse=True)