Compare commits

...

1 Commits

Author SHA1 Message Date
Step35
832b23286b feat(dependency-graph): add transitive closure and deep chain analysis
Some checks failed
Test / pytest (pull_request) Failing after 7s
- Implement transitive_closure(): computes full dependency tree for each node
- Implement find_deep_chains(): identifies longest paths in dependency graph
- JSON output now includes `transitive` and `deep_chains` fields
- Added comprehensive unit tests in scripts/test_dependency_graph.py (9 tests)
- Handles cycles correctly, excludes self-references from closure

Meets acceptance criteria for #111:
   Builds transitive dep tree
   Identifies deep chains and circular deps
   Output: transitive dependency graph (via --format json)

Closes #111
2026-04-26 05:08:23 -04:00
2 changed files with 245 additions and 0 deletions

View File

@@ -180,6 +180,89 @@ def to_mermaid(graph: dict) -> str:
return "\n".join(lines)
def transitive_closure(graph: dict) -> dict:
"""Compute transitive closure for each node (all indirect deps)."""
closure = {}
# Build adjacency list
adj = {node: set(data.get("dependencies", [])) for node, data in graph.items()}
all_nodes = set(adj.keys()) | set().union(*adj.values())
for node in all_nodes:
visited = set()
stack = list(adj.get(node, set()))
while stack:
current = stack.pop()
if current not in visited:
visited.add(current)
stack.extend(adj.get(current, set()))
# Remove self-reference: a node's transitive deps should not include itself
visited.discard(node)
closure[node] = visited
return closure
def find_deep_chains(graph: dict) -> list[list[str]]:
"""Find the longest simple paths in the dependency graph (ignoring cycles)."""
from collections import defaultdict
adj = {node: list(data.get("dependencies", [])) for node, data in graph.items()}
deepest = []
max_len = 0
def dfs(node: str, path: list, visited: set):
nonlocal deepest, max_len
# Stop if we hit a cycle (node already in path)
if node in path:
return
new_path = path + [node]
if node not in adj or not adj[node]:
# leaf
if len(new_path) > max_len:
max_len = len(new_path)
deepest = [new_path.copy()]
elif len(new_path) == max_len:
deepest.append(new_path.copy())
else:
for neighbor in adj[node]:
dfs(neighbor, new_path.copy(), visited | {node})
for start in graph:
dfs(start, [], set())
return deepest
def format_transitive_markdown(closure: dict) -> str:
"""Render transitive closure as a markdown table."""
lines = ["# Transitive Dependencies\n\n"]
lines.append("| Node | Transitive Dependencies | Count |\n")
lines.append("|------|------------------------|-------|\n")
for node in sorted(closure.keys()):
deps = closure[node]
deps_str = ", ".join(sorted(deps)) if deps else "(none)"
lines.append(f"| {node} | {deps_str} | {len(deps)} |\n")
return "".join(lines)
def format_deep_chains_markdown(chains: list[list[str]]) -> str:
"""Render longest dependency chains as a markdown list."""
lines = ["# Deepest Dependency Chains\n\n"]
if not chains:
lines.append("No chains found.\n")
return "".join(lines)
max_len = max(len(c) for c in chains)
lines.append(f"*Longest chain length:* {max_len}\n\n")
for i, chain in enumerate(sorted(chains, key=lambda c: (-len(c), " -> ".join(c))), 1):
lines.append(f"**Chain {i}** ({len(chain)} nodes)\n\n")
indent = " "
for j, node in enumerate(chain):
arrow = "" if j < len(chain)-1 else ""
lines.append(f"{indent}{arrow}{node}\n")
lines.append("\n")
return "".join(lines)
def main():
parser = argparse.ArgumentParser(description="Build cross-repo dependency graph")
parser.add_argument("repos_dir", nargs="?", help="Directory containing repos")
@@ -228,13 +311,20 @@ def main():
elif args.format == "mermaid":
output = to_mermaid(results)
else:
# Compute transitive and deep chains
closure = transitive_closure(results)
deep_chains = find_deep_chains(results)
output = json.dumps({
"repos": results,
"cycles": cycles,
"transitive": {node: sorted(deps) for node, deps in closure.items()},
"deep_chains": [chain for chain in deep_chains if len(chain) > 1],
"summary": {
"total_repos": len(results),
"total_deps": sum(len(r["dependencies"]) for r in results.values()),
"cycles_found": len(cycles),
"transitive_pairs": sum(len(deps) for deps in closure.values()),
"longest_chain_length": max((len(c) for c in deep_chains), default=0),
}
}, indent=2)

View File

@@ -0,0 +1,155 @@
#!/usr/bin/env python3
"""Tests for dependency_graph.py — transitive closure and deep chain detection."""
import json
import sys
import os
import tempfile
import shutil
from pathlib import Path
sys.path.insert(0, os.path.dirname(__file__) or ".")
import importlib.util
spec = importlib.util.spec_from_file_location(
"dg", os.path.join(os.path.dirname(__file__) or ".", "dependency_graph.py")
)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
transitive_closure = mod.transitive_closure
find_deep_chains = mod.find_deep_chains
detect_cycles = mod.detect_cycles
def make_graph(edges: dict[str, list[str]]) -> dict:
"""Build graph dict in expected format: {repo: {"dependencies": [...]}}."""
return {
node: {"dependencies": sorted(deps), "files_scanned": 1}
for node, deps in edges.items()
}
def test_transitive_closure_simple_chain():
graph = make_graph({
"A": ["B"],
"B": ["C"],
"C": [],
})
closure = transitive_closure(graph)
assert closure["A"] == {"B", "C"}
assert closure["B"] == {"C"}
assert closure["C"] == set()
print("✅ Simple chain transitive closure")
def test_transitive_closure_diamond():
graph = make_graph({
"A": ["B", "C"],
"B": ["D"],
"C": ["D"],
"D": [],
})
closure = transitive_closure(graph)
assert closure["A"] == {"B", "C", "D"}
assert closure["B"] == {"D"}
assert closure["C"] == {"D"}
assert closure["D"] == set()
print("✅ Diamond closure")
def test_transitive_closure_with_cycle():
graph = make_graph({
"A": ["B"],
"B": ["C"],
"C": ["A"], # cycle
})
closure = transitive_closure(graph)
assert closure["A"] == {"B", "C"}
assert closure["B"] == {"C", "A"}
assert closure["C"] == {"A", "B"}
print("✅ Cycle in transitive closure")
def test_find_deep_chains_simple():
graph = make_graph({
"A": ["B"],
"B": ["C"],
"C": [],
})
chains = find_deep_chains(graph)
chains_sorted = sorted(chains, key=len, reverse=True)
assert len(chains_sorted) == 1
assert chains_sorted[0] == ["A", "B", "C"]
print("✅ Simple deep chain")
def test_find_deep_chains_multiple():
graph = make_graph({
"A": ["B", "C"],
"B": ["D"],
"C": ["E"],
"D": [],
"E": [],
})
chains = find_deep_chains(graph)
lengths = [len(c) for c in chains]
assert max(lengths) == 3
print("✅ Multiple chains detected")
def test_find_deep_chains_with_cycle_does_not_infinite_loop():
graph = make_graph({
"A": ["B"],
"B": ["C"],
"C": ["A"],
})
chains = find_deep_chains(graph)
print(f"✅ Cycle handled: found {len(chains)} chains")
def test_empty_graph():
graph = {}
assert transitive_closure(graph) == {}
assert find_deep_chains(graph) == []
print("✅ Empty graph handled")
def test_detect_cycles_shorthand():
graph = make_graph({
"A": ["B"],
"B": ["C"],
"C": ["A"],
})
cycles = detect_cycles(graph)
assert len(cycles) == 1
assert set(cycles[0]) == {"A", "B", "C"}
print("✅ Cycle detection works")
def test_chain_length_reporting():
graph = make_graph({
"root": ["a", "b"],
"a": ["c"],
"b": ["d"],
"c": ["e"],
"d": [],
"e": [],
})
chains = find_deep_chains(graph)
max_len = max(len(c) for c in chains)
assert max_len == 4
print(f"✅ Longest chain length: {max_len}")
if __name__ == "__main__":
test_transitive_closure_simple_chain()
test_transitive_closure_diamond()
test_transitive_closure_with_cycle()
test_find_deep_chains_simple()
test_find_deep_chains_multiple()
test_find_deep_chains_with_cycle_does_not_infinite_loop()
test_empty_graph()
test_detect_cycles_shorthand()
test_chain_length_reporting()
print("\n✅ All dependency graph tests passed")