diff --git a/src/cgis/query/analysis/module_graph.py b/src/cgis/query/analysis/module_graph.py index 1f817116..4dbf29af 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,9 +72,10 @@ 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]: - """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: @@ -83,6 +84,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, 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 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 +119,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 +228,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 +278,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..838d4a44 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: @@ -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 == []