graphify/tests/test_hypergraph.py

"""Tests for hyperedge support in graphify."""
from __future__ import annotations
import json
import tempfile
from pathlib import Path

import networkx as nx
import pytest

from graphify.build import build_from_json
from graphify.export import attach_hyperedges, to_json
from graphify.report import generate


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

SAMPLE_EXTRACTION = {
    "nodes": [
        {"id": "BasicAuth", "label": "BasicAuth", "file_type": "code", "source_file": "auth.py"},
        {"id": "DigestAuth", "label": "DigestAuth", "file_type": "code", "source_file": "auth.py"},
        {"id": "Request", "label": "Request", "file_type": "code", "source_file": "http.py"},
        {"id": "Response", "label": "Response", "file_type": "code", "source_file": "http.py"},
        {"id": "BaseClient", "label": "BaseClient", "file_type": "code", "source_file": "client.py"},
    ],
    "edges": [
        {"source": "BasicAuth", "target": "Request", "relation": "uses", "confidence": "EXTRACTED", "confidence_score": 1.0, "source_file": "auth.py"},
    ],
    "hyperedges": [
        {
            "id": "auth_flow",
            "label": "Auth Flow",
            "nodes": ["BasicAuth", "DigestAuth", "Request", "Response", "BaseClient"],
            "relation": "participate_in",
            "confidence": "INFERRED",
            "confidence_score": 0.75,
            "source_file": "auth.py",
        }
    ],
    "input_tokens": 10,
    "output_tokens": 5,
}

SAMPLE_DETECTION = {
    "total_files": 3,
    "total_words": 500,
    "files": {"code": ["auth.py", "http.py", "client.py"]},
    "skipped_sensitive": [],
    "warning": None,
}


# ---------------------------------------------------------------------------
# 1. Hyperedges survive build_from_json round-trip
# ---------------------------------------------------------------------------

def test_build_from_json_stores_hyperedges():
    G = build_from_json(SAMPLE_EXTRACTION)
    assert "hyperedges" in G.graph
    assert len(G.graph["hyperedges"]) == 1
    assert G.graph["hyperedges"][0]["id"] == "auth_flow"


def test_build_from_json_relativizes_hyperedge_source_file(tmp_path):
    """build_from_json(root=...) must relativize hyperedge source_file like it
    already does for nodes and edges. to_json writes G.graph['hyperedges']
    verbatim and has no root parameter, so an absolute path emitted by a semantic
    subagent would otherwise leak into graph.json (#1418)."""
    base = tmp_path.resolve()
    abs_doc = base / "docs" / "CLAUDE.md"
    extraction = {
        "nodes": [
            {"id": "a", "label": "A", "file_type": "document", "source_file": str(abs_doc)},
        ],
        "edges": [],
        "hyperedges": [
            {
                "id": "arch",
                "label": "Architecture",
                "nodes": ["a"],
                "relation": "participate_in",
                "confidence": "INFERRED",
                "confidence_score": 0.75,
                "source_file": str(abs_doc),
            }
        ],
    }
    G = build_from_json(extraction, root=str(base))
    assert G.graph["hyperedges"][0]["source_file"] == "docs/CLAUDE.md"
    # Anchor: the node path is relativized the same way (the contract this mirrors).
    assert G.nodes["a"]["source_file"] == "docs/CLAUDE.md"


def test_build_from_json_no_hyperedges():
    extraction = {**SAMPLE_EXTRACTION, "hyperedges": []}
    G = build_from_json(extraction)
    assert G.graph.get("hyperedges", []) == []


def test_build_from_json_missing_hyperedges_key():
    extraction = {k: v for k, v in SAMPLE_EXTRACTION.items() if k != "hyperedges"}
    G = build_from_json(extraction)
    assert G.graph.get("hyperedges", []) == []


# ---------------------------------------------------------------------------
# 2. attach_hyperedges deduplicates by id
# ---------------------------------------------------------------------------

def test_attach_hyperedges_adds_new():
    G = nx.Graph()
    attach_hyperedges(G, [{"id": "auth_flow", "label": "Auth Flow", "nodes": ["A", "B", "C"]}])
    assert len(G.graph["hyperedges"]) == 1


def test_attach_hyperedges_deduplicates():
    G = nx.Graph()
    h = {"id": "auth_flow", "label": "Auth Flow", "nodes": ["A", "B", "C"]}
    attach_hyperedges(G, [h])
    attach_hyperedges(G, [h])  # second call with same id should not duplicate
    assert len(G.graph["hyperedges"]) == 1


def test_attach_hyperedges_multiple_different_ids():
    G = nx.Graph()
    attach_hyperedges(G, [
        {"id": "flow_a", "label": "Flow A", "nodes": ["A", "B", "C"]},
        {"id": "flow_b", "label": "Flow B", "nodes": ["D", "E", "F"]},
    ])
    assert len(G.graph["hyperedges"]) == 2


def test_attach_hyperedges_skips_entry_without_id():
    G = nx.Graph()
    attach_hyperedges(G, [{"label": "No ID", "nodes": ["A", "B", "C"]}])
    assert G.graph.get("hyperedges", []) == []


# ---------------------------------------------------------------------------
# 3. to_json includes hyperedges key
# ---------------------------------------------------------------------------

def test_to_json_includes_hyperedges():
    G = build_from_json(SAMPLE_EXTRACTION)
    communities = {0: list(G.nodes())}
    with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
        path = f.name
    to_json(G, communities, path)
    data = json.loads(Path(path).read_text())
    assert "hyperedges" in data
    assert len(data["hyperedges"]) == 1
    assert data["hyperedges"][0]["id"] == "auth_flow"


def test_to_json_hyperedges_empty_when_none():
    extraction = {**SAMPLE_EXTRACTION, "hyperedges": []}
    G = build_from_json(extraction)
    communities = {0: list(G.nodes())}
    with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
        path = f.name
    to_json(G, communities, path)
    data = json.loads(Path(path).read_text())
    assert "hyperedges" in data
    assert data["hyperedges"] == []


# ---------------------------------------------------------------------------
# 4. Hyperedges loaded from graph.json via build_from_json
# ---------------------------------------------------------------------------

def test_hyperedges_roundtrip_via_json_file():
    """Write graph.json then reload it - hyperedges must survive."""
    G = build_from_json(SAMPLE_EXTRACTION)
    communities = {0: list(G.nodes())}
    with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
        path = f.name
    to_json(G, communities, path)

    # Reload the JSON as if build_from_json were called on it
    data = json.loads(Path(path).read_text())
    G2 = build_from_json({
        "nodes": [{"id": n["id"], **{k: v for k, v in n.items() if k != "id"}} for n in data["nodes"]],
        "edges": [{"source": e["source"], "target": e["target"], **{k: v for k, v in e.items() if k not in ("source", "target")}} for e in data.get("links", [])],
        "hyperedges": data.get("hyperedges", []),
    })
    assert G2.graph.get("hyperedges", []) != []
    assert G2.graph["hyperedges"][0]["id"] == "auth_flow"


# ---------------------------------------------------------------------------
# 5. Report includes hyperedges section when hyperedges present
# ---------------------------------------------------------------------------

def _make_report(G):
    communities = {0: list(G.nodes())}
    cohesion = {0: 1.0}
    labels = {0: "All"}
    gods = [{"label": "BasicAuth", "degree": 2}]
    surprises = []
    return generate(G, communities, cohesion, labels, gods, surprises, SAMPLE_DETECTION, {"input": 10, "output": 5}, ".")


def test_report_includes_hyperedges_section():
    G = build_from_json(SAMPLE_EXTRACTION)
    report = _make_report(G)
    assert "## Hyperedges (group relationships)" in report
    assert "Auth Flow" in report
    assert "INFERRED 0.75" in report


def test_report_includes_hyperedge_node_list():
    G = build_from_json(SAMPLE_EXTRACTION)
    report = _make_report(G)
    # Node IDs should appear in the report line
    assert "BasicAuth" in report
    assert "DigestAuth" in report


# ---------------------------------------------------------------------------
# 6. Report skips hyperedges section when none present
# ---------------------------------------------------------------------------

def test_report_skips_hyperedges_section_when_empty():
    extraction = {**SAMPLE_EXTRACTION, "hyperedges": []}
    G = build_from_json(extraction)
    report = _make_report(G)
    assert "## Hyperedges" not in report


def test_report_skips_hyperedges_section_when_key_missing():
    extraction = {k: v for k, v in SAMPLE_EXTRACTION.items() if k != "hyperedges"}
    G = build_from_json(extraction)
    report = _make_report(G)
    assert "## Hyperedges" not in report


# ---------------------------------------------------------------------------
# 7. Hyperedge member-key alias normalization (#1561)
# ---------------------------------------------------------------------------

def _alias_extraction():
    """Three hyperedges, one per member-key spelling: nodes / members / node_ids."""
    return {
        "nodes": [
            {"id": "a", "label": "A", "file_type": "code", "source_file": "m.py"},
            {"id": "b", "label": "B", "file_type": "code", "source_file": "m.py"},
            {"id": "c", "label": "C", "file_type": "code", "source_file": "m.py"},
        ],
        "edges": [],
        "hyperedges": [
            {"id": "he_nodes", "label": "canon", "nodes": ["a", "b", "c"]},
            {"id": "he_members", "label": "alias1", "members": ["a", "b", "c"]},
            {"id": "he_node_ids", "label": "alias2", "node_ids": ["a", "b", "c"]},
        ],
    }


def test_build_normalizes_member_aliases_to_nodes():
    G = build_from_json(_alias_extraction())
    hes = {he["id"]: he for he in G.graph["hyperedges"]}
    for hid in ("he_nodes", "he_members", "he_node_ids"):
        assert hes[hid]["nodes"] == ["a", "b", "c"], hid
        # alias keys are dropped post-normalization
        assert "members" not in hes[hid]
        assert "node_ids" not in hes[hid]


def test_build_dedups_alias_members_preserving_order():
    extraction = {
        "nodes": [
            {"id": "a", "label": "A", "file_type": "code", "source_file": "m.py"},
            {"id": "b", "label": "B", "file_type": "code", "source_file": "m.py"},
        ],
        "edges": [],
        "hyperedges": [{"id": "h", "label": "x", "members": ["a", "a", "b"]}],
    }
    G = build_from_json(extraction)
    assert G.graph["hyperedges"][0]["nodes"] == ["a", "b"]
    assert "members" not in G.graph["hyperedges"][0]


def test_build_canonical_nodes_wins_over_alias():
    extraction = {
        "nodes": [
            {"id": "a", "label": "A", "file_type": "code", "source_file": "m.py"},
            {"id": "b", "label": "B", "file_type": "code", "source_file": "m.py"},
            {"id": "x", "label": "X", "file_type": "code", "source_file": "m.py"},
        ],
        "edges": [],
        "hyperedges": [
            {"id": "h", "label": "x", "nodes": ["a", "b"], "members": ["x"]},
        ],
    }
    G = build_from_json(extraction)
    he = G.graph["hyperedges"][0]
    assert he["nodes"] == ["a", "b"]  # canonical untouched
    assert "members" not in he  # stray alias dropped


def test_build_rekeys_alias_keyed_hyperedge_members():
    """Alias normalization must run BEFORE the semantic id-remap loop so a
    `members`-keyed hyperedge's refs get rekeyed alongside `nodes`-keyed ones."""
    # Non-AST node whose id uses the OLD short stem (`mod_foo`) for source_file
    # pkg/mod.py -> new canonical stem pkg_mod -> remap mod_foo => pkg_mod_foo.
    extraction = {
        "nodes": [
            {"id": "mod_foo", "label": "foo", "file_type": "code", "source_file": "pkg/mod.py"},
            {"id": "mod_bar", "label": "bar", "file_type": "code", "source_file": "pkg/mod.py"},
        ],
        "edges": [],
        "hyperedges": [
            {"id": "h", "label": "x", "members": ["mod_foo", "mod_bar"]},
        ],
    }
    G = build_from_json(extraction)
    he = G.graph["hyperedges"][0]
    assert he["nodes"] == ["pkg_mod_foo", "pkg_mod_bar"]


def test_build_warns_once_per_aliased_hyperedge(capsys):
    build_from_json(_alias_extraction())
    err = capsys.readouterr().err
    # one warning each for the two alias hyperedges, none for the nodes-keyed one
    assert err.count("normalizing") == 2
    assert "he_members" in err and "members" in err
    assert "he_node_ids" in err and "node_ids" in err
    assert "he_nodes" not in err