diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index ea3754d9..e0712781 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -11,6 +11,7 @@ repos: hooks: - id: mypy args: [--config-file=pyproject.toml] + files: ^src/ additional_dependencies: - pydantic>=2.13.4 - structlog>=25.5.0 diff --git a/src/cgis/cli.py b/src/cgis/cli.py index 9b4f8f01..75d271d0 100644 --- a/src/cgis/cli.py +++ b/src/cgis/cli.py @@ -1,6 +1,6 @@ """ "CLI to run pipeline.""" -import json +import json as _json from enum import StrEnum from pathlib import Path @@ -14,6 +14,8 @@ from cgis.extractors.python_extractor import PythonExtractor, file_path_to_module_fqn from cgis.extractors.typescript_extractor import TypeScriptExtractor from cgis.pipeline import IngestionPipeline +from cgis.query.analyzer import AnalyzerEngine +from cgis.query.anomaly import AnomalyType, ArchitecturalAnomaly from cgis.query.engine import BEHAVIORAL_EDGE_TYPES, QueryEngine from cgis.query.mermaid import MermaidCompiler from cgis.resolver.uplift import SemanticUpliftEngine @@ -89,7 +91,7 @@ def _write_graph_output( output_path = Path(output) output_path.parent.mkdir(parents=True, exist_ok=True) with output_path.open("w", encoding="utf-8") as f: - json.dump(graph_data, f, indent=2) + _json.dump(graph_data, f, indent=2) else: with SQLiteStore(output) as store: store.save_graph(nodes, resolved_edges, overwrite=True) @@ -584,5 +586,84 @@ def _build_structure_tree( ) +_SEVERITY_COLOUR = { + AnomalyType.CIRCULAR_DEPENDENCY: "red", + AnomalyType.ZONE_OF_PAIN: "yellow", + AnomalyType.GOD_OBJECT: "magenta", +} + + +@app.command() +def analyze( + db: str = typer.Option(_DEFAULT_DB, "--db", "-d", help=_DEFAULT_DB_HELP), + min_severity: float = typer.Option( + 0.0, + "--min-severity", + "-s", + min=0.0, + max=1.0, + help="Only show anomalies at or above this severity score (0.0-1.0)", + ), + output_format: OutputFormat = typer.Option( + OutputFormat.TEXT, "--format", "-f", help="Output format: text or json" + ), +) -> None: + """ + Detect architectural anti-patterns in the ingested code graph. + + Runs three detectors: circular dependencies (Tarjan SCC), Zone of Pain + (Uncle Bob's instability / abstractness metrics), and God Objects. + Requires a .db graph file produced by the `ingest` command. + """ + path = Path(db) + if not path.is_file(): + console.print(f"[bold red]❌ Database not found:[/bold red] {db}. Run `ingest` first.") + raise typer.Exit(code=1) + + with SQLiteStore(db) as store: + report = AnalyzerEngine(store).run() + + visible = [a for a in report.anomalies if a.severity_score >= min_severity] + + if output_format == OutputFormat.MERMAID: + console.print("[bold red]❌ JSON/text only for analyze — mermaid not supported.[/bold red]") + raise typer.Exit(code=1) + + if output_format.value == "json": + typer.echo(_json.dumps([a.model_dump() for a in visible], indent=2)) + return + + console.print( + f"\n[bold blue]🔍 Architectural Health Report[/bold blue] " + f"[dim]{db}[/dim]\n" + f" Nodes analysed : [cyan]{report.total_nodes_analyzed}[/cyan]\n" + f" Anomalies found: [{'red' if report.total_anomalies else 'green'}]" + f"{report.total_anomalies}[/{'red' if report.total_anomalies else 'green'}]\n" + ) + + if not visible: + console.print("[bold green]✅ No anomalies above the severity threshold.[/bold green]") + return + + by_type: dict[AnomalyType, list[ArchitecturalAnomaly]] = {} + for a in visible: + by_type.setdefault(a.type, []).append(a) + + for anomaly_type, items in by_type.items(): + colour = _SEVERITY_COLOUR.get(anomaly_type, "white") + console.print(f"[bold {colour}]{'━' * 60}[/bold {colour}]") + console.print( + f"[bold {colour}]{anomaly_type.value}[/bold {colour}] ({len(items)} found)\n" + ) + for a in sorted(items, key=lambda x: x.severity_score, reverse=True): + console.print( + f" [yellow]{a.focal_fqn}[/yellow] " + f"severity=[bold {colour}]{a.severity_score:.2f}[/bold {colour}]" + ) + for key, val in a.metrics.items(): + console.print(f" [dim]{key}:[/dim] {val}") + console.print(f" [italic]💡 {a.refactoring_hint}[/italic]\n") + + if __name__ == "__main__": # pragma: no cover app() diff --git a/src/cgis/query/analyzer.py b/src/cgis/query/analyzer.py new file mode 100644 index 00000000..294961e5 --- /dev/null +++ b/src/cgis/query/analyzer.py @@ -0,0 +1,327 @@ +"""Architectural anti-pattern detector powered by graph traversal algorithms.""" + +import structlog + +from cgis.core.models import Edge, EdgeType, Node, NodeNamespace, NodeType +from cgis.query.anomaly import AnomalyType, ArchitecturalAnomaly, ArchitecturalReport +from cgis.storage.sqlite_store import SQLiteStore + +log = structlog.getLogger(__name__) + +# Thresholds -- tuned for typical Python codebases +_ZONE_OF_PAIN_DISTANCE = ( + 0.70 # D-distance threshold; Zone of Pain: D > 0.7 AND I <= 0.3 (stable + concrete) +) +_ZONE_OF_PAIN_MAX_INSTABILITY = 0.30 # I <= 0.3 means "stable" (more dependents than dependencies) +_ZONE_OF_PAIN_MIN_AFFERENT = 3 # require >= 3 callers to avoid flagging obscure low-usage classes +_GOD_OBJECT_MIN_METHODS = 10 # classes below this are too small to qualify regardless of coupling +_GOD_OBJECT_MIN_EFFERENT = 5 # low coupling means cohesion is still acceptable + + +def _build_adjacency(edges: list[Edge], allowed_types: frozenset[EdgeType]) -> dict[str, list[str]]: + """Build a directed adjacency list from edges matching allowed_types.""" + adj: dict[str, list[str]] = {} + for edge in edges: + if edge.type not in allowed_types: + continue + if edge.target.startswith(("raw_call:", "raw_import:")): + continue + adj.setdefault(edge.source, []).append(edge.target) + return adj + + +def _pop_scc(v: str, stack: list[str], on_stack: dict[str, bool]) -> list[str]: + """Pop nodes from stack until v is reached, forming one SCC.""" + scc: list[str] = [] + while True: + w = stack.pop() + on_stack[w] = False + scc.append(w) + if w == v: + break + return scc + + +def _tarjan_step( + v: str, + i: int, + adj: dict[str, list[str]], + index: dict[str, int], + lowlink: dict[str, int], + on_stack: dict[str, bool], + stack: list[str], + work: list[tuple[str, int]], + sccs: list[list[str]], + counter: list[int], +) -> None: + """Process one step of the iterative Tarjan walk: advance or backtrack.""" + neighbors = adj.get(v, []) + if i < len(neighbors): + work[-1] = (v, i + 1) + w = neighbors[i] + if w not in index: + index[w] = counter[0] + lowlink[w] = counter[0] + counter[0] += 1 + stack.append(w) + on_stack[w] = True + work.append((w, 0)) + elif on_stack.get(w, False): + lowlink[v] = min(lowlink[v], index[w]) + else: + work.pop() + if work: + lowlink[work[-1][0]] = min(lowlink[work[-1][0]], lowlink[v]) + if lowlink[v] == index[v]: + sccs.append(_pop_scc(v, stack, on_stack)) + + +def _tarjan_scc(adj: dict[str, list[str]]) -> list[list[str]]: + """Iterative Tarjan's SCC algorithm. + + Returns all strongly connected components. O(V+E), no recursion limit. + Any returned component with len > 1 is a cycle. + """ + index: dict[str, int] = {} + lowlink: dict[str, int] = {} + on_stack: dict[str, bool] = {} + stack: list[str] = [] + sccs: list[list[str]] = [] + counter = [0] + + for start in sorted(adj): + if start in index: + continue + index[start] = counter[0] + lowlink[start] = counter[0] + counter[0] += 1 + stack.append(start) + on_stack[start] = True + work: list[tuple[str, int]] = [(start, 0)] + while work: + v, i = work[-1] + _tarjan_step(v, i, adj, index, lowlink, on_stack, stack, work, sccs, counter) + + return sccs + + +def _build_method_to_class(nodes: list[Node], edges: list[Edge]) -> dict[str, str]: + """Return a mapping of method FQN -> parent class FQN.""" + method_ids = {n.id for n in nodes if n.type == NodeType.METHOD} + result: dict[str, str] = {} + for edge in edges: + if edge.type in (EdgeType.CONTAINS, EdgeType.DECLARES) and edge.target in method_ids: + result[edge.target] = edge.source + return result + + +def _compute_class_coupling( + internal_classes: set[str], + edges: list[Edge], + method_to_class: dict[str, str], +) -> tuple[dict[str, set[str]], dict[str, set[str]]]: + """Compute afferent (Ca) and efferent (Ce) coupling per internal class.""" + ca: dict[str, set[str]] = {fqn: set() for fqn in internal_classes} + ce: dict[str, set[str]] = {fqn: set() for fqn in internal_classes} + for edge in edges: + if edge.type != EdgeType.CALLS: + continue + src = method_to_class.get(edge.source, edge.source) + tgt = method_to_class.get(edge.target, edge.target) + if src == tgt: + continue + if tgt in internal_classes: + ca[tgt].add(src) + if src in internal_classes: + ce[src].add(tgt) + return ca, ce + + +def _get_abstract_fqns(internal_classes: set[str], nodes: list[Node]) -> set[str]: + """Return FQNs of internal classes that are abstract (metadata only). + + Only metadata-flagged classes qualify; using EXTENDS would misclassify + concrete subclasses (e.g. UserService(Base)) as abstract. + """ + return {n.id for n in nodes if n.id in internal_classes and n.metadata.get("is_abstract")} + + +def _build_class_method_map( + internal_classes: set[str], nodes: list[Node], edges: list[Edge] +) -> dict[str, set[str]]: + """Return a mapping of class FQN -> set of its method FQNs.""" + method_ids = {n.id for n in nodes if n.type == NodeType.METHOD} + result: dict[str, set[str]] = {fqn: set() for fqn in internal_classes} + for edge in edges: + if ( + edge.type in (EdgeType.CONTAINS, EdgeType.DECLARES) + and edge.source in internal_classes + and edge.target in method_ids + ): + result[edge.source].add(edge.target) + return result + + +class AnalyzerEngine: + """Detects architectural anti-patterns from a fully-ingested code graph.""" + + def __init__(self, store: SQLiteStore) -> None: + """Load all nodes and edges from the store for in-memory analysis.""" + self._nodes: list[Node] = store.get_all_nodes() + self._edges: list[Edge] = store.get_all_edges() + + def detect_cycles(self) -> list[ArchitecturalAnomaly]: + """Find circular dependencies using Tarjan's SCC on IMPORTS edges. + + Returns one anomaly per cycle, containing the full cycle as a list + in metrics["cycle_members"]. + """ + internal_fqns = { + n.id + for n in self._nodes + if n.namespace == NodeNamespace.INTERNAL and n.type in (NodeType.FILE, NodeType.MODULE) + } + adj = _build_adjacency(self._edges, frozenset({EdgeType.IMPORTS})) + adj = { + k: [v for v in vs if v in internal_fqns] for k, vs in adj.items() if k in internal_fqns + } + + anomalies: list[ArchitecturalAnomaly] = [] + for scc in _tarjan_scc(adj): + if len(scc) < 2: + continue + sorted_members = sorted(scc) + severity = min(1.0, (len(scc) - 1) / 4) + anomalies.append( + ArchitecturalAnomaly( + id="cycle_" + "_".join(sorted_members[:3]), + type=AnomalyType.CIRCULAR_DEPENDENCY, + focal_fqn=sorted_members[0], + severity_score=round(severity, 2), + metrics={"cycle_length": len(scc), "cycle_members": sorted_members}, + refactoring_hint=( + "Break the cycle with a Mediator, shared abstraction, or by " + "extracting the shared dependency into a third module." + ), + ) + ) + + log.info("Cycle detection complete.", cycles_found=len(anomalies)) + return anomalies + + def detect_zone_of_pain(self) -> list[ArchitecturalAnomaly]: + """Detect classes in Uncle Bob's Zone of Pain (stable + concrete). + + Uses afferent/efferent coupling from CALLS edges to compute instability I, + and node metadata to determine abstractness A. + High stability (I->0) + low abstractness (A=0) means D-distance approaches 1.0. + """ + internal_classes = { + n.id + for n in self._nodes + if n.type == NodeType.CLASS and n.namespace == NodeNamespace.INTERNAL + } + if not internal_classes: + return [] + + method_to_class = _build_method_to_class(self._nodes, self._edges) + ca, ce = _compute_class_coupling(internal_classes, self._edges, method_to_class) + abstract_fqns = _get_abstract_fqns(internal_classes, self._nodes) + + anomalies: list[ArchitecturalAnomaly] = [] + for fqn in internal_classes: + ca_count = len(ca[fqn]) + ce_count = len(ce[fqn]) + total = ca_count + ce_count + if total == 0 or ca_count < _ZONE_OF_PAIN_MIN_AFFERENT: + continue + instability = ce_count / total + abstractness = 1.0 if fqn in abstract_fqns else 0.0 + d_distance = abs(abstractness + instability - 1.0) + if ( + d_distance >= _ZONE_OF_PAIN_DISTANCE + and instability <= _ZONE_OF_PAIN_MAX_INSTABILITY + ): + anomalies.append( + ArchitecturalAnomaly( + id=f"zop_{fqn}", + type=AnomalyType.ZONE_OF_PAIN, + focal_fqn=fqn, + severity_score=round(d_distance, 2), + metrics={ + "afferent_coupling": ca_count, + "efferent_coupling": ce_count, + "instability": round(instability, 3), + "abstractness": abstractness, + "d_distance": round(d_distance, 3), + }, + refactoring_hint=( + "Introduce an Abstract Base Class or Protocol to invert the " + "dependency, making this module stable-and-abstract rather than " + "stable-and-concrete." + ), + ) + ) + + log.info("Zone of Pain detection complete.", found=len(anomalies)) + return anomalies + + def detect_god_objects(self) -> list[ArchitecturalAnomaly]: + """Detect God Objects: classes with too many methods and high efferent coupling.""" + internal_classes = { + n.id + for n in self._nodes + if n.type == NodeType.CLASS and n.namespace == NodeNamespace.INTERNAL + } + if not internal_classes: + return [] + + class_methods = _build_class_method_map(internal_classes, self._nodes, self._edges) + method_to_class = {m: cls for cls, methods in class_methods.items() for m in methods} + efferent: dict[str, set[str]] = {fqn: set() for fqn in internal_classes} + for edge in self._edges: + if edge.type != EdgeType.CALLS: + continue + src_class = method_to_class.get(edge.source) + if src_class is None: + continue + tgt_class = method_to_class.get(edge.target, edge.target) + if tgt_class != src_class: + efferent[src_class].add(tgt_class) + + anomalies: list[ArchitecturalAnomaly] = [] + for fqn in internal_classes: + m_count = len(class_methods[fqn]) + e_count = len(efferent[fqn]) + if m_count < _GOD_OBJECT_MIN_METHODS or e_count < _GOD_OBJECT_MIN_EFFERENT: + continue + severity = min( + 1.0, + (m_count / _GOD_OBJECT_MIN_METHODS) * (e_count / _GOD_OBJECT_MIN_EFFERENT) / 4, + ) + anomalies.append( + ArchitecturalAnomaly( + id=f"god_{fqn}", + type=AnomalyType.GOD_OBJECT, + focal_fqn=fqn, + severity_score=round(severity, 2), + metrics={"method_count": m_count, "efferent_coupling": e_count}, + refactoring_hint=( + "Apply Single Responsibility Principle: extract cohesive groups of " + "methods into focused collaborator classes." + ), + ) + ) + + log.info("God Object detection complete.", found=len(anomalies)) + return anomalies + + def run(self) -> ArchitecturalReport: + """Run all detectors and return a consolidated health report.""" + anomalies = self.detect_cycles() + self.detect_zone_of_pain() + self.detect_god_objects() + internal_count = sum(1 for n in self._nodes if n.namespace == NodeNamespace.INTERNAL) + return ArchitecturalReport( + total_nodes_analyzed=internal_count, + total_anomalies=len(anomalies), + anomalies=tuple(anomalies), + ) diff --git a/src/cgis/query/anomaly.py b/src/cgis/query/anomaly.py new file mode 100644 index 00000000..a69258c3 --- /dev/null +++ b/src/cgis/query/anomaly.py @@ -0,0 +1,34 @@ +"""Pydantic models for architectural anomaly reporting.""" + +from enum import StrEnum + +from pydantic import BaseModel, Field + + +class AnomalyType(StrEnum): + """Categories of architectural anti-patterns detectable from the code graph.""" + + CIRCULAR_DEPENDENCY = "CIRCULAR_DEPENDENCY" + ZONE_OF_PAIN = "ZONE_OF_PAIN" + GOD_OBJECT = "GOD_OBJECT" + + +class ArchitecturalAnomaly(BaseModel, frozen=True): + """A single detected architectural anti-pattern with supporting metrics.""" + + id: str = Field(..., description="Unique anomaly identifier (deterministic)") + type: AnomalyType + focal_fqn: str = Field(..., description="FQN of the primary node violating the pattern") + severity_score: float = Field(..., ge=0.0, le=1.0, description="Normalised severity 0-1") + metrics: dict[str, float | int | list[str]] = Field( + ..., description="Raw computed metrics supporting this finding" + ) + refactoring_hint: str = Field(..., description="Suggested untangling strategy") + + +class ArchitecturalReport(BaseModel, frozen=True): + """Full architectural health report for a codebase snapshot.""" + + total_nodes_analyzed: int + total_anomalies: int + anomalies: tuple[ArchitecturalAnomaly, ...] diff --git a/tests/unit/test_analyzer.py b/tests/unit/test_analyzer.py new file mode 100644 index 00000000..e14a2a10 --- /dev/null +++ b/tests/unit/test_analyzer.py @@ -0,0 +1,450 @@ +"""Unit tests for AnalyzerEngine: cycle detection, Zone of Pain, God Object.""" + +from collections.abc import Generator +from typing import Any + +import pytest + +from cgis.core.models import Edge, EdgeType, Node, NodeNamespace, NodeType +from cgis.query.analyzer import AnalyzerEngine, _tarjan_scc +from cgis.query.anomaly import AnomalyType +from cgis.storage.sqlite_store import SQLiteStore + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def store() -> Generator[SQLiteStore, None, None]: + """In-memory SQLite store, connected and disconnected per test.""" + s = SQLiteStore(":memory:") + s.connect() + yield s + s.disconnect() + + +def _file_node(fqn: str) -> Node: + """Helper: create an internal FILE node.""" + return Node( + id=fqn, + type=NodeType.FILE, + name=fqn.rsplit(".", maxsplit=1)[-1], + file_path=fqn.replace(".", "/") + ".py", + start_line=1, + end_line=1, + namespace=NodeNamespace.INTERNAL, + ) + + +def _class_node(fqn: str, **kwargs: Any) -> Node: # noqa: ANN401 + """Helper: create an internal CLASS node.""" + return Node( + id=fqn, + type=NodeType.CLASS, + name=fqn.rsplit(".", maxsplit=1)[-1], + file_path=fqn.split(".", maxsplit=1)[0] + ".py", + start_line=1, + end_line=50, + namespace=NodeNamespace.INTERNAL, + **kwargs, + ) + + +def _method_node(fqn: str) -> Node: + """Helper: create an internal METHOD node.""" + return Node( + id=fqn, + type=NodeType.METHOD, + name=fqn.rsplit(".", maxsplit=1)[-1], + file_path=fqn.split(".", maxsplit=1)[0] + ".py", + start_line=1, + end_line=5, + namespace=NodeNamespace.INTERNAL, + ) + + +def _imports(src: str, tgt: str) -> Edge: + """Helper: IMPORTS edge between two FQNs.""" + return Edge(id=f"{src}->imp->{tgt}", source=src, target=tgt, type=EdgeType.IMPORTS) + + +def _calls(src: str, tgt: str) -> Edge: + """Helper: CALLS edge between two FQNs.""" + return Edge(id=f"{src}->call->{tgt}", source=src, target=tgt, type=EdgeType.CALLS) + + +def _contains(parent: str, child: str) -> Edge: + """Helper: CONTAINS edge from parent to child.""" + return Edge(id=f"{parent}->has->{child}", source=parent, target=child, type=EdgeType.CONTAINS) + + +# --------------------------------------------------------------------------- +# _tarjan_scc (unit tests — no DB) +# --------------------------------------------------------------------------- + + +def test_tarjan_simple_cycle() -> None: + """A→B→A is detected as one 2-node SCC.""" + adj = {"A": ["B"], "B": ["A"]} + sccs = _tarjan_scc(adj) + cycles = [s for s in sccs if len(s) >= 2] + assert len(cycles) == 1 + assert set(cycles[0]) == {"A", "B"} + + +def test_tarjan_three_node_cycle() -> None: + """A→B→C→A is detected as one 3-node SCC.""" + adj = {"A": ["B"], "B": ["C"], "C": ["A"]} + sccs = _tarjan_scc(adj) + cycles = [s for s in sccs if len(s) >= 2] + assert len(cycles) == 1 + assert set(cycles[0]) == {"A", "B", "C"} + + +def test_tarjan_no_cycle() -> None: + """A→B→C (DAG) produces no SCCs with len > 1.""" + adj = {"A": ["B"], "B": ["C"], "C": []} + sccs = _tarjan_scc(adj) + assert all(len(s) == 1 for s in sccs) + + +def test_tarjan_two_separate_cycles() -> None: + """Two independent cycles are each detected separately.""" + adj = {"A": ["B"], "B": ["A"], "C": ["D"], "D": ["C"]} + sccs = _tarjan_scc(adj) + cycles = [s for s in sccs if len(s) >= 2] + assert len(cycles) == 2 + + +def test_tarjan_self_loop() -> None: + """A self-loop (A→A) is not reported as a multi-node SCC.""" + adj = {"A": ["A"]} + sccs = _tarjan_scc(adj) + cycles = [s for s in sccs if len(s) >= 2] + assert len(cycles) == 0 + + +def test_tarjan_empty_graph() -> None: + """Empty adjacency list produces no SCCs.""" + assert _tarjan_scc({}) == [] + + +# --------------------------------------------------------------------------- +# detect_cycles (integration with store) +# --------------------------------------------------------------------------- + + +def test_detect_cycles_two_node(store: SQLiteStore) -> None: + """Two files that mutually import each other produce one CIRCULAR_DEPENDENCY anomaly.""" + nodes = [_file_node("pkg.a"), _file_node("pkg.b")] + edges = [_imports("pkg.a", "pkg.b"), _imports("pkg.b", "pkg.a")] + store.save_graph(nodes, edges) + + report = AnalyzerEngine(store).detect_cycles() + + assert len(report) == 1 + anomaly = report[0] + assert anomaly.type == AnomalyType.CIRCULAR_DEPENDENCY + assert set(anomaly.metrics["cycle_members"]) == {"pkg.a", "pkg.b"} # type: ignore[arg-type] + assert anomaly.severity_score == pytest.approx(0.25) + + +def test_detect_cycles_three_node(store: SQLiteStore) -> None: + """A→B→C→A cycle is flagged with severity > two-node cycle.""" + nodes = [_file_node("a"), _file_node("b"), _file_node("c")] + edges = [_imports("a", "b"), _imports("b", "c"), _imports("c", "a")] + store.save_graph(nodes, edges) + + report = AnalyzerEngine(store).detect_cycles() + + assert len(report) == 1 + assert report[0].metrics["cycle_length"] == 3 + assert report[0].severity_score > 0.25 + + +def test_detect_cycles_no_cycle(store: SQLiteStore) -> None: + """Linear imports produce no anomalies.""" + nodes = [_file_node("x"), _file_node("y"), _file_node("z")] + edges = [_imports("x", "y"), _imports("y", "z")] + store.save_graph(nodes, edges) + + assert AnalyzerEngine(store).detect_cycles() == [] + + +def test_detect_cycles_ignores_external(store: SQLiteStore) -> None: + """Imports from external packages (non-INTERNAL nodes) are not flagged.""" + internal = _file_node("myapp.core") + external = Node( + id="requests", + type=NodeType.MODULE, + name="requests", + file_path="EXTERNAL", + start_line=0, + end_line=0, + namespace=NodeNamespace.EXTERNAL, + ) + edges = [_imports("myapp.core", "requests"), _imports("requests", "myapp.core")] + store.save_graph([internal, external], edges) + + assert AnalyzerEngine(store).detect_cycles() == [] + + +def test_detect_cycles_ignores_unresolved_imports(store: SQLiteStore) -> None: + """Unresolved (raw_import:) edges are skipped and never form a cycle.""" + nodes = [_file_node("a"), _file_node("b")] + edges = [ + Edge(id="a->raw->B", source="a", target="raw_import:B", type=EdgeType.IMPORTS), + _imports("b", "a"), + ] + store.save_graph(nodes, edges) + + assert AnalyzerEngine(store).detect_cycles() == [] + + +def test_detect_cycles_empty_graph(store: SQLiteStore) -> None: + """Empty graph produces no anomalies.""" + assert AnalyzerEngine(store).detect_cycles() == [] + + +# --------------------------------------------------------------------------- +# detect_zone_of_pain +# --------------------------------------------------------------------------- + + +def _seed_zone_of_pain(store: SQLiteStore, caller_count: int = 8) -> str: + """Seed a stable-concrete class called by many callers.""" + target = _class_node("app.UserService") + methods = [_method_node(f"app.UserService.m{i}") for i in range(3)] + callers = [_class_node(f"app.Controller{i}") for i in range(caller_count)] + caller_methods = [_method_node(f"app.Controller{i}.handle") for i in range(caller_count)] + + nodes = [target, *methods, *callers, *caller_methods] + edges: list[Edge] = [_contains("app.UserService", m.id) for m in methods] + edges += [ + e + for i, cm in enumerate(caller_methods) + for e in (_contains(f"app.Controller{i}", cm.id), _calls(cm.id, methods[0].id)) + ] + + store.save_graph(nodes, edges) + return "app.UserService" + + +def test_detect_zone_of_pain_flags_stable_concrete(store: SQLiteStore) -> None: + """A heavily-depended-upon concrete class is flagged as ZONE_OF_PAIN.""" + fqn = _seed_zone_of_pain(store, caller_count=8) + anomalies = AnalyzerEngine(store).detect_zone_of_pain() + + assert any(a.focal_fqn == fqn and a.type == AnomalyType.ZONE_OF_PAIN for a in anomalies) + + +def test_detect_zone_of_pain_abstract_class_not_flagged(store: SQLiteStore) -> None: + """A class flagged is_abstract in metadata is not flagged even with high Ca.""" + target = _class_node("app.BaseService", metadata={"is_abstract": True}) + methods = [_method_node(f"app.BaseService.m{i}") for i in range(3)] + callers = [_class_node(f"app.C{i}") for i in range(8)] + caller_methods = [_method_node(f"app.C{i}.h") for i in range(8)] + + nodes = [target, *methods, *callers, *caller_methods] + edges: list[Edge] = [ + *[_contains("app.BaseService", m.id) for m in methods], + *[ + e + for i, cm in enumerate(caller_methods) + for e in (_contains(f"app.C{i}", cm.id), _calls(cm.id, methods[0].id)) + ], + ] + + store.save_graph(nodes, edges) + anomalies = AnalyzerEngine(store).detect_zone_of_pain() + + assert not any(a.focal_fqn == "app.BaseService" for a in anomalies) + + +def test_detect_zone_of_pain_high_instability_not_flagged(store: SQLiteStore) -> None: + """A class above the instability threshold (I > 0.30) is not in the Zone of Pain.""" + # 8 callers + 4 callees → I = 4/12 ≈ 0.33, which is > _ZONE_OF_PAIN_MAX_INSTABILITY + target = _class_node("app.FlexService") + methods = [_method_node(f"app.FlexService.m{i}") for i in range(3)] + callers = [_class_node(f"app.Caller{i}") for i in range(8)] + caller_methods = [_method_node(f"app.Caller{i}.h") for i in range(8)] + callees = [_class_node(f"app.Dep{i}") for i in range(4)] + callee_methods = [_method_node(f"app.Dep{i}.op") for i in range(4)] + + nodes = [target, *methods, *callers, *caller_methods, *callees, *callee_methods] + edges: list[Edge] = [ + *[_contains("app.FlexService", m.id) for m in methods], + *[_contains(f"app.Dep{i}", cm.id) for i, cm in enumerate(callee_methods)], + *[_contains(f"app.Caller{i}", cm.id) for i, cm in enumerate(caller_methods)], + *[_calls(cm.id, methods[0].id) for cm in caller_methods], + *[_calls(methods[0].id, cm.id) for cm in callee_methods], + ] + store.save_graph(nodes, edges) + + assert not any( + a.focal_fqn == "app.FlexService" for a in AnalyzerEngine(store).detect_zone_of_pain() + ) + + +def test_detect_zone_of_pain_low_coupling_not_flagged(store: SQLiteStore) -> None: + """A class with only 1 caller is not in the Zone of Pain.""" + _seed_zone_of_pain(store, caller_count=1) + anomalies = AnalyzerEngine(store).detect_zone_of_pain() + + assert not any(a.focal_fqn == "app.UserService" for a in anomalies) + + +# --------------------------------------------------------------------------- +# detect_god_objects +# --------------------------------------------------------------------------- + + +def _seed_god_object( + store: SQLiteStore, + method_count: int = 12, + efferent_count: int = 6, +) -> str: + """Seed a class with many methods calling many distinct collaborators.""" + god = _class_node("app.GodClass") + methods = [_method_node(f"app.GodClass.m{i}") for i in range(method_count)] + collaborators = [_class_node(f"app.Collab{i}") for i in range(efferent_count)] + collab_methods = [_method_node(f"app.Collab{i}.op") for i in range(efferent_count)] + + nodes = [god, *methods, *collaborators, *collab_methods] + edges: list[Edge] = [ + *[_contains("app.GodClass", m.id) for m in methods], + *[_contains(f"app.Collab{i}", cm.id) for i, cm in enumerate(collab_methods)], + *[_calls(m.id, collab_methods[i].id) for i, m in enumerate(methods[:efferent_count])], + ] + + store.save_graph(nodes, edges) + return "app.GodClass" + + +def test_detect_god_object_flagged(store: SQLiteStore) -> None: + """Class with 12 methods and 6 efferent targets is a God Object.""" + fqn = _seed_god_object(store, method_count=12, efferent_count=6) + anomalies = AnalyzerEngine(store).detect_god_objects() + + assert any(a.focal_fqn == fqn and a.type == AnomalyType.GOD_OBJECT for a in anomalies) + + +def test_detect_god_object_small_class_not_flagged(store: SQLiteStore) -> None: + """Class with 3 methods is not a God Object regardless of coupling.""" + _seed_god_object(store, method_count=3, efferent_count=6) + anomalies = AnalyzerEngine(store).detect_god_objects() + + assert anomalies == [] + + +def test_detect_god_object_boundary_methods_not_flagged(store: SQLiteStore) -> None: + """Class with exactly 9 methods (one below threshold) is not flagged.""" + _seed_god_object(store, method_count=9, efferent_count=6) + anomalies = AnalyzerEngine(store).detect_god_objects() + + assert anomalies == [] + + +def test_detect_god_object_low_efferent_not_flagged(store: SQLiteStore) -> None: + """Class with many methods but low efferent coupling is not flagged.""" + _seed_god_object(store, method_count=15, efferent_count=2) + anomalies = AnalyzerEngine(store).detect_god_objects() + + assert anomalies == [] + + +def test_detect_god_object_self_calls_not_counted(store: SQLiteStore) -> None: + """Methods calling sibling methods in the same class do not inflate efferent coupling.""" + god = _class_node("app.GodClass") + methods = [_method_node(f"app.GodClass.m{i}") for i in range(12)] + nodes = [god, *methods] + edges: list[Edge] = [ + *[_contains("app.GodClass", m.id) for m in methods], + *[_calls(methods[i].id, methods[(i + 1) % 12].id) for i in range(12)], + ] + store.save_graph(nodes, edges) + + assert AnalyzerEngine(store).detect_god_objects() == [] + + +def test_detect_god_object_unresolved_calls_count_as_efferent(store: SQLiteStore) -> None: + """raw_call: targets count as distinct efferent targets (external dependencies).""" + god = _class_node("app.GodClass") + methods = [_method_node(f"app.GodClass.m{i}") for i in range(12)] + nodes = [god, *methods] + edges: list[Edge] = [ + *[_contains("app.GodClass", m.id) for m in methods], + *[ + Edge( + id=f"raw_{i}", + source=methods[i].id, + target=f"raw_call:ext_fn_{i}", + type=EdgeType.CALLS, + ) + for i in range(6) + ], + ] + store.save_graph(nodes, edges) + + anomalies = AnalyzerEngine(store).detect_god_objects() + assert any(a.focal_fqn == "app.GodClass" for a in anomalies) + + +def test_detect_god_object_duplicate_raw_calls_deduplicated(store: SQLiteStore) -> None: + """Multiple calls to the same raw_call: target count as one efferent target (set dedup).""" + god = _class_node("app.GodClass") + methods = [_method_node(f"app.GodClass.m{i}") for i in range(12)] + nodes = [god, *methods] + edges: list[Edge] = [ + *[_contains("app.GodClass", m.id) for m in methods], + *[ + Edge(id=f"raw_{i}", source=m.id, target="raw_call:same_fn", type=EdgeType.CALLS) + for i, m in enumerate(methods) + ], + ] + store.save_graph(nodes, edges) + + anomalies = AnalyzerEngine(store).detect_god_objects() + assert anomalies == [] # efferent == 1 (same target deduped) < _GOD_OBJECT_MIN_EFFERENT + + +def test_detect_zone_of_pain_no_class_nodes(store: SQLiteStore) -> None: + """Graph with only FILE nodes (no CLASS nodes) returns no Zone of Pain anomalies.""" + store.save_graph([_file_node("app.main"), _file_node("app.utils")], []) + + assert AnalyzerEngine(store).detect_zone_of_pain() == [] + + +def test_detect_god_objects_no_class_nodes(store: SQLiteStore) -> None: + """Graph with only FILE nodes (no CLASS nodes) returns no God Object anomalies.""" + store.save_graph([_file_node("app.main")], []) + + assert AnalyzerEngine(store).detect_god_objects() == [] + + +# --------------------------------------------------------------------------- +# run (full report) +# --------------------------------------------------------------------------- + + +def test_run_empty_graph(store: SQLiteStore) -> None: + """Empty graph produces a zero-anomaly report.""" + report = AnalyzerEngine(store).run() + + assert report.total_anomalies == 0 + assert report.anomalies == () + + +def test_run_returns_all_detectors(store: SQLiteStore) -> None: + """run() aggregates anomalies from all three detectors.""" + # Seed a cycle + store.save_graph( + [_file_node("a"), _file_node("b")], + [_imports("a", "b"), _imports("b", "a")], + ) + report = AnalyzerEngine(store).run() + + assert report.total_anomalies >= 1 + types = {a.type for a in report.anomalies} + assert AnomalyType.CIRCULAR_DEPENDENCY in types