netx/packages/netx-topology-mcp/tests/test_layout_structure.py

"""Structure / gravity detection for layout planning."""

from __future__ import annotations

from netx_topology_mcp.layout_structure import analyze_graph_structure


def _nodes_edges_core_bar():
    """Two cores own access rings; ANs are low-degree decorative."""
    nodes = [
        {"fabric_node_id": "c1", "name": "X-CN1-a", "role": "core", "x": 0, "y": 0},
        {"fabric_node_id": "c2", "name": "X-CN2-a", "role": "core", "x": 200, "y": 0},
        {"fabric_node_id": "a1", "name": "X-AN1-a", "role": "aggregation", "x": 100, "y": 50},
        {"fabric_node_id": "a2", "name": "X-AN2-a", "role": "aggregation", "x": 300, "y": 50},
    ]
    edges = [
        {"a_node_id": "c1", "b_node_id": "c2"},
        {"a_node_id": "c1", "b_node_id": "a1"},
        {"a_node_id": "c2", "b_node_id": "a2"},
    ]
    # Access petals hanging on cores
    for i in range(12):
        eid = f"e{i}"
        nodes.append(
            {
                "fabric_node_id": eid,
                "name": f"X-EN{i}-a",
                "role": "access",
                "x": i * 40,
                "y": 200,
            }
        )
        hub = "c1" if i < 6 else "c2"
        edges.append({"a_node_id": hub, "b_node_id": eid})
        if i % 2 == 1:
            edges.append({"a_node_id": f"e{i - 1}", "b_node_id": eid})
    return nodes, edges


def _nodes_edges_agg_bar():
    """No cores; ANs own access rings."""
    nodes = [
        {"fabric_node_id": "a1", "name": "Y-AN1-a", "role": "aggregation", "x": 0, "y": 0},
        {"fabric_node_id": "a2", "name": "Y-AN2-a", "role": "aggregation", "x": 400, "y": 0},
    ]
    edges = [{"a_node_id": "a1", "b_node_id": "a2"}]
    for i in range(16):
        eid = f"e{i}"
        nodes.append(
            {
                "fabric_node_id": eid,
                "name": f"Y-EN{i}-a",
                "role": "access",
                "x": i * 30,
                "y": 200,
            }
        )
        hub = "a1" if i < 8 else "a2"
        edges.append({"a_node_id": hub, "b_node_id": eid})
        if i % 2 == 1:
            edges.append({"a_node_id": f"e{i - 1}", "b_node_id": eid})
    return nodes, edges


def test_core_bar_gravity() -> None:
    nodes, edges = _nodes_edges_core_bar()
    s = analyze_graph_structure(nodes, edges)
    assert s["gravity"]["type"] == "core_bar"
    assert s["gravity"]["anchor_layer"] == "core"
    assert "agg" in s["gravity"]["decorative_layers"] or s["layers"]["agg"]["territory_frac"] < 0.2
    assert s["gravity"]["recipe_preference"][0] in {"compact", "corridor"}
    assert s["advice"]["skip_rings_first"] is True
    assert s["gravity"]["geometry_hint"] in {"core_center", "core_top"}


def test_agg_bar_gravity() -> None:
    nodes, edges = _nodes_edges_agg_bar()
    s = analyze_graph_structure(nodes, edges)
    assert s["gravity"]["type"] == "agg_bar"
    assert s["gravity"]["anchor_layer"] == "agg"
    assert s["gravity"]["recipe_preference"][0] == "rings"
    assert s["advice"]["skip_rings_first"] is False
    assert len(s["hubs"]) >= 2


def test_hubs_include_stubs() -> None:
    nodes, edges = _nodes_edges_agg_bar()
    s = analyze_graph_structure(nodes, edges, hub_top_k=4, stub_top_k=10)
    assert s["hubs"][0]["access_neighbors"] >= 1
    assert s["stubs"]
    assert "headline" in s


def _nodes_edges_chains():
    """Several path components (SID-TOB-like): mostly deg≤2, not hub-spoke."""
    nodes: list[dict] = []
    edges: list[dict] = []
    # three chains of lengths 12 / 8 / 5; one has a CN/AN on the path (still chain)
    chains = [
        [("c0", "Z-CN1-a", "core")]
        + [(f"a{i}", f"Z-EN{i}-a", "access") for i in range(1, 12)],
        [(f"b{i}", f"Y-EN{i}-a", "access") for i in range(8)],
        [(f"d{i}", f"X-EN{i}-a", "access") for i in range(5)],
    ]
    for ch in chains:
        for fid, name, role in ch:
            nodes.append({"fabric_node_id": fid, "name": name, "role": role, "x": 0, "y": 0})
        for i in range(len(ch) - 1):
            edges.append({"a_node_id": ch[i][0], "b_node_id": ch[i + 1][0]})
    return nodes, edges


def test_chains_gravity_before_hub_fallback() -> None:
    nodes, edges = _nodes_edges_chains()
    s = analyze_graph_structure(nodes, edges)
    assert s["shape"]["primary"] == "chains"
    assert s["gravity"]["type"] == "chains"
    assert s["gravity"]["geometry_hint"] == "chain_rows"
    assert s["gravity"]["recipe_preference"][0] == "corridor"
    assert s["advice"]["skip_rings_first"] is True
    assert s["advice"]["decompose_by_component"] is True
    assert s["shape"]["component_count"] == 3
    assert all(b["shape"] in {"chain", "tiny"} for b in s["shape"]["blocks"])