diff --git a/tools/spectral/.gitignore b/tools/spectral/.gitignore new file mode 100644 index 0000000..60c9b1c --- /dev/null +++ b/tools/spectral/.gitignore @@ -0,0 +1,2 @@ +__pycache__ +output/* \ No newline at end of file diff --git a/tools/spectral/spectral_analysis.py b/tools/spectral/spectral_analysis.py new file mode 100644 index 0000000..5b4b869 --- /dev/null +++ b/tools/spectral/spectral_analysis.py @@ -0,0 +1,1463 @@ +#!/usr/bin/env python3 +"""Spectral analysis tool for dependency DAGs. + +Reads a GraphViz DOT file (produced by the depgraph tool) and applies spectral +graph theory (Laplacian eigenvalues, Fiedler vectors) to derive quantitative +complexity metrics and visual analysis of codebase structural coupling. + +Usage: + python spectral_analysis.py deps.dot [-o OUTPUT_DIR] [--no-plots] [--json] +""" + +from __future__ import annotations + +import argparse +import json +import math +import os +import re +import sys +from dataclasses import dataclass, field +from typing import Any + +import numpy as np +from scipy import sparse + + +# ─── Data Structures ────────────────────────────────────────────────────────── + +@dataclass +class Node: + name: str + module: str + + +@dataclass +class Edge: + source: str + target: str + label: str + edge_type: str # "field" or "trait_impl" + cross_module: bool + + +@dataclass +class DependencyGraph: + nodes: list[Node] = field(default_factory=list) + edges: list[Edge] = field(default_factory=list) + modules: list[str] = field(default_factory=list) # ordered module names + node_to_module: dict[str, str] = field(default_factory=dict) + + +@dataclass +class SpectralResults: + eigenvalues: np.ndarray + eigenvectors: np.ndarray + fiedler_value: float + fiedler_vector: np.ndarray + adjacency: np.ndarray + adjacency_sym: np.ndarray + laplacian: np.ndarray + node_names: list[str] + node_modules: list[str] + + +@dataclass +class ModuleCouplingResult: + module_names: list[str] + coupling_matrix: np.ndarray # directed + cross_module_edges: int + total_edges: int + + +@dataclass +class ComplexityMetrics: + algebraic_connectivity: float + normalized_algebraic_connectivity: float + spectral_entropy: float + normalized_spectral_entropy: float + edge_density: float + cross_module_ratio: float + spectral_radius: float + normalized_spectral_radius: float + cci: float + n_nodes: int + n_edges: int + n_modules: int + connected_components: int + + +# ─── DOT Parser ─────────────────────────────────────────────────────────────── + +def parse_dot(text: str) -> DependencyGraph: + """Parse a depgraph-generated DOT file into a DependencyGraph. + + Uses a line-by-line state machine to extract: + - subgraph cluster_ blocks -> nodes with module membership + - A -> B [label="...", style=..., ...] -> edges with classification + """ + graph = DependencyGraph() + current_module: str | None = None + module_order: list[str] = [] + seen_nodes: set[str] = set() + + for line in text.splitlines(): + stripped = line.strip() + + # Entering a subgraph cluster + m = re.match(r'subgraph\s+cluster_(\w+)\s*\{', stripped) + if m: + current_module = m.group(1) + if current_module not in module_order: + module_order.append(current_module) + continue + + # Closing brace - exit current subgraph if we're in one + if stripped == '}' and current_module is not None: + current_module = None + continue + + # Node definition inside a subgraph: NodeName [label="...", ...] + if current_module is not None: + node_match = re.match(r'(\w+)\s*\[', stripped) + if node_match: + node_name = node_match.group(1) + # Skip DOT keywords + if node_name in ('label', 'style', 'node', 'edge', 'graph', + 'subgraph', 'digraph', 'rankdir', 'fontname', + 'fontsize', 'labelloc', 'compound', 'newrank', + 'splines', 'fillcolor', 'color'): + continue + if node_name not in seen_nodes: + seen_nodes.add(node_name) + graph.nodes.append(Node(name=node_name, module=current_module)) + graph.node_to_module[node_name] = current_module + continue + + # Edge definition: A -> B [label="...", style=..., ...] + edge_match = re.match( + r'(\w+)\s*->\s*(\w+)\s*\[(.+)\];', stripped + ) + if edge_match: + src = edge_match.group(1) + tgt = edge_match.group(2) + attrs_str = edge_match.group(3) + + # Extract label + label_match = re.search(r'label="([^"]*)"', attrs_str) + label = label_match.group(1) if label_match else "" + + # Classify edge type + style_match = re.search(r'style=(\w+)', attrs_str) + style = style_match.group(1) if style_match else "solid" + edge_type = "trait_impl" if style == "dotted" else "field" + + # Determine cross-module status + src_mod = graph.node_to_module.get(src) + tgt_mod = graph.node_to_module.get(tgt) + cross = src_mod is not None and tgt_mod is not None and src_mod != tgt_mod + + graph.edges.append(Edge( + source=src, target=tgt, label=label, + edge_type=edge_type, cross_module=cross, + )) + continue + + graph.modules = module_order + return graph + + +# ─── Matrix Construction ────────────────────────────────────────────────────── + +def get_node_ordering(graph: DependencyGraph) -> list[str]: + """Order nodes by module order, then alphabetical within module.""" + module_index = {m: i for i, m in enumerate(graph.modules)} + return sorted( + [n.name for n in graph.nodes], + key=lambda name: ( + module_index.get(graph.node_to_module.get(name, ""), 999), + name, + ), + ) + + +def build_adjacency(graph: DependencyGraph, node_order: list[str]) -> np.ndarray: + """Build directed binary adjacency matrix.""" + n = len(node_order) + idx = {name: i for i, name in enumerate(node_order)} + A = np.zeros((n, n), dtype=float) + for edge in graph.edges: + i = idx.get(edge.source) + j = idx.get(edge.target) + if i is not None and j is not None: + A[i, j] = 1.0 + return A + + +def symmetrize(A: np.ndarray) -> np.ndarray: + """OR-symmetrize: A_sym[i,j] = 1 if A[i,j] or A[j,i].""" + return np.clip(A + A.T, 0, 1) + + +def build_laplacian(A_sym: np.ndarray) -> np.ndarray: + """Build graph Laplacian L = D - A_sym.""" + D = np.diag(A_sym.sum(axis=1)) + return D - A_sym + + +# ─── Spectral Analysis ──────────────────────────────────────────────────────── + +def compute_spectral(graph: DependencyGraph) -> SpectralResults: + """Compute full spectral analysis of the dependency graph.""" + node_order = get_node_ordering(graph) + n = len(node_order) + + A = build_adjacency(graph, node_order) + A_sym = symmetrize(A) + L = build_laplacian(A_sym) + + if n == 0: + return SpectralResults( + eigenvalues=np.array([]), + eigenvectors=np.array([[]]), + fiedler_value=0.0, + fiedler_vector=np.array([]), + adjacency=A, adjacency_sym=A_sym, laplacian=L, + node_names=node_order, + node_modules=[graph.node_to_module.get(name, "") for name in node_order], + ) + + eigenvalues, eigenvectors = np.linalg.eigh(L) + + # Clean up near-zero eigenvalues + eigenvalues = np.where(np.abs(eigenvalues) < 1e-10, 0.0, eigenvalues) + + if n == 1: + fiedler_value = 0.0 + fiedler_vector = np.array([0.0]) + elif n >= 2: + fiedler_value = float(eigenvalues[1]) + fiedler_vector = eigenvectors[:, 1] + else: + fiedler_value = 0.0 + fiedler_vector = np.array([]) + + return SpectralResults( + eigenvalues=eigenvalues, + eigenvectors=eigenvectors, + fiedler_value=fiedler_value, + fiedler_vector=fiedler_vector, + adjacency=A, + adjacency_sym=A_sym, + laplacian=L, + node_names=node_order, + node_modules=[graph.node_to_module.get(name, "") for name in node_order], + ) + + +# ─── Module Coupling ────────────────────────────────────────────────────────── + +def compute_module_coupling(graph: DependencyGraph) -> ModuleCouplingResult: + """Compute directed module-level coupling matrix.""" + modules = graph.modules + n = len(modules) + mod_idx = {m: i for i, m in enumerate(modules)} + M = np.zeros((n, n), dtype=float) + + cross = 0 + total = len(graph.edges) + + for edge in graph.edges: + src_mod = graph.node_to_module.get(edge.source) + tgt_mod = graph.node_to_module.get(edge.target) + if src_mod is not None and tgt_mod is not None: + i = mod_idx.get(src_mod) + j = mod_idx.get(tgt_mod) + if i is not None and j is not None: + M[i, j] += 1.0 + if src_mod != tgt_mod: + cross += 1 + + return ModuleCouplingResult( + module_names=modules, + coupling_matrix=M, + cross_module_edges=cross, + total_edges=total, + ) + + +# ─── Complexity Metrics ─────────────────────────────────────────────────────── + +def count_connected_components(A_sym: np.ndarray) -> int: + """Count connected components using BFS on the symmetrized adjacency.""" + n = A_sym.shape[0] + if n == 0: + return 0 + visited = set() + components = 0 + for start in range(n): + if start in visited: + continue + components += 1 + queue = [start] + visited.add(start) + while queue: + node = queue.pop(0) + for neighbor in range(n): + if A_sym[node, neighbor] > 0 and neighbor not in visited: + visited.add(neighbor) + queue.append(neighbor) + return components + + +def compute_spectral_entropy(eigenvalues: np.ndarray) -> float: + """Compute spectral entropy from positive Laplacian eigenvalues. + + H(lambda) = -sum(p_i * log2(p_i)) where p_i = lambda_i / sum(lambdas) + over positive eigenvalues. + """ + positive = eigenvalues[eigenvalues > 1e-10] + if len(positive) == 0: + return 0.0 + p = positive / positive.sum() + # Avoid log(0) + p = p[p > 0] + return float(-np.sum(p * np.log2(p))) + + +def compute_complexity_metrics( + spectral: SpectralResults, + coupling: ModuleCouplingResult, +) -> ComplexityMetrics: + """Compute the Connectome Complexity Index (CCI) and all sub-metrics.""" + n = len(spectral.node_names) + n_edges = int(spectral.adjacency.sum()) # directed edge count + n_modules = len(coupling.module_names) + components = count_connected_components(spectral.adjacency_sym) + + if n <= 1: + return ComplexityMetrics( + algebraic_connectivity=0.0, + normalized_algebraic_connectivity=0.0, + spectral_entropy=0.0, + normalized_spectral_entropy=0.0, + edge_density=0.0, + cross_module_ratio=0.0, + spectral_radius=0.0, + normalized_spectral_radius=0.0, + cci=0.0, + n_nodes=n, + n_edges=n_edges, + n_modules=n_modules, + connected_components=components, + ) + + # Sub-metric 1: Normalized algebraic connectivity (lambda_2 / n) + algebraic_connectivity = spectral.fiedler_value + norm_alg_conn = algebraic_connectivity / n + + # Sub-metric 2: Spectral entropy + raw_entropy = compute_spectral_entropy(spectral.eigenvalues) + positive_count = int(np.sum(spectral.eigenvalues > 1e-10)) + max_entropy = math.log2(positive_count) if positive_count > 1 else 1.0 + norm_entropy = raw_entropy / max_entropy if max_entropy > 0 else 0.0 + + # Sub-metric 3: Edge density |E| / (n*(n-1)) + edge_density = n_edges / (n * (n - 1)) if n > 1 else 0.0 + + # Sub-metric 4: Cross-module coupling ratio + cross_ratio = (coupling.cross_module_edges / coupling.total_edges + if coupling.total_edges > 0 else 0.0) + + # Sub-metric 5: Normalized spectral radius (max eigenvalue of A_sym / (n-1)) + if spectral.adjacency_sym.shape[0] > 0: + eig_A = np.linalg.eigvalsh(spectral.adjacency_sym) + spectral_radius = float(np.max(np.abs(eig_A))) + else: + spectral_radius = 0.0 + norm_spec_radius = spectral_radius / (n - 1) if n > 1 else 0.0 + + # CCI = weighted sum + cci = ( + 0.25 * norm_alg_conn + + 0.25 * norm_entropy + + 0.15 * edge_density + + 0.20 * cross_ratio + + 0.15 * norm_spec_radius + ) + + return ComplexityMetrics( + algebraic_connectivity=algebraic_connectivity, + normalized_algebraic_connectivity=norm_alg_conn, + spectral_entropy=raw_entropy, + normalized_spectral_entropy=norm_entropy, + edge_density=edge_density, + cross_module_ratio=cross_ratio, + spectral_radius=spectral_radius, + normalized_spectral_radius=norm_spec_radius, + cci=cci, + n_nodes=n, + n_edges=n_edges, + n_modules=n_modules, + connected_components=components, + ) + + +# ─── Full Pipeline ──────────────────────────────────────────────────────────── + +@dataclass +class AnalysisResult: + graph: DependencyGraph + spectral: SpectralResults + coupling: ModuleCouplingResult + metrics: ComplexityMetrics + + +def run_analysis(graph: DependencyGraph) -> AnalysisResult: + """Run the full spectral analysis pipeline on a DependencyGraph.""" + spectral = compute_spectral(graph) + coupling = compute_module_coupling(graph) + metrics = compute_complexity_metrics(spectral, coupling) + return AnalysisResult( + graph=graph, + spectral=spectral, + coupling=coupling, + metrics=metrics, + ) + + +# ─── Text Report ────────────────────────────────────────────────────────────── + +def generate_report(result: AnalysisResult) -> str: + """Generate a text report of the spectral analysis.""" + s = result.spectral + m = result.metrics + c = result.coupling + lines: list[str] = [] + + def w(text: str = "") -> None: + lines.append(text) + + w("=" * 72) + w(" SPECTRAL ANALYSIS REPORT — Dependency DAG") + w("=" * 72) + w() + + # Graph summary + w("GRAPH SUMMARY") + w("-" * 40) + w(f" Nodes: {m.n_nodes}") + w(f" Directed edges: {m.n_edges}") + w(f" Modules: {m.n_modules}") + w(f" Connected components: {m.connected_components}") + w(f" Modules: {', '.join(c.module_names)}") + w() + + # Eigenvalue spectrum + w("LAPLACIAN EIGENVALUE SPECTRUM") + w("-" * 40) + for i, ev in enumerate(s.eigenvalues): + marker = " <-- Fiedler value (lambda_2)" if i == 1 else "" + w(f" lambda_{i:2d} = {ev:8.4f}{marker}") + w() + if len(s.eigenvalues) > 1: + spectral_gap = float(s.eigenvalues[-1] - s.eigenvalues[1]) + w(f" Spectral gap (lambda_max - lambda_2): {spectral_gap:.4f}") + w(f" Fiedler value (algebraic connectivity): {s.fiedler_value:.4f}") + w() + + # Fiedler vector analysis + if len(s.fiedler_vector) > 0: + w("FIEDLER VECTOR — SPECTRAL BISECTION") + w("-" * 40) + # Sort by fiedler value + indices = np.argsort(s.fiedler_vector) + w(" Partition A (Fiedler < 0):") + for idx in indices: + if s.fiedler_vector[idx] < 0: + w(f" {s.node_names[idx]:25s} [{s.node_modules[idx]:12s}] " + f"f = {s.fiedler_vector[idx]:+.4f}") + w(" ────────────────────────────────────") + w(" Partition B (Fiedler >= 0):") + for idx in indices: + if s.fiedler_vector[idx] >= 0: + w(f" {s.node_names[idx]:25s} [{s.node_modules[idx]:12s}] " + f"f = {s.fiedler_vector[idx]:+.4f}") + w() + + # Module coupling + w("MODULE COUPLING MATRIX (directed edge counts)") + w("-" * 40) + header = " " + " " * 14 + "".join(f"{name:>10s}" for name in c.module_names) + w(header) + for i, row_name in enumerate(c.module_names): + row = f" {row_name:12s} " + "".join( + f"{int(c.coupling_matrix[i, j]):10d}" for j in range(len(c.module_names)) + ) + w(row) + w() + w(f" Cross-module edges: {c.cross_module_edges} / {c.total_edges} " + f"({m.cross_module_ratio:.1%})") + w() + + # Complexity metrics + w("CONNECTOME COMPLEXITY INDEX (CCI)") + w("-" * 40) + w(f" {'Sub-metric':<40s} {'Raw':>10s} {'Normalized':>10s} {'Weight':>8s} {'Contrib':>8s}") + w(f" {'─' * 40} {'─' * 10} {'─' * 10} {'─' * 8} {'─' * 8}") + + rows = [ + ("Algebraic connectivity (lambda_2/n)", + f"{m.algebraic_connectivity:.4f}", f"{m.normalized_algebraic_connectivity:.4f}", + "0.25", f"{0.25 * m.normalized_algebraic_connectivity:.4f}"), + ("Spectral entropy (H/log2(k))", + f"{m.spectral_entropy:.4f}", f"{m.normalized_spectral_entropy:.4f}", + "0.25", f"{0.25 * m.normalized_spectral_entropy:.4f}"), + ("Edge density (|E|/n(n-1))", + f"{m.edge_density:.4f}", f"{m.edge_density:.4f}", + "0.15", f"{0.15 * m.edge_density:.4f}"), + ("Cross-module coupling ratio", + f"{m.cross_module_ratio:.4f}", f"{m.cross_module_ratio:.4f}", + "0.20", f"{0.20 * m.cross_module_ratio:.4f}"), + ("Spectral radius (rho/(n-1))", + f"{m.spectral_radius:.4f}", f"{m.normalized_spectral_radius:.4f}", + "0.15", f"{0.15 * m.normalized_spectral_radius:.4f}"), + ] + for label, raw, norm, weight, contrib in rows: + w(f" {label:<40s} {raw:>10s} {norm:>10s} {weight:>8s} {contrib:>8s}") + w(f" {'─' * 40} {'─' * 10} {'─' * 10} {'─' * 8} {'─' * 8}") + w(f" {'CCI (weighted sum)':<40s} {'':>10s} {'':>10s} {'1.00':>8s} {m.cci:8.4f}") + w() + + # Interpretation + if m.cci < 0.3: + interp = "LOW complexity — well-decomposed architecture" + elif m.cci < 0.6: + interp = "MODERATE complexity — typical well-structured codebase" + else: + interp = "HIGH complexity — consider reviewing module boundaries" + w(f" Interpretation: {interp}") + w() + w("=" * 72) + + return "\n".join(lines) + + +# ─── Dashboard Visualization ───────────────────────────────────────────────── + +# Module colors matching the depgraph tool +MODULE_COLORS = { + "error": "#4caf50", + "config": "#8bc34a", + "channel": "#ffeb3b", + "actor": "#2196f3", + "address_map": "#9c27b0", + "runtime": "#f44336", + "worker": "#ff9800", + "python": "#795548", +} + +DEFAULT_COLOR = "#9e9e9e" + + +def get_module_color(module: str) -> str: + return MODULE_COLORS.get(module, DEFAULT_COLOR) + + +def generate_dashboard(result: AnalysisResult, output_path: str) -> None: + """Generate spectral dashboard PNG (16x12, 150 DPI, dark theme).""" + import matplotlib + matplotlib.use("Agg") + import matplotlib.pyplot as plt + from matplotlib.gridspec import GridSpec + + s = result.spectral + m = result.metrics + c = result.coupling + + # Dark theme + plt.rcParams.update({ + "figure.facecolor": "#1a1a2e", + "axes.facecolor": "#16213e", + "axes.edgecolor": "#e0e0e0", + "axes.labelcolor": "#e0e0e0", + "text.color": "#e0e0e0", + "xtick.color": "#e0e0e0", + "ytick.color": "#e0e0e0", + "grid.color": "#2a2a4a", + "grid.alpha": 0.5, + }) + + fig = plt.figure(figsize=(16, 12), dpi=150) + gs = GridSpec(2, 2, figure=fig, hspace=0.35, wspace=0.3, + left=0.07, right=0.95, top=0.92, bottom=0.06) + + fig.suptitle("Spectral Analysis Dashboard — Dependency DAG", + fontsize=16, fontweight="bold", color="#e0e0e0") + + # ── Top-left: Eigenvalue spectrum ── + ax1 = fig.add_subplot(gs[0, 0]) + n = len(s.eigenvalues) + colors_eig = ["#ff4444" if i == 1 else "#4fc3f7" for i in range(n)] + markerline, stemlines, baseline = ax1.stem( + range(n), s.eigenvalues, linefmt="-", markerfmt="o", basefmt=" " + ) + markerline.set_color("#4fc3f7") + markerline.set_markersize(5) + stemlines.set_color("#4fc3f7") + stemlines.set_alpha(0.6) + # Highlight lambda_2 + if n > 1: + ax1.plot(1, s.eigenvalues[1], "o", color="#ff4444", markersize=10, + zorder=5, label=f"$\\lambda_2$ = {s.fiedler_value:.4f}") + ax1.legend(fontsize=10, loc="upper left", + facecolor="#16213e", edgecolor="#444") + ax1.set_xlabel("Index") + ax1.set_ylabel("Eigenvalue") + ax1.set_title("Laplacian Eigenvalue Spectrum", fontsize=12, fontweight="bold") + ax1.grid(True, alpha=0.3) + + # ── Top-right: Fiedler vector ── + ax2 = fig.add_subplot(gs[0, 1]) + if len(s.fiedler_vector) > 0: + sorted_indices = np.argsort(s.fiedler_vector) + sorted_values = s.fiedler_vector[sorted_indices] + sorted_names = [s.node_names[i] for i in sorted_indices] + sorted_modules = [s.node_modules[i] for i in sorted_indices] + bar_colors = [get_module_color(mod) for mod in sorted_modules] + + bars = ax2.barh(range(len(sorted_values)), sorted_values, + color=bar_colors, edgecolor="none", height=0.8) + ax2.axvline(x=0, color="#ff4444", linewidth=1.5, linestyle="--", + alpha=0.8, label="Bisection boundary") + ax2.set_yticks(range(len(sorted_names))) + ax2.set_yticklabels(sorted_names, fontsize=6) + ax2.set_xlabel("Fiedler value") + ax2.set_title("Fiedler Vector (spectral bisection)", fontsize=12, + fontweight="bold") + + # Legend for modules + unique_modules = [] + seen = set() + for mod in sorted_modules: + if mod not in seen: + seen.add(mod) + unique_modules.append(mod) + from matplotlib.patches import Patch + legend_patches = [Patch(facecolor=get_module_color(mod), label=mod) + for mod in unique_modules] + ax2.legend(handles=legend_patches, fontsize=7, loc="lower right", + facecolor="#16213e", edgecolor="#444", ncol=2) + else: + ax2.text(0.5, 0.5, "No Fiedler vector\n(single node graph)", + ha="center", va="center", fontsize=14, transform=ax2.transAxes) + ax2.set_title("Fiedler Vector", fontsize=12, fontweight="bold") + + # ── Bottom-left: Module coupling heatmap ── + ax3 = fig.add_subplot(gs[1, 0]) + if len(c.module_names) > 0: + im = ax3.imshow(c.coupling_matrix, cmap="YlOrRd", aspect="auto") + ax3.set_xticks(range(len(c.module_names))) + ax3.set_xticklabels(c.module_names, rotation=45, ha="right", fontsize=8) + ax3.set_yticks(range(len(c.module_names))) + ax3.set_yticklabels(c.module_names, fontsize=8) + ax3.set_title("Module Coupling (directed edge counts)", fontsize=12, + fontweight="bold") + ax3.set_xlabel("Target module") + ax3.set_ylabel("Source module") + + # Annotate cells + for i in range(len(c.module_names)): + for j in range(len(c.module_names)): + val = int(c.coupling_matrix[i, j]) + if val > 0: + text_color = "white" if val > c.coupling_matrix.max() * 0.6 else "black" + ax3.text(j, i, str(val), ha="center", va="center", + fontsize=8, color=text_color, fontweight="bold") + + plt.colorbar(im, ax=ax3, shrink=0.8) + else: + ax3.text(0.5, 0.5, "No modules", ha="center", va="center", + fontsize=14, transform=ax3.transAxes) + ax3.set_title("Module Coupling", fontsize=12, fontweight="bold") + + # ── Bottom-right: Metrics panel ── + ax4 = fig.add_subplot(gs[1, 1]) + ax4.axis("off") + + # CCI interpretation + if m.cci < 0.3: + cci_color = "#4caf50" + cci_label = "LOW" + elif m.cci < 0.6: + cci_color = "#ff9800" + cci_label = "MODERATE" + else: + cci_color = "#f44336" + cci_label = "HIGH" + + text_lines = [ + ("GRAPH", "", False), + (f" Nodes: {m.n_nodes} Edges: {m.n_edges} " + f"Modules: {m.n_modules} Components: {m.connected_components}", "", False), + ("", "", False), + ("SPECTRAL METRICS", "", False), + (f" Algebraic connectivity (lambda_2): {m.algebraic_connectivity:.4f}", "", False), + (f" Normalized (lambda_2/n): {m.normalized_algebraic_connectivity:.4f}", "", False), + (f" Spectral entropy: {m.spectral_entropy:.4f}", "", False), + (f" Normalized entropy: {m.normalized_spectral_entropy:.4f}", "", False), + (f" Spectral radius: {m.spectral_radius:.4f}", "", False), + (f" Normalized radius: {m.normalized_spectral_radius:.4f}", "", False), + ("", "", False), + ("COUPLING METRICS", "", False), + (f" Edge density: {m.edge_density:.4f}", "", False), + (f" Cross-module ratio: {m.cross_module_ratio:.1%}", "", False), + ("", "", False), + (f" CCI = {m.cci:.4f} [{cci_label}]", cci_color, True), + ] + + y = 0.95 + for text, color, bold in text_lines: + if not text: + y -= 0.04 + continue + fontsize = 11 if bold else 9 + weight = "bold" if bold else "normal" + c_val = color if color else "#e0e0e0" + ax4.text(0.05, y, text, transform=ax4.transAxes, fontsize=fontsize, + fontweight=weight, color=c_val, fontfamily="monospace", + verticalalignment="top") + y -= 0.055 + + ax4.set_title("Complexity Metrics", fontsize=12, fontweight="bold") + + plt.savefig(output_path, dpi=150, facecolor=fig.get_facecolor(), + edgecolor="none", bbox_inches="tight") + plt.close(fig) + + +# ─── Interactive HTML Dashboard ─────────────────────────────────────────────── + +def generate_dashboard_html( + result: AnalysisResult, output_path: str, *, dot_source: str = "" +) -> None: + """Generate an interactive HTML dashboard with GraphViz DAG + spectral panels.""" + s = result.spectral + m = result.metrics + c = result.coupling + + # Prepare data as JSON for embedding + sorted_indices = list(np.argsort(s.fiedler_vector)) if len(s.fiedler_vector) > 0 else [] + fiedler_data = [] + for idx in sorted_indices: + fiedler_data.append({ + "name": s.node_names[idx], + "module": s.node_modules[idx], + "value": float(s.fiedler_vector[idx]), + }) + + eigenvalue_data = [{"index": i, "value": float(v)} + for i, v in enumerate(s.eigenvalues)] + + coupling_data = { + "modules": c.module_names, + "matrix": c.coupling_matrix.tolist(), + } + + # Module colors + all_modules = list(dict.fromkeys(n.module for n in result.graph.nodes)) + module_colors_json = {mod: get_module_color(mod) for mod in all_modules} + + # CCI interpretation + if m.cci < 0.3: + cci_color = "#4caf50" + cci_label = "LOW" + cci_desc = "well-decomposed architecture" + elif m.cci < 0.6: + cci_color = "#ff9800" + cci_label = "MODERATE" + cci_desc = "typical well-structured codebase" + else: + cci_color = "#f44336" + cci_label = "HIGH" + cci_desc = "consider reviewing module boundaries" + + metrics_json = { + "n_nodes": m.n_nodes, + "n_edges": m.n_edges, + "n_modules": m.n_modules, + "connected_components": m.connected_components, + "algebraic_connectivity": round(m.algebraic_connectivity, 4), + "normalized_algebraic_connectivity": round(m.normalized_algebraic_connectivity, 4), + "spectral_entropy": round(m.spectral_entropy, 4), + "normalized_spectral_entropy": round(m.normalized_spectral_entropy, 4), + "edge_density": round(m.edge_density, 4), + "cross_module_ratio": round(m.cross_module_ratio, 4), + "spectral_radius": round(m.spectral_radius, 4), + "normalized_spectral_radius": round(m.normalized_spectral_radius, 4), + "cci": round(m.cci, 4), + "cci_label": cci_label, + "cci_color": cci_color, + "cci_desc": cci_desc, + "fiedler_value": round(s.fiedler_value, 4), + } + + data_blob = json.dumps({ + "eigenvalues": eigenvalue_data, + "fiedler": fiedler_data, + "coupling": coupling_data, + "metrics": metrics_json, + "module_colors": module_colors_json, + }) + + # Escape DOT source for embedding in a JS template literal + dot_escaped = (dot_source + .replace("\\", "\\\\") + .replace("`", "\\`") + .replace("${", "\\${")) + + html = _DASHBOARD_HTML_TEMPLATE.replace("__DATA_BLOB__", data_blob) + html = html.replace("__DOT_BLOB__", dot_escaped) + + with open(output_path, "w") as f: + f.write(html) + + +_DASHBOARD_HTML_TEMPLATE = r""" + + +swactor — dependency analysis + + + + +
+
swactor — dependency analysis
+ + +
+ +
+
+ + + + scroll to zoom · drag to pan · click node to focus +
+
+
Loading Graphviz…
+
+ +
+
+
+

λ Laplacian Eigenvalue Spectrum

+ +
+ +
+

✂ Fiedler Vector — Spectral Bisection

+ +
+ +
+

▦ Module Coupling (directed edge counts)

+ +
+ +
+

∑ Complexity Metrics

+
+
+
+
+ +
+ + + + + + + +""" + + +# ─── JSON Output ────────────────────────────────────────────────────────────── + +def metrics_to_dict(result: AnalysisResult) -> dict[str, Any]: + """Convert analysis results to a JSON-serializable dict.""" + m = result.metrics + s = result.spectral + c = result.coupling + + return { + "graph": { + "n_nodes": m.n_nodes, + "n_edges": m.n_edges, + "n_modules": m.n_modules, + "connected_components": m.connected_components, + "modules": c.module_names, + }, + "spectral": { + "eigenvalues": s.eigenvalues.tolist(), + "fiedler_value": s.fiedler_value, + "fiedler_vector": s.fiedler_vector.tolist(), + "node_names": s.node_names, + "node_modules": s.node_modules, + }, + "module_coupling": { + "module_names": c.module_names, + "coupling_matrix": c.coupling_matrix.tolist(), + "cross_module_edges": c.cross_module_edges, + "total_edges": c.total_edges, + }, + "metrics": { + "algebraic_connectivity": m.algebraic_connectivity, + "normalized_algebraic_connectivity": m.normalized_algebraic_connectivity, + "spectral_entropy": m.spectral_entropy, + "normalized_spectral_entropy": m.normalized_spectral_entropy, + "edge_density": m.edge_density, + "cross_module_ratio": m.cross_module_ratio, + "spectral_radius": m.spectral_radius, + "normalized_spectral_radius": m.normalized_spectral_radius, + "cci": m.cci, + }, + } + + +# ─── CLI ────────────────────────────────────────────────────────────────────── + +def main() -> None: + parser = argparse.ArgumentParser( + description="Spectral analysis of dependency DAGs" + ) + parser.add_argument("dot_file", help="Path to DOT file (from depgraph)") + parser.add_argument("-o", "--output-dir", default=".", + help="Output directory (default: current directory)") + parser.add_argument("--no-plots", action="store_true", + help="Text report only (no matplotlib dependency)") + parser.add_argument("--json", action="store_true", + help="Also output spectral_metrics.json") + args = parser.parse_args() + + # Read and parse DOT + dot_text = open(args.dot_file).read() + graph = parse_dot(dot_text) + print(f"Parsed {len(graph.nodes)} nodes, {len(graph.edges)} edges, " + f"{len(graph.modules)} modules") + + # Run analysis + result = run_analysis(graph) + + # Ensure output directory exists + os.makedirs(args.output_dir, exist_ok=True) + + # Generate report + report = generate_report(result) + print(report) + report_path = os.path.join(args.output_dir, "spectral_report.txt") + with open(report_path, "w") as f: + f.write(report) + print(f"\nReport saved to {report_path}") + + # Generate interactive HTML dashboard + html_path = os.path.join(args.output_dir, "spectral_dashboard.html") + generate_dashboard_html(result, html_path, dot_source=dot_text) + print(f"Interactive dashboard saved to {html_path}") + + # Generate static PNG dashboard + if not args.no_plots: + dashboard_path = os.path.join(args.output_dir, "spectral_dashboard.png") + generate_dashboard(result, dashboard_path) + print(f"Static dashboard saved to {dashboard_path}") + + # Generate JSON + if args.json: + json_path = os.path.join(args.output_dir, "spectral_metrics.json") + with open(json_path, "w") as f: + json.dump(metrics_to_dict(result), f, indent=2) + print(f"JSON saved to {json_path}") + + +if __name__ == "__main__": + main() diff --git a/tools/spectral/test_spectral.py b/tools/spectral/test_spectral.py new file mode 100644 index 0000000..03f933b --- /dev/null +++ b/tools/spectral/test_spectral.py @@ -0,0 +1,727 @@ +#!/usr/bin/env python3 +"""Comprehensive tests for the spectral analysis tool.""" + +from __future__ import annotations + +import copy +import json +import math +import os +import random +import tempfile +import unittest + +import numpy as np + +from spectral_analysis import ( + AnalysisResult, + ComplexityMetrics, + DependencyGraph, + Edge, + ModuleCouplingResult, + Node, + SpectralResults, + build_adjacency, + build_laplacian, + compute_complexity_metrics, + compute_module_coupling, + compute_spectral, + compute_spectral_entropy, + count_connected_components, + generate_report, + get_node_ordering, + metrics_to_dict, + parse_dot, + run_analysis, + symmetrize, +) + + +# ─── Helpers ────────────────────────────────────────────────────────────────── + +def _make_graph( + names: list[str], + modules: list[str], + edge_pairs: list[tuple[str, str]], + module_order: list[str] | None = None, +) -> DependencyGraph: + """Build a DependencyGraph from names, module assignments, and edges.""" + assert len(names) == len(modules) + graph = DependencyGraph() + seen_modules: list[str] = [] + for name, mod in zip(names, modules): + graph.nodes.append(Node(name=name, module=mod)) + graph.node_to_module[name] = mod + if mod not in seen_modules: + seen_modules.append(mod) + if module_order is not None: + graph.modules = module_order + else: + graph.modules = seen_modules + for src, tgt in edge_pairs: + src_mod = graph.node_to_module.get(src, "") + tgt_mod = graph.node_to_module.get(tgt, "") + cross = src_mod != tgt_mod + graph.edges.append(Edge( + source=src, target=tgt, label="dep", + edge_type="field", cross_module=cross, + )) + return graph + + +# ─── DOT Parser Tests ───────────────────────────────────────────────────────── + +class TestDotParser(unittest.TestCase): + def test_minimal_dot(self): + dot = '''digraph test { + subgraph cluster_mod1 { + label="mod1"; + A [label="A", fillcolor="#fff"]; + } + A -> A [label="self", style=dashed, color="#666", penwidth=1]; +}''' + g = parse_dot(dot) + self.assertEqual(len(g.nodes), 1) + self.assertEqual(g.nodes[0].name, "A") + self.assertEqual(g.nodes[0].module, "mod1") + self.assertEqual(len(g.edges), 1) + + def test_two_module_dot(self): + dot = '''digraph test { + subgraph cluster_alpha { + label="alpha"; + X [label="X"]; + Y [label="Y"]; + } + subgraph cluster_beta { + label="beta"; + Z [label="Z"]; + } + X -> Y [label="dep", style=dashed, color="#666", penwidth=1]; + X -> Z [label="dep", style=solid, color="#00f", penwidth=1.5]; +}''' + g = parse_dot(dot) + self.assertEqual(len(g.nodes), 3) + self.assertEqual(len(g.modules), 2) + self.assertEqual(g.modules, ["alpha", "beta"]) + self.assertEqual(g.node_to_module["X"], "alpha") + self.assertEqual(g.node_to_module["Z"], "beta") + + # Edge classification + intra = [e for e in g.edges if not e.cross_module] + cross = [e for e in g.edges if e.cross_module] + self.assertEqual(len(intra), 1) + self.assertEqual(len(cross), 1) + + def test_trait_impl_classification(self): + dot = '''digraph test { + subgraph cluster_m { + label="m"; + A [label="A"]; + B [label="B"]; + } + A -> B [label="impl", style=dotted, color="#666", penwidth=1]; +}''' + g = parse_dot(dot) + self.assertEqual(g.edges[0].edge_type, "trait_impl") + + def test_real_deps_dot(self): + """Parse the real deps.dot and verify expected counts.""" + dot_path = os.path.join(os.path.dirname(__file__), "..", "..", "deps.dot") + if not os.path.exists(dot_path): + self.skipTest("deps.dot not found") + with open(dot_path) as f: + dot = f.read() + g = parse_dot(dot) + self.assertEqual(len(g.nodes), 36, f"Expected 36 nodes, got {len(g.nodes)}") + self.assertEqual(len(g.edges), 89, f"Expected 89 edges, got {len(g.edges)}") + self.assertEqual(len(g.modules), 8, f"Expected 8 modules, got {len(g.modules)}") + + def test_empty_dot(self): + dot = "digraph empty {}" + g = parse_dot(dot) + self.assertEqual(len(g.nodes), 0) + self.assertEqual(len(g.edges), 0) + + +# ─── Matrix Construction Tests ──────────────────────────────────────────────── + +class TestMatrixConstruction(unittest.TestCase): + def test_two_node_adjacency(self): + g = _make_graph(["A", "B"], ["m", "m"], [("A", "B")]) + order = get_node_ordering(g) + A = build_adjacency(g, order) + self.assertEqual(A.shape, (2, 2)) + idx_a = order.index("A") + idx_b = order.index("B") + self.assertEqual(A[idx_a, idx_b], 1.0) + self.assertEqual(A[idx_b, idx_a], 0.0) + + def test_symmetrize_directed(self): + A = np.array([[0, 1, 0], + [0, 0, 1], + [0, 0, 0]], dtype=float) + S = symmetrize(A) + expected = np.array([[0, 1, 0], + [1, 0, 1], + [0, 1, 0]], dtype=float) + np.testing.assert_array_equal(S, expected) + + def test_symmetrize_idempotent(self): + """Symmetrizing an already-symmetric matrix should not change it.""" + A = np.array([[0, 1, 1], + [1, 0, 1], + [1, 1, 0]], dtype=float) + S = symmetrize(A) + np.testing.assert_array_equal(S, A) + + def test_laplacian_p3(self): + """Path graph P3: A-B-C.""" + A_sym = np.array([[0, 1, 0], + [1, 0, 1], + [0, 1, 0]], dtype=float) + L = build_laplacian(A_sym) + expected = np.array([[1, -1, 0], + [-1, 2, -1], + [0, -1, 1]], dtype=float) + np.testing.assert_array_equal(L, expected) + + def test_laplacian_k3(self): + """Complete graph K3.""" + A_sym = np.array([[0, 1, 1], + [1, 0, 1], + [1, 1, 0]], dtype=float) + L = build_laplacian(A_sym) + expected = np.array([[2, -1, -1], + [-1, 2, -1], + [-1, -1, 2]], dtype=float) + np.testing.assert_array_equal(L, expected) + + +# ─── Spectral Analysis Tests ───────────────────────────────────────────────── + +class TestSpectralAnalysis(unittest.TestCase): + def test_p3_eigenvalues(self): + """Path P3 should have eigenvalues {0, 1, 3}.""" + g = _make_graph(["A", "B", "C"], ["m", "m", "m"], + [("A", "B"), ("B", "C")]) + s = compute_spectral(g) + np.testing.assert_allclose(sorted(s.eigenvalues), [0, 1, 3], atol=1e-10) + + def test_k4_eigenvalues(self): + """Complete K4 should have eigenvalues {0, 4, 4, 4}.""" + names = ["A", "B", "C", "D"] + edges = [(a, b) for a in names for b in names if a != b] + g = _make_graph(names, ["m"] * 4, edges) + s = compute_spectral(g) + np.testing.assert_allclose(sorted(s.eigenvalues), [0, 4, 4, 4], atol=1e-10) + + def test_star_s4_fiedler(self): + """Star graph S4 (center + 3 leaves): lambda_2 = 1.""" + g = _make_graph( + ["C", "L1", "L2", "L3"], ["m"] * 4, + [("C", "L1"), ("C", "L2"), ("C", "L3")], + ) + s = compute_spectral(g) + self.assertAlmostEqual(s.fiedler_value, 1.0, places=10) + + def test_disconnected_graph(self): + """Disconnected graph should have lambda_2 = 0.""" + g = _make_graph( + ["A", "B", "C", "D"], ["m1", "m1", "m2", "m2"], + [("A", "B"), ("C", "D")], + module_order=["m1", "m2"], + ) + s = compute_spectral(g) + self.assertAlmostEqual(s.fiedler_value, 0.0, places=10) + + def test_barbell_fiedler_separation(self): + """Barbell graph: two K3 cliques connected by a bridge. + + Fiedler vector should separate the two cliques (different signs). + """ + # Clique 1: A, B, C fully connected + # Clique 2: D, E, F fully connected + # Bridge: C-D + names = ["A", "B", "C", "D", "E", "F"] + edges = [ + ("A", "B"), ("A", "C"), ("B", "C"), + ("D", "E"), ("D", "F"), ("E", "F"), + ("C", "D"), + ] + g = _make_graph(names, ["m1", "m1", "m1", "m2", "m2", "m2"], edges, + module_order=["m1", "m2"]) + s = compute_spectral(g) + + # Clique 1 nodes should have same sign, clique 2 opposite + order = s.node_names + fv = s.fiedler_vector + idx = {name: i for i, name in enumerate(order)} + + clique1_signs = [np.sign(fv[idx[n]]) for n in ["A", "B", "C"]] + clique2_signs = [np.sign(fv[idx[n]]) for n in ["D", "E", "F"]] + + # All in clique 1 should have the same sign + self.assertTrue(all(s == clique1_signs[0] for s in clique1_signs), + f"Clique 1 signs should be uniform: {clique1_signs}") + # All in clique 2 should have the same sign + self.assertTrue(all(s == clique2_signs[0] for s in clique2_signs), + f"Clique 2 signs should be uniform: {clique2_signs}") + # The two cliques should have opposite signs + self.assertNotEqual(clique1_signs[0], clique2_signs[0], + "Cliques should have opposite Fiedler signs") + + def test_single_node(self): + g = _make_graph(["A"], ["m"], []) + s = compute_spectral(g) + self.assertEqual(s.fiedler_value, 0.0) + self.assertEqual(len(s.eigenvalues), 1) + + def test_empty_graph(self): + g = DependencyGraph() + s = compute_spectral(g) + self.assertEqual(s.fiedler_value, 0.0) + self.assertEqual(len(s.eigenvalues), 0) + + +# ─── Module Coupling Tests ──────────────────────────────────────────────────── + +class TestModuleCoupling(unittest.TestCase): + def test_directed_counts(self): + g = _make_graph( + ["A", "B", "C"], ["m1", "m1", "m2"], + [("A", "C"), ("B", "C"), ("C", "A")], + module_order=["m1", "m2"], + ) + c = compute_module_coupling(g) + # m1->m2: 2 edges (A->C, B->C) + # m2->m1: 1 edge (C->A) + idx_m1 = c.module_names.index("m1") + idx_m2 = c.module_names.index("m2") + self.assertEqual(c.coupling_matrix[idx_m1, idx_m2], 2.0) + self.assertEqual(c.coupling_matrix[idx_m2, idx_m1], 1.0) + + def test_cross_module_ratio(self): + g = _make_graph( + ["A", "B", "C", "D"], ["m1", "m1", "m2", "m2"], + [("A", "B"), ("A", "C"), ("C", "D")], + module_order=["m1", "m2"], + ) + c = compute_module_coupling(g) + # 1 cross-module edge (A->C) out of 3 total + self.assertEqual(c.cross_module_edges, 1) + self.assertEqual(c.total_edges, 3) + + def test_intra_only(self): + g = _make_graph( + ["A", "B"], ["m1", "m1"], + [("A", "B")], + module_order=["m1"], + ) + c = compute_module_coupling(g) + self.assertEqual(c.cross_module_edges, 0) + self.assertEqual(c.coupling_matrix[0, 0], 1.0) + + +# ─── Complexity Metrics Tests ───────────────────────────────────────────────── + +class TestComplexityMetrics(unittest.TestCase): + def test_k4_spectral_entropy(self): + """K4 has uniform positive eigenvalues {4,4,4} -> entropy = log2(3).""" + evals = np.array([0.0, 4.0, 4.0, 4.0]) + H = compute_spectral_entropy(evals) + self.assertAlmostEqual(H, math.log2(3), places=10) + + def test_star_entropy_less_than_complete(self): + """Star graph has less uniform eigenvalues than complete graph.""" + # Star S4: eigenvalues are 0, 1, 1, 4 + star_evals = np.array([0.0, 1.0, 1.0, 4.0]) + k4_evals = np.array([0.0, 4.0, 4.0, 4.0]) + H_star = compute_spectral_entropy(star_evals) + H_k4 = compute_spectral_entropy(k4_evals) + self.assertLess(H_star, H_k4) + + def test_cci_in_range(self): + """CCI should always be in [0, 1].""" + for _ in range(20): + n = random.randint(2, 10) + names = [f"N{i}" for i in range(n)] + mods = [f"m{i % 3}" for i in range(n)] + edges = [] + for _ in range(random.randint(1, n * 2)): + a, b = random.sample(names, 2) + edges.append((a, b)) + g = _make_graph(names, mods, edges, + module_order=sorted(set(mods))) + result = run_analysis(g) + self.assertGreaterEqual(result.metrics.cci, 0.0, + "CCI should be >= 0") + self.assertLessEqual(result.metrics.cci, 1.0, + "CCI should be <= 1") + + def test_cci_increases_with_coupling(self): + """Adding cross-module edges should increase CCI.""" + # Base graph: two modules, minimal coupling + g1 = _make_graph( + ["A", "B", "C", "D"], ["m1", "m1", "m2", "m2"], + [("A", "B"), ("C", "D"), ("A", "C")], + module_order=["m1", "m2"], + ) + # More coupling + g2 = _make_graph( + ["A", "B", "C", "D"], ["m1", "m1", "m2", "m2"], + [("A", "B"), ("C", "D"), ("A", "C"), ("A", "D"), + ("B", "C"), ("B", "D"), ("C", "A"), ("D", "B")], + module_order=["m1", "m2"], + ) + r1 = run_analysis(g1) + r2 = run_analysis(g2) + self.assertLess(r1.metrics.cci, r2.metrics.cci) + + def test_connected_components(self): + A_sym = np.array([ + [0, 1, 0, 0], + [1, 0, 0, 0], + [0, 0, 0, 1], + [0, 0, 1, 0], + ], dtype=float) + self.assertEqual(count_connected_components(A_sym), 2) + + def test_single_component(self): + A_sym = np.array([ + [0, 1, 1], + [1, 0, 1], + [1, 1, 0], + ], dtype=float) + self.assertEqual(count_connected_components(A_sym), 1) + + +# ─── Complexity Ladder ──────────────────────────────────────────────────────── + +class TestComplexityLadder(unittest.TestCase): + """Verify CCI correctly orders synthetic codebases of increasing complexity.""" + + def _rung1_linear_chain(self) -> DependencyGraph: + """5 nodes in a single module, linear chain A->B->C->D->E.""" + return _make_graph( + ["A", "B", "C", "D", "E"], + ["m1"] * 5, + [("A", "B"), ("B", "C"), ("C", "D"), ("D", "E")], + module_order=["m1"], + ) + + def _rung2_clean_tree(self) -> DependencyGraph: + """6 nodes across 2 modules, tree with mostly intra-module edges.""" + return _make_graph( + ["R", "A", "B", "C", "D", "E"], + ["core", "core", "core", "util", "util", "util"], + [ + ("R", "A"), ("A", "B"), ("R", "C"), # intra core + ("D", "E"), # intra util + ("R", "D"), ("C", "E"), # 2 cross edges + ], + module_order=["core", "util"], + ) + + def _rung3_layered_dag(self) -> DependencyGraph: + """8 nodes across 3 modules in a layered architecture.""" + return _make_graph( + ["C1", "C2", "S1", "S2", "S3", "D1", "D2", "D3"], + ["ctrl", "ctrl", "svc", "svc", "svc", "data", "data", "data"], + [ + ("C1", "C2"), # intra ctrl + ("S1", "S2"), ("S2", "S3"), # intra svc + ("D1", "D2"), ("D2", "D3"), # intra data + ("C1", "S1"), ("C1", "S2"), ("C2", "S3"), # ctrl->svc + ("S1", "D1"), ("S2", "D2"), ("S3", "D3"), # svc->data + ], + module_order=["ctrl", "svc", "data"], + ) + + def _rung4_diamond_cross(self) -> DependencyGraph: + """10 nodes across 5 modules with diamond patterns and cross-coupling.""" + return _make_graph( + ["A1", "A2", "B1", "B2", "C1", "C2", "D1", "D2", "E1", "E2"], + ["ma", "ma", "mb", "mb", "mc", "mc", "md", "md", "me", "me"], + [ + ("A1", "A2"), ("B1", "B2"), ("C1", "C2"), # intra + ("D1", "D2"), ("E1", "E2"), # intra + # Diamonds across modules + ("A1", "B1"), ("A1", "C1"), ("B1", "D1"), ("C1", "D1"), + ("A2", "B2"), ("A2", "C2"), ("B2", "D2"), ("C2", "D2"), + # Extra cross-coupling + ("D1", "E1"), ("D2", "E2"), ("B1", "E1"), + ], + module_order=["ma", "mb", "mc", "md", "me"], + ) + + def _rung5_hub_backlinks(self) -> DependencyGraph: + """10 nodes across 5 modules, hub-dominated with back-edges.""" + return _make_graph( + ["Hub", "A1", "A2", "B1", "B2", "C1", "C2", "D1", "D2", "D3"], + ["core", "sa", "sa", "sb", "sb", "sc", "sc", "sd", "sd", "sd"], + [ + ("A1", "A2"), ("B1", "B2"), ("C1", "C2"), # intra + ("D1", "D2"), ("D2", "D3"), # intra + # Hub connections (cross-module) + ("Hub", "A1"), ("Hub", "B1"), ("Hub", "C1"), ("Hub", "D1"), + ("A1", "Hub"), ("B1", "Hub"), ("C1", "Hub"), + # Additional cross-module + ("A1", "B1"), ("B1", "C1"), ("C1", "D1"), + ("A2", "B2"), ("B2", "C2"), ("C2", "D2"), + ("A1", "D1"), ("B2", "D3"), + ], + module_order=["core", "sa", "sb", "sc", "sd"], + ) + + def _rung6_dense_mesh(self) -> DependencyGraph: + """10 nodes across 4 modules with heavy cross-module coupling.""" + names = ["X1", "X2", "X3", "Y1", "Y2", "Y3", "Z1", "Z2", "W1", "W2"] + mods = ["mx", "mx", "mx", "my", "my", "my", "mz", "mz", "mw", "mw"] + # Dense cross-module edges + edges = [ + # intra + ("X1", "X2"), ("X2", "X3"), ("Y1", "Y2"), ("Y2", "Y3"), + ("Z1", "Z2"), ("W1", "W2"), + # cross - nearly every module to every other + ("X1", "Y1"), ("X1", "Z1"), ("X1", "W1"), + ("X2", "Y2"), ("X2", "Z2"), ("X2", "W2"), + ("X3", "Y3"), ("X3", "Z1"), + ("Y1", "X1"), ("Y1", "Z1"), ("Y1", "W1"), + ("Y2", "X2"), ("Y2", "Z2"), + ("Y3", "X3"), ("Y3", "W2"), + ("Z1", "X1"), ("Z1", "Y1"), ("Z1", "W1"), + ("Z2", "X2"), ("Z2", "Y2"), ("Z2", "W2"), + ("W1", "X1"), ("W1", "Y1"), ("W1", "Z1"), + ("W2", "X2"), ("W2", "Y2"), ("W2", "Z2"), + ] + return _make_graph(names, mods, edges, + module_order=["mx", "my", "mz", "mw"]) + + def test_complexity_ladder(self): + """CCI must strictly increase across the ladder rungs.""" + ladder = [ + self._rung1_linear_chain(), + self._rung2_clean_tree(), + self._rung3_layered_dag(), + self._rung4_diamond_cross(), + self._rung5_hub_backlinks(), + self._rung6_dense_mesh(), + ] + ccis = [run_analysis(g).metrics.cci for g in ladder] + for i in range(len(ccis) - 1): + self.assertLess( + ccis[i], ccis[i + 1], + f"Rung {i + 1} (CCI={ccis[i]:.4f}) should be less complex " + f"than rung {i + 2} (CCI={ccis[i + 1]:.4f})" + ) + + +# ─── Perturbation Tests ────────────────────────────────────────────────────── + +class TestPerturbation(unittest.TestCase): + """Test that CCI responds correctly to architectural changes on the real graph.""" + + def _load_real_graph(self) -> DependencyGraph: + dot_path = os.path.join(os.path.dirname(__file__), "..", "..", "deps.dot") + if not os.path.exists(dot_path): + self.skipTest("deps.dot not found") + with open(dot_path) as f: + return parse_dot(f.read()) + + def test_remove_most_coupled_module(self): + """Removing the runtime module should decrease CCI.""" + g = self._load_real_graph() + original_cci = run_analysis(g).metrics.cci + + # Remove runtime nodes and their edges + g2 = DependencyGraph() + g2.modules = [m for m in g.modules if m != "runtime"] + for node in g.nodes: + if node.module != "runtime": + g2.nodes.append(node) + g2.node_to_module[node.name] = node.module + runtime_nodes = {n.name for n in g.nodes if n.module == "runtime"} + for edge in g.edges: + if edge.source not in runtime_nodes and edge.target not in runtime_nodes: + src_mod = g2.node_to_module.get(edge.source, "") + tgt_mod = g2.node_to_module.get(edge.target, "") + g2.edges.append(Edge( + source=edge.source, target=edge.target, label=edge.label, + edge_type=edge.edge_type, + cross_module=src_mod != tgt_mod, + )) + + reduced_cci = run_analysis(g2).metrics.cci + self.assertLess(reduced_cci, original_cci, + f"Removing runtime should decrease CCI: " + f"{reduced_cci:.4f} vs {original_cci:.4f}") + + def test_add_random_cross_edges(self): + """Adding 10 random cross-module edges should increase CCI.""" + g = self._load_real_graph() + original_cci = run_analysis(g).metrics.cci + + g2 = copy.deepcopy(g) + random.seed(42) + node_names = [n.name for n in g2.nodes] + added = 0 + attempts = 0 + while added < 10 and attempts < 100: + src, tgt = random.sample(node_names, 2) + src_mod = g2.node_to_module[src] + tgt_mod = g2.node_to_module[tgt] + if src_mod != tgt_mod: + g2.edges.append(Edge( + source=src, target=tgt, label="added", + edge_type="field", cross_module=True, + )) + added += 1 + attempts += 1 + + augmented_cci = run_analysis(g2).metrics.cci + self.assertGreater(augmented_cci, original_cci, + f"Adding cross-module edges should increase CCI: " + f"{augmented_cci:.4f} vs {original_cci:.4f}") + + def test_merge_modules_decreases_cci(self): + """Merging two small modules into one should decrease CCI. + + Merging error + config into a single module reduces cross-module + edges (their mutual and outward coupling consolidates), lowering CCI. + """ + g = self._load_real_graph() + original_cci = run_analysis(g).metrics.cci + + # Merge error and config into "error_config" + merge_set = {"error", "config"} + merged_name = "error_config" + + g2 = DependencyGraph() + g2.modules = [merged_name if m in merge_set else m + for m in g.modules if m not in merge_set] + if merged_name not in g2.modules: + g2.modules.insert(0, merged_name) + # Deduplicate + seen = set() + g2.modules = [m for m in g2.modules if not (m in seen or seen.add(m))] + + for node in g.nodes: + new_mod = merged_name if node.module in merge_set else node.module + g2.nodes.append(Node(name=node.name, module=new_mod)) + g2.node_to_module[node.name] = new_mod + + for edge in g.edges: + src_mod = g2.node_to_module.get(edge.source, "") + tgt_mod = g2.node_to_module.get(edge.target, "") + g2.edges.append(Edge( + source=edge.source, target=edge.target, label=edge.label, + edge_type=edge.edge_type, + cross_module=src_mod != tgt_mod, + )) + + merged_cci = run_analysis(g2).metrics.cci + self.assertLess(merged_cci, original_cci, + f"Merging error+config should decrease CCI: " + f"{merged_cci:.4f} vs {original_cci:.4f}") + + +# ─── Integration Tests ──────────────────────────────────────────────────────── + +class TestIntegration(unittest.TestCase): + def test_full_pipeline_real_graph(self): + """Run full pipeline on real deps.dot and sanity-check outputs.""" + dot_path = os.path.join(os.path.dirname(__file__), "..", "..", "deps.dot") + if not os.path.exists(dot_path): + self.skipTest("deps.dot not found") + with open(dot_path) as f: + graph = parse_dot(f.read()) + + result = run_analysis(graph) + + # Basic sanity checks + self.assertEqual(result.metrics.n_nodes, 36) + self.assertEqual(result.metrics.n_edges, 89) + self.assertEqual(result.metrics.n_modules, 8) + + # Connected graph -> lambda_2 > 0 + self.assertGreater(result.spectral.fiedler_value, 0, + "Connected graph should have lambda_2 > 0") + + # CCI should be in a reasonable range for a well-structured codebase + self.assertGreater(result.metrics.cci, 0.05) + self.assertLess(result.metrics.cci, 0.9) + + # Eigenvalues should be non-negative (Laplacian property) + self.assertTrue(np.all(result.spectral.eigenvalues >= -1e-10), + "Laplacian eigenvalues should be non-negative") + + # First eigenvalue should be 0 + self.assertAlmostEqual(result.spectral.eigenvalues[0], 0.0, places=8) + + def test_report_generation(self): + """Verify report contains expected sections.""" + dot_path = os.path.join(os.path.dirname(__file__), "..", "..", "deps.dot") + if not os.path.exists(dot_path): + self.skipTest("deps.dot not found") + with open(dot_path) as f: + graph = parse_dot(f.read()) + result = run_analysis(graph) + report = generate_report(result) + + self.assertIn("GRAPH SUMMARY", report) + self.assertIn("LAPLACIAN EIGENVALUE SPECTRUM", report) + self.assertIn("FIEDLER VECTOR", report) + self.assertIn("MODULE COUPLING MATRIX", report) + self.assertIn("CONNECTOME COMPLEXITY INDEX", report) + + def test_json_output(self): + """Verify JSON output is well-formed and contains expected keys.""" + g = _make_graph( + ["A", "B", "C"], ["m1", "m1", "m2"], + [("A", "B"), ("A", "C")], + module_order=["m1", "m2"], + ) + result = run_analysis(g) + d = metrics_to_dict(result) + + self.assertIn("graph", d) + self.assertIn("spectral", d) + self.assertIn("module_coupling", d) + self.assertIn("metrics", d) + self.assertEqual(d["graph"]["n_nodes"], 3) + self.assertIsInstance(d["spectral"]["eigenvalues"], list) + self.assertIsInstance(d["metrics"]["cci"], float) + + # Should be JSON-serializable + json_str = json.dumps(d) + self.assertIsInstance(json_str, str) + + def test_dashboard_generation(self): + """Verify dashboard PNG can be generated without errors.""" + try: + import matplotlib + except ImportError: + self.skipTest("matplotlib not available") + + g = _make_graph( + ["A", "B", "C", "D"], ["m1", "m1", "m2", "m2"], + [("A", "B"), ("A", "C"), ("C", "D")], + module_order=["m1", "m2"], + ) + result = run_analysis(g) + + with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f: + path = f.name + try: + from spectral_analysis import generate_dashboard + generate_dashboard(result, path) + self.assertTrue(os.path.exists(path)) + self.assertGreater(os.path.getsize(path), 1000, + "Dashboard should be a non-trivial PNG") + finally: + os.unlink(path) + + +if __name__ == "__main__": + unittest.main()