From bf76d064caf2c3d052612548b1e7eb0264e114b3 Mon Sep 17 00:00:00 2001 From: zaebee Date: Fri, 23 Oct 2026 07:58:55 +0000 Subject: [PATCH 1/3] refactor(module-graph): clear the three Sonar issues #743 left - transitive_reduction (S3776, 18): the condensation is built in _condense, and "reachable through a third component" is _implied over a generic _reach, instead of a nested search. - build_module_graph (S3776, 16): the edge loop moves to _collect. - test_module_graph (S3415): the one assertion with its expected side first is turned round. Behaviour is unchanged: the same 25 tests pass, and query's self-drift is 0.230 (tolerance 0.24). Closes #746 Co-Authored-By: Claude Opus 5.5 --- src/cgis/query/analysis/module_graph.py | 94 ++++++++++++++----------- tests/unit/test_module_graph.py | 2 +- 2 files changed, 52 insertions(+), 44 deletions(-) diff --git a/src/cgis/query/analysis/module_graph.py b/src/cgis/query/analysis/module_graph.py index 1f817116..4720d476 100644 --- a/src/cgis/query/analysis/module_graph.py +++ b/src/cgis/query/analysis/module_graph.py @@ -23,7 +23,7 @@ from cgis.storage.sqlite_store import SQLiteStore if TYPE_CHECKING: - from collections.abc import Iterable, Sequence + from collections.abc import Hashable, Iterable, Mapping, Sequence from cgis.core.models import Edge, Node @@ -72,7 +72,7 @@ def _under(fqn: str, prefix: str) -> bool: return fqn == prefix or fqn.startswith(prefix + ".") -def _reach(succ: dict[str, set[str]], start: str) -> set[str]: +def _reach[N: Hashable](succ: Mapping[N, set[N]], start: N) -> set[N]: """Every node reachable from ``start`` by one step or more.""" stack, seen = list(succ.get(start, ())), set(succ.get(start, ())) while stack: @@ -83,6 +83,30 @@ def _reach(succ: dict[str, set[str]], start: str) -> set[str]: return seen +def _condense( + edges: list[ModuleEdge], +) -> tuple[dict[str, frozenset[str]], dict[frozenset[str], set[frozenset[str]]]]: + """Each node's strongly connected component, and the DAG between components.""" + succ: dict[str, set[str]] = {} + for e in edges: + succ.setdefault(e.source, set()).add(e.target) + succ.setdefault(e.target, set()) + reach = {n: _reach(succ, n) for n in succ} + comp = {n: frozenset({n} | {m for m in reach[n] if n in reach[m]}) for n in succ} + dag: dict[frozenset[str], set[frozenset[str]]] = {} + for e in edges: + if comp[e.source] != comp[e.target]: + dag.setdefault(comp[e.source], set()).add(comp[e.target]) + return comp, dag + + +def _implied( + dag: dict[frozenset[str], set[frozenset[str]]], a: frozenset[str], b: frozenset[str] +) -> bool: + """Is component ``b`` reachable from ``a`` through some third component.""" + return any(b in _reach(dag, c) for c in dag.get(a, ()) if c != b) + + def transitive_reduction(edges: Iterable[ModuleEdge]) -> list[ModuleEdge]: """Drop each edge implied by a longer path, keeping every edge of a cycle. @@ -94,35 +118,11 @@ def transitive_reduction(edges: Iterable[ModuleEdge]) -> list[ModuleEdge]: third. Input order is preserved. """ edge_list = list(edges) - succ: dict[str, set[str]] = {} - for e in edge_list: - succ.setdefault(e.source, set()).add(e.target) - succ.setdefault(e.target, set()) - reach = {n: _reach(succ, n) for n in succ} - comp = {n: frozenset({n} | {m for m in reach[n] if n in reach[m]}) for n in succ} - cdag: dict[frozenset[str], set[frozenset[str]]] = {} - for e in edge_list: - if comp[e.source] != comp[e.target]: - cdag.setdefault(comp[e.source], set()).add(comp[e.target]) - - def implied(a: frozenset[str], b: frozenset[str]) -> bool: - """Is component ``b`` reachable from ``a`` through some third component.""" - stack = [c for c in cdag.get(a, ()) if c != b] - seen = set(stack) - while stack: - current = stack.pop() - for nxt in cdag.get(current, ()): - if nxt == b: - return True - if nxt not in seen: - seen.add(nxt) - stack.append(nxt) - return False - + comp, dag = _condense(edge_list) return [ e for e in edge_list - if comp[e.source] == comp[e.target] or not implied(comp[e.source], comp[e.target]) + if comp[e.source] == comp[e.target] or not _implied(dag, comp[e.source], comp[e.target]) ] @@ -227,6 +227,28 @@ def _reduced( return sorted(kept, key=lambda e: (e.kind, e.source, e.target)), dropped +def _collect( + edges: Sequence[Edge], scope: _Scope, layers: list[str] +) -> tuple[list[ModuleEdge], dict[str, set[str]]]: + """The member-to-member edges of ``layers``, and per member the packages outside it reaches.""" + wanted = {edge_type: layer for layer in layers for edge_type in KINDS[layer]} + found: dict[ModuleEdge, None] = {} + outside: dict[str, set[str]] = {} + for e in edges: + kind = wanted.get(e.type) + ends = None if kind is None or e.type_only else _ends(e, scope) + if kind is None or ends is None: + continue + src, dst_file = ends + if not _under(dst_file, scope.prefix): + outside.setdefault(src, set()).add(_outside_package(dst_file, scope.packages)) + continue + dst = _group(dst_file, scope.prefix, scope.expand) + if dst in scope.members and dst != src: + found[ModuleEdge(src, dst, kind)] = None + return list(found), outside + + def build_module_graph( nodes: Sequence[Node], edges: Sequence[Edge], @@ -255,21 +277,7 @@ def build_module_graph( members=set(members), through=_imported_through(edges, holder) if expand else {}, ) - wanted = {edge_type: layer for layer in chosen for edge_type in KINDS[layer]} - found: dict[ModuleEdge, None] = {} - outside: dict[str, set[str]] = {} - for e in edges: - kind = wanted.get(e.type) - ends = None if kind is None or e.type_only else _ends(e, scope) - if kind is None or ends is None: - continue - src, dst_file = ends - if not _under(dst_file, prefix): - outside.setdefault(src, set()).add(_outside_package(dst_file, packages)) - continue - dst = _group(dst_file, prefix, expand) - if dst in scope.members and dst != src: - found[ModuleEdge(src, dst, kind)] = None + found, outside = _collect(edges, scope, chosen) kept, dropped = _reduced(found, chosen, reduce) return ModuleGraph( prefix=prefix, diff --git a/tests/unit/test_module_graph.py b/tests/unit/test_module_graph.py index ff85e263..1ca7c257 100644 --- a/tests/unit/test_module_graph.py +++ b/tests/unit/test_module_graph.py @@ -343,7 +343,7 @@ def test_mermaid_ids_stay_distinct_when_names_flatten_alike() -> None: ids = [line.split("[")[0].strip() for line in lines if '["' in line] assert len(set(ids)) == 2 edge = next(line for line in lines if "-->" in line).split(" --> ") - assert {edge[0].strip(), edge[1].strip()} == set(ids) + assert set(ids) == {edge[0].strip(), edge[1].strip()} def test_cli_reports_a_bad_layer_before_touching_the_graph(tmp_path: Path) -> None: From 0d33e960f6910555133f53439acebb7bafe47684 Mon Sep 17 00:00:00 2001 From: zaebee Date: Fri, 23 Oct 2026 08:17:15 +0000 Subject: [PATCH 2/3] perf(module-graph): one walk per implied check again Review of #747: _implied ran a separate full _reach from each successor, where the code it replaced did one DFS from all of them, so the "behaviour- preserving" refactor made the reduction up to 2.6x slower on a synthetic 2000-node / 12000-edge graph. _reach now takes several starts, and _implied walks once from all of a's other successors. Measured with the reviewer's bench against the pre-refactor function: same output on 3000 random graphs; ratio now 1.0x / 1.5x / 1.2x (cyclic 0.05, cyclic 0.01, DAG; was 1.0x / 2.6x / 1.9x). The remainder is the old early exit on reaching b, kept out to stay under Sonar's complexity limit; real module graphs have tens of nodes. Co-Authored-By: Claude Opus 5.5 --- src/cgis/query/analysis/module_graph.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/src/cgis/query/analysis/module_graph.py b/src/cgis/query/analysis/module_graph.py index 4720d476..4dbf29af 100644 --- a/src/cgis/query/analysis/module_graph.py +++ b/src/cgis/query/analysis/module_graph.py @@ -72,9 +72,10 @@ def _under(fqn: str, prefix: str) -> bool: return fqn == prefix or fqn.startswith(prefix + ".") -def _reach[N: Hashable](succ: Mapping[N, set[N]], start: N) -> set[N]: - """Every node reachable from ``start`` by one step or more.""" - stack, seen = list(succ.get(start, ())), set(succ.get(start, ())) +def _reach[N: Hashable](succ: Mapping[N, set[N]], starts: Iterable[N]) -> set[N]: + """``starts`` and every node reachable from them, in one walk.""" + stack = list(starts) + seen = set(stack) while stack: for nxt in succ.get(stack.pop(), ()): if nxt not in seen: @@ -91,7 +92,7 @@ def _condense( for e in edges: succ.setdefault(e.source, set()).add(e.target) succ.setdefault(e.target, set()) - reach = {n: _reach(succ, n) for n in succ} + reach = {n: _reach(succ, succ[n]) for n in succ} comp = {n: frozenset({n} | {m for m in reach[n] if n in reach[m]}) for n in succ} dag: dict[frozenset[str], set[frozenset[str]]] = {} for e in edges: @@ -104,7 +105,7 @@ def _implied( dag: dict[frozenset[str], set[frozenset[str]]], a: frozenset[str], b: frozenset[str] ) -> bool: """Is component ``b`` reachable from ``a`` through some third component.""" - return any(b in _reach(dag, c) for c in dag.get(a, ()) if c != b) + return b in _reach(dag, [c for c in dag.get(a, ()) if c != b]) def transitive_reduction(edges: Iterable[ModuleEdge]) -> list[ModuleEdge]: From 83129bc138c73d57d88095fad3590a4064356943 Mon Sep 17 00:00:00 2001 From: zaebee Date: Fri, 23 Oct 2026 08:34:50 +0000 Subject: [PATCH 3/3] test(module-graph): a type-only import draws no edge From guardian's review of #747: the type_only filter in the edge collection had no test. One now ingests an import under TYPE_CHECKING and expects no edge; it fails with the filter removed. Co-Authored-By: Claude Opus 5.5 --- tests/unit/test_module_graph.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/tests/unit/test_module_graph.py b/tests/unit/test_module_graph.py index 1ca7c257..838d4a44 100644 --- a/tests/unit/test_module_graph.py +++ b/tests/unit/test_module_graph.py @@ -352,3 +352,23 @@ def test_cli_reports_a_bad_layer_before_touching_the_graph(tmp_path: Path) -> No app, ["module-graph", "pkg", "--db", str(tmp_path / "none.db"), "-l", "vibes"] ) assert result.exit_code == 2 + + +def test_type_only_imports_are_not_dependencies(tmp_path: Path) -> None: + """An import under `if TYPE_CHECKING:` breaks no cycle and draws no edge (#499).""" + root = tmp_path / "repo" + for rel, text in { + "pkg/__init__.py": "", + "pkg/a.py": ( + "from typing import TYPE_CHECKING\n\nif TYPE_CHECKING:\n from pkg.b import B\n" + ), + "pkg/b.py": "class B:\n pass\n", + }.items(): + (root / rel).parent.mkdir(parents=True, exist_ok=True) + (root / rel).write_text(text) + db_path = str(tmp_path / "g.db") + store = SQLiteStore(db_path) + store.connect() + with store: + IngestionPipeline(extractors([])).run(str(root), store=store, rebuild=True) + assert module_graph(db_path, "pkg", layers=("imports",)).edges == []