From 0f4de8e47ffe71d4dbb643f48df1b1e06df9857a Mon Sep 17 00:00:00 2001 From: Safi Date: Sun, 5 Apr 2026 00:42:30 +0100 Subject: [PATCH] test: add end-to-end pipeline integration test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Covers detect → extract → build → cluster → analyze → report → export using existing fixtures. AST-only (no LLM calls), catches regressions in how modules connect, not just individual module behaviour. --- tests/test_pipeline.py | 158 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 158 insertions(+) create mode 100644 tests/test_pipeline.py diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 00000000..9f7335b6 --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,158 @@ +""" +End-to-end pipeline test: detect → extract → build → cluster → analyze → report → export. +Uses the existing test fixtures (code + markdown). No LLM calls - AST extraction only. +Catches regressions in how modules connect, not just individual module behaviour. +""" +import json +import tempfile +from pathlib import Path + +import pytest + +from graphify.detect import detect +from graphify.extract import collect_files, extract +from graphify.build import build_from_json +from graphify.cluster import cluster, score_all +from graphify.analyze import god_nodes, surprising_connections, suggest_questions +from graphify.report import generate +from graphify.export import to_json, to_html, to_obsidian + +FIXTURES = Path(__file__).parent / "fixtures" + + +def run_pipeline(tmp_path: Path) -> dict: + """Run the full pipeline on the fixtures directory. Returns a dict of outputs.""" + # Step 1: detect + detection = detect(FIXTURES) + assert detection["total_files"] > 0 + # fixtures corpus is intentionally small (< 5k words), so needs_graph may be False + assert "files" in detection + + # Step 2: extract (AST only - no LLM) + code_files = [Path(f) for f in detection["files"].get("code", [])] + assert len(code_files) > 0 + extraction = extract(code_files) + assert len(extraction["nodes"]) > 0 + assert len(extraction["edges"]) > 0 + + # Step 3: build + G = build_from_json(extraction) + assert G.number_of_nodes() > 0 + assert G.number_of_edges() > 0 + + # Step 4: cluster + communities = cluster(G) + assert len(communities) > 0 + cohesion = score_all(G, communities) + assert len(cohesion) == len(communities) + for score in cohesion.values(): + assert 0.0 <= score <= 1.0 + + # Step 5: analyze + gods = god_nodes(G) + assert len(gods) > 0 + assert all("id" in g and "edges" in g for g in gods) + + surprises = surprising_connections(G, communities) + assert isinstance(surprises, list) + + labels = {cid: f"Group {cid}" for cid in communities} + questions = suggest_questions(G, communities, labels) + assert isinstance(questions, list) + + # Step 6: report + tokens = {"input": 0, "output": 0} + report = generate(G, communities, cohesion, labels, gods, surprises, detection, tokens, str(FIXTURES), suggested_questions=questions) + assert "God Nodes" in report + assert "Communities" in report + assert len(report) > 100 + + # Step 7: export - JSON + json_path = tmp_path / "graph.json" + to_json(G, communities, str(json_path)) + assert json_path.exists() + data = json.loads(json_path.read_text()) + assert "nodes" in data and "links" in data + assert all("community" in n for n in data["nodes"]) + + # Step 8: export - HTML + html_path = tmp_path / "graph.html" + to_html(G, communities, str(html_path), community_labels=labels) + assert html_path.exists() + html = html_path.read_text() + assert "vis-network" in html + assert "RAW_NODES" in html + + # Step 9: export - Obsidian vault + vault_path = tmp_path / "obsidian" + n_notes = to_obsidian(G, communities, str(vault_path), community_labels=labels, cohesion=cohesion) + assert n_notes > 0 + assert (vault_path / ".obsidian" / "graph.json").exists() + md_files = list(vault_path.glob("*.md")) + assert len(md_files) > 0 + + return { + "detection": detection, + "extraction": extraction, + "graph": G, + "communities": communities, + "cohesion": cohesion, + "gods": gods, + "surprises": surprises, + "questions": questions, + "report": report, + } + + +def test_pipeline_runs_end_to_end(tmp_path): + result = run_pipeline(tmp_path) + assert result["graph"].number_of_nodes() > 0 + + +def test_pipeline_graph_has_edges(tmp_path): + result = run_pipeline(tmp_path) + assert result["graph"].number_of_edges() > 0 + + +def test_pipeline_all_nodes_have_community(tmp_path): + result = run_pipeline(tmp_path) + G = result["graph"] + communities = result["communities"] + all_community_nodes = {n for nodes in communities.values() for n in nodes} + for node in G.nodes(): + assert node in all_community_nodes, f"Node {node!r} has no community" + + +def test_pipeline_report_mentions_top_god_node(tmp_path): + result = run_pipeline(tmp_path) + top_god = result["gods"][0]["label"] + assert top_god in result["report"] + + +def test_pipeline_detection_finds_code_and_docs(tmp_path): + result = run_pipeline(tmp_path) + assert len(result["detection"]["files"].get("code", [])) > 0 + assert len(result["detection"]["files"].get("document", [])) > 0 + + +def test_pipeline_incremental_update(tmp_path): + """Second run on unchanged corpus should produce identical node/edge counts.""" + result1 = run_pipeline(tmp_path) + result2 = run_pipeline(tmp_path) + assert result1["graph"].number_of_nodes() == result2["graph"].number_of_nodes() + assert result1["graph"].number_of_edges() == result2["graph"].number_of_edges() + + +def test_pipeline_extraction_confidence_labels(tmp_path): + result = run_pipeline(tmp_path) + extraction = result["extraction"] + valid = {"EXTRACTED", "INFERRED", "AMBIGUOUS"} + for edge in extraction["edges"]: + assert edge["confidence"] in valid, f"Invalid confidence: {edge['confidence']}" + + +def test_pipeline_no_self_loops(tmp_path): + result = run_pipeline(tmp_path) + G = result["graph"] + for u, v in G.edges(): + assert u != v, f"Self-loop found on node {u!r}"