hansjone/netx
1
0
Fork 0
mirror of https://github.com/hansjone/netx.git synced 2026-10-16 01:50:45 +08:00
Code Issues Projects Releases 1 Packages Wiki Activity Actions
netx/packages/netx-topology-mcp/tests/test_mcp_topology.py
oliver ed820126df Fix the 5 failing netx-topology-mcp tests. 
Stale tests (behavior changed intentionally on 2026-08-12):
- Tool count is 16: suggestSinkHubs and classifyTopologyFabricNodes are
  documented public tools; assert them by name.
- compose_views is no longer a public layout action; assert it is
  rejected, and cover the background-job path via orbit_sweep instead.
- Dual-unit petals are nested ellipse bands with a hollow eye, not
  parallel lanes; assert side split, ordering inside the portals and
  zero crossings instead of portal-aligned lane ends.

Real bug (test failed since it was added on 2026-08-09):
press_hot_edges could never reel a long spoke in to its r_use target,
because max_disp (r0*0.55) is smaller than r0 - r_use. Inward pulls along
the spoke cannot fling a leaf across the bbox, so give them a cap of
r0 - r_use. On a 1483-node dev view crossings after the op drop
25501 -> 25346; a 500-node view is unchanged; runtime unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-12 22:16:37 +08:00

641 lines
25 KiB
Python
Raw Blame History

"""Tests for netx topology HTTP MCP server."""
from __future__ import annotations
import json
import subprocess
import sys
from unittest.mock import patch
from netx_topology_mcp.http_tools import (
HTTP_MCP_TOOLS,
call_http_tool,
resolve_draw_target,
tools_for_scopes,
)
from netx_topology_mcp.server import _fetch_scopes
def test_tool_list_has_draw_and_query_tools() -> None:
names = {str(t.get("name") or "") for t in HTTP_MCP_TOOLS}
assert len(names) == 16
assert "sinkTopologyDualUnits" in names
assert "suggestSinkHubs" in names
assert "classifyTopologyFabricNodes" in names
assert "copyTopologyViewNodes" in names
assert "createTopologyFolder" in names
assert "createTopologyView" not in names
assert "listTopologyViews" not in names
assert "queryTopologyFabricNodes" in names
assert "listTopologyFabricNodes" not in names
assert "searchTopologyFabricNodes" not in names
assert "getTopologyFabricSummary" not in names
assert "addTopologyViewNodes" in names
assert "updateTopologyViewPositions" in names
assert "queryTopologyEdges" in names
assert "analyzeTopologyViewLayout" in names
assert "layoutTopologyView" in names
assert "getTopologyTree" in names
assert "createTopologyManualEdge" not in names
assert "populateTopologyView" not in names
folder_tool = next(t for t in HTTP_MCP_TOOLS if t["name"] == "createTopologyFolder")
assert "Does not create a canvas" not in str(folder_tool.get("description") or "")
assert "view_id" in str(folder_tool.get("description") or "")
def test_create_folder_requires_name() -> None:
out = call_http_tool("createTopologyFolder", {})
assert out.get("isError") is True
payload = json.loads(out["content"][0]["text"])
assert payload["error"] == "name_required"
def test_create_folder_posts_region_and_resolves_view() -> None:
tree = {
"root": {
"id": "sys-root",
"kind": "root",
"children": [
{
"id": "f1",
"name": "华北",
"kind": "region",
"views": [],
"ne_count": 0,
"children": [
{
"id": "rm1",
"name": "根图",
"kind": "region",
"is_system": True,
"ne_count": 0,
"views": [{"id": "v-root-map", "name": "根图", "kind": "physical"}],
"children": [],
}
],
}
],
}
}
def fake_http(method: str, path: str, body: dict | None = None, **_kwargs):
if method == "POST" and path == "/v1/topology/folders":
return {"ok": True, "data": {"id": "f1", "name": "华北", "kind": "region"}}
if method == "GET" and path == "/v1/topology/tree":
return {"ok": True, "data": tree}
return {"ok": False, "error": f"unexpected {method} {path}"}
with patch("netx_topology_mcp.http_tools.http_json", side_effect=fake_http) as mock_http:
out = call_http_tool(
"createTopologyFolder",
{"name": "华北", "sort_order": 1, "locale": "zh"},
)
assert mock_http.call_count == 2
post = mock_http.call_args_list[0]
assert post[0][0] == "POST"
assert post[0][1] == "/v1/topology/folders"
body = post[1]["body"]
assert body["name"] == "华北"
assert body["kind"] == "region"
assert body["sort_order"] == 1
assert body["locale"] == "zh"
payload = json.loads(out["content"][0]["text"])
assert payload["ok"] is True
assert payload["id"] == "f1"
assert payload["view_id"] == "v-root-map"
assert payload["canvas_folder_id"] == "rm1"
assert "view_id" in str(payload.get("hint") or "")
def test_resolve_draw_target_nested_region() -> None:
tree = {
"root": {
"id": "sys",
"children": [
{
"id": "nav",
"views": [],
"children": [
{
"id": "rm",
"name": "Root map",
"is_system": True,
"views": [{"id": "v0", "kind": "physical"}],
"children": [
{
"id": "zone",
"ne_count": 3,
"views": [{"id": "vz", "kind": "physical"}],
"children": [],
}
],
}
],
}
],
}
}
tip = resolve_draw_target(tree, "zone")
assert tip["view_id"] == "vz"
assert tip["canvas_folder_id"] == "zone"
assert tip["ne_count"] == 3
def test_add_nodes_rejects_managed_ume_ids() -> None:
out = call_http_tool(
"addTopologyViewNodes",
{"view_id": "v1", "managed_ne_ids": ["m1"], "fabric_node_ids": ["f1"]},
)
assert out.get("isError") is True
payload = json.loads(out["content"][0]["text"])
assert payload["error"] == "fabric_nodes_only"
def test_add_nodes_posts_fabric_ids_only() -> None:
with patch("netx_topology_mcp.http_tools.http_json") as mock_http:
mock_http.return_value = {"ok": True, "data": {"ok": True, "added": 2, "return_graph": False}}
out = call_http_tool(
"addTopologyViewNodes",
{"view_id": "v1", "fabric_node_ids": ["f1", "f2"], "layout": "grid"},
)
body = mock_http.call_args[1]["body"]
assert body["fabric_node_ids"] == ["f1", "f2"]
assert body["managed_ne_ids"] == []
assert body["ume_ne_ids"] == []
assert body["return_graph"] is False
payload = json.loads(out["content"][0]["text"])
assert payload["ok"] is True
def test_add_nodes_filter_posts_without_ids() -> None:
with patch("netx_topology_mcp.http_tools.http_json") as mock_http:
mock_http.return_value = {
"ok": True,
"data": {"ok": True, "added": 10, "matched": 40, "next_offset": 10, "truncated": True},
}
out = call_http_tool(
"addTopologyViewNodes",
{"view_id": "v1", "keyword": "BJ-", "limit": 10, "offset": 0},
)
body = mock_http.call_args[1]["body"]
assert body["keyword"] == "BJ-"
assert body["fabric_node_ids"] == []
assert body["return_graph"] is False
payload = json.loads(out["content"][0]["text"])
assert payload["added"] == 10
assert payload["next_offset"] == 10
def test_update_positions_layout_filter() -> None:
with patch("netx_topology_mcp.http_tools.http_json") as mock_http:
mock_http.return_value = {"ok": True, "data": {"ok": True, "updated": 5}}
out = call_http_tool(
"updateTopologyViewPositions",
{"view_id": "v1", "layout": "grid", "keyword": "core", "origin_x": 0, "origin_y": 0},
)
body = mock_http.call_args[1]["body"]
assert body["layout"] == "grid"
assert body["keyword"] == "core"
assert body["return_graph"] is False
payload = json.loads(out["content"][0]["text"])
assert payload["updated"] == 5
def test_create_view_tool_removed() -> None:
try:
call_http_tool("createTopologyView", {"name": "map1", "folder_id": "f1"})
raise AssertionError("createTopologyView should be unregistered")
except ValueError as e:
assert "unknown tool" in str(e)
def test_query_edges_defaults_to_adjacency() -> None:
with patch("netx_topology_mcp.http_tools.http_json") as mock_http:
mock_http.return_value = {
"ok": True,
"data": {
"total": 2,
"page": 1,
"page_size": 100,
"items": [
{
"a_node_id": "A",
"b_node_id": "B",
"a_name": "ne-a",
"b_name": "ne-b",
"a_ip": "1.1.1.1",
"b_ip": "2.2.2.2",
"a_port": "p1",
"b_port": "p2",
},
{
"a_node_id": "B",
"b_node_id": "A",
"a_name": "ne-b",
"b_name": "ne-a",
"a_port": "p3",
"b_port": "p4",
},
],
},
}
out = call_http_tool("queryTopologyEdges", {"node_id": "A", "page_size": 100})
payload = json.loads(out["content"][0]["text"])
assert payload["ok"] is True
assert payload["detail"] == "adjacency"
assert payload["link_count"] == 1
assert payload["links"][0]["link_count"] == 2
assert "items" not in payload
assert payload["peer_count"] == 1
assert payload["peers"][0]["node_id"] == "B"
assert payload["peers"][0]["link_count"] == 2
assert payload["peers_complete"] is True
out_ports = call_http_tool(
"queryTopologyEdges", {"node_id": "A", "detail": "ports", "page_size": 100}
)
ports = json.loads(out_ports["content"][0]["text"])
assert ports["detail"] == "ports"
assert len(ports["items"]) == 2
def test_tools_for_scopes_filters_write() -> None:
read_only = {str(t.get("name") or "") for t in tools_for_scopes(["ne:read"])}
assert "queryTopologyEdges" in read_only
assert "queryTopologyFabricNodes" in read_only
assert "createTopologyFolder" not in read_only
assert "classifyTopologyFabricNodes" not in read_only
write = {str(t.get("name") or "") for t in tools_for_scopes(["ne:read", "ne:write"])}
assert "createTopologyView" not in write
assert "createTopologyFolder" in write
assert "classifyTopologyFabricNodes" in write
def test_classify_topology_fabric_nodes_match_tag() -> None:
with patch("netx_topology_mcp.http_tools.http_json") as mock_http:
mock_http.return_value = {
"ok": True,
"data": {
"pattern": "CORE",
"match_field": "name",
"total_matched": 1,
"samples": [{"id": "n1", "name": "CORE-1", "attrs": {"x": 1}}],
"fabric_node_ids": ["n1"],
},
}
out = call_http_tool(
"classifyTopologyFabricNodes",
{"action": "match", "pattern": "CORE", "match_field": "name"},
)
payload = json.loads(out["content"][0]["text"])
assert payload.get("action") == "match"
assert payload.get("fabric_node_ids") == ["n1"]
assert mock_http.call_args[0][:2] == ("POST", "/v1/topology/fabric/nodes/match")
mock_http.return_value = {
"ok": True,
"data": {"dry_run": True, "matched": 1, "updated": 0, "level": 1.0, "samples": []},
}
out = call_http_tool(
"classifyTopologyFabricNodes",
{
"action": "tag",
"fabric_node_ids": ["n1"],
"level": 1.0,
"dry_run": True,
},
)
payload = json.loads(out["content"][0]["text"])
assert payload.get("action") == "tag"
assert payload.get("dry_run") is True
assert mock_http.call_args[0][:2] == ("POST", "/v1/topology/fabric/nodes/tags/bulk")
assert mock_http.call_args[1]["body"]["level"] == 1.0
def test_query_fabric_nodes_modes() -> None:
with patch("netx_topology_mcp.http_tools.http_json") as mock_http:
mock_http.return_value = {"ok": True, "data": {"node_count": 3, "edge_count": 2}}
out = call_http_tool("queryTopologyFabricNodes", {"mode": "summary"})
payload = json.loads(out["content"][0]["text"])
assert payload.get("mode") == "summary"
assert mock_http.call_args[0][1] == "/v1/topology/fabric/summary"
mock_http.return_value = {
"ok": True,
"data": {"items": [{"id": "n1", "name": "A"}], "total": 1, "page": 1, "page_size": 50},
}
out = call_http_tool("queryTopologyFabricNodes", {"q": "A"})
payload = json.loads(out["content"][0]["text"])
assert payload.get("mode") == "search"
assert "/search" in mock_http.call_args[0][1]
out = call_http_tool(
"queryTopologyFabricNodes",
{"mode": "list", "keyword": "core", "role": "core"},
)
payload = json.loads(out["content"][0]["text"])
assert payload.get("mode") == "list"
assert mock_http.call_args[0][1] == "/v1/topology/fabric/nodes"
assert mock_http.call_args[1]["params"]["keyword"] == "core"
def test_get_topology_tree_compacts_by_default() -> None:
tree = {
"root": {
"id": "r",
"name": "Network",
"kind": "root",
"ne_count": 2,
"external_ref": "drop-me",
"views": [],
"children": [
{
"id": "a",
"name": "RootA",
"kind": "region",
"ne_count": 2,
"views": [],
"children": [
{
"id": "rm",
"name": "根图",
"kind": "region",
"is_system": True,
"ne_count": 2,
"views": [{"id": "v1", "name": "根图", "kind": "physical", "node_count": 2}],
"children": [
{
"id": "deep",
"name": "Deep",
"kind": "region",
"ne_count": 0,
"views": [],
"children": [],
}
],
}
],
}
],
}
}
with patch("netx_topology_mcp.http_tools.http_json") as mock_http:
mock_http.return_value = {"ok": True, "data": tree}
out = call_http_tool("getTopologyTree", {"max_depth": 1})
payload = json.loads(out["content"][0]["text"])
assert payload["compact"] is True
assert "external_ref" not in payload["root"]
child = payload["root"]["children"][0]
assert child["name"] == "RootA"
# depth 1: grandchildren collapsed
assert child["children_truncated"] == 1
assert child["children"] == []
def test_get_topology_view_defaults_to_summary() -> None:
graph = {
"view": {"id": "v1", "name": "根图", "folder_id": "f1", "kind": "physical", "node_count": 2},
"nodes": [
{"fabric_node_id": "n1", "name": "A", "ip": "1.1.1.1", "x": 1, "y": 2},
{"fabric_node_id": "n2", "name": "B", "ip": "2.2.2.2", "x": 3, "y": 4},
],
"edges": [{"id": "e1", "a_node_id": "n1", "b_node_id": "n2", "display_label": "huge" * 50}],
"outside_peers": [{"fabric_node_id": "n3"}],
"truncated": False,
}
with patch("netx_topology_mcp.http_tools.http_json") as mock_http:
mock_http.return_value = {"ok": True, "data": graph}
out = call_http_tool("getTopologyView", {"view_id": "v1"})
payload = json.loads(out["content"][0]["text"])
assert payload["detail"] == "summary"
assert payload["node_count"] == 2
assert payload["edge_count"] == 1
assert payload["link_count"] == 1
assert payload["links"][0]["a_node_id"] == "n1"
assert payload["links"][0]["b_node_id"] == "n2"
assert "edges" not in payload
assert len(payload["sample_nodes"]) == 2
out_full = call_http_tool("getTopologyView", {"view_id": "v1", "detail": "full"})
full = json.loads(out_full["content"][0]["text"])
assert full["detail"] == "full"
assert len(full["edges"]) == 1
def test_project_neighbors_defaults_to_summary() -> None:
graph = {
"view": {"id": "v1", "name": "根图", "node_count": 3},
"nodes": [{"fabric_node_id": "n1", "name": "A", "ip": "", "x": 0, "y": 0}],
"edges": [],
"outside_peers": [],
"truncated": True,
"truncate_reason": "membership_cap",
}
with patch("netx_topology_mcp.http_tools.http_json") as mock_http:
mock_http.return_value = {"ok": True, "data": graph}
out = call_http_tool("projectTopologyNeighbors", {"view_id": "v1"})
payload = json.loads(out["content"][0]["text"])
assert payload["detail"] == "summary"
assert payload["projected"] is True
assert payload["truncated"] is True
assert "edges" not in payload
def test_project_neighbors_passes_region_folder_and_reports_skips() -> None:
graph = {
"view": {"id": "v1", "name": "根图", "node_count": 2},
"nodes": [{"fabric_node_id": "n1", "name": "A", "ip": "", "x": 0, "y": 0}],
"edges": [],
"outside_peers": [],
"out_of_region_skipped": 3,
"out_of_region_sample": [
{"fabric_node_id": "x1", "name": "X", "region_folder_id": "other"}
],
}
with patch("netx_topology_mcp.http_tools.http_json") as mock_http:
mock_http.return_value = {"ok": True, "data": graph}
out = call_http_tool(
"projectTopologyNeighbors",
{"view_id": "v1", "region_folder_id": "reg-A"},
)
payload = json.loads(out["content"][0]["text"])
assert payload["projected"] is True
assert payload["region_folder_id"] == "reg-A"
assert payload["out_of_region_skipped"] == 3
assert payload["out_of_region_sample"][0]["fabric_node_id"] == "x1"
assert mock_http.call_args.kwargs["body"]["region_folder_id"] == "reg-A"
def test_neighborhood_returns_links_not_ports() -> None:
with patch("netx_topology_mcp.http_tools.http_json") as mock_http:
mock_http.return_value = {
"ok": True,
"data": {
"center_node_id": "A",
"depth": 1,
"nodes": [
{
"id": "A",
"name": "ne-a",
"ip": "1.1.1.1",
"attrs": {"sources": ["ume"], "huge": "x" * 200},
"world_x": 1,
"world_y": 2,
}
],
"edges": [
{
"id": "e1",
"a_node_id": "A",
"b_node_id": "B",
"source": "ume",
"status": "active",
"a_port": "p1",
"attrs": {"display_label": "LONG" * 40},
},
{
"id": "e2",
"a_node_id": "A",
"b_node_id": "B",
"source": "ume",
"status": "active",
"a_port": "p2",
},
],
},
}
out = call_http_tool("queryTopologyNeighborhood", {"node_id": "A"})
payload = json.loads(out["content"][0]["text"])
assert payload["node_count"] == 1
assert payload["edge_count"] == 2
assert payload["link_count"] == 1
assert payload["links"][0]["link_count"] == 2
assert "edges" not in payload
assert "attrs" not in payload["nodes"][0]
assert "world_x" not in payload["nodes"][0]
def test_fetch_scopes_unwraps_envelope() -> None:
with patch("netx_topology_mcp.server.http_json") as mock_http:
mock_http.return_value = {"ok": True, "data": {"scopes": ["ne:read", "ne:write"]}}
assert _fetch_scopes() == ["ne:read", "ne:write"]
def test_stdio_initialize_and_tools_list() -> None:
import os
# Force scopes fetch to fail so tools/list returns the full catalog.
env = os.environ.copy()
env["NETX_API_URL"] = "http://127.0.0.1:1"
env.pop("NETX_API_TOKEN", None)
proc = subprocess.Popen(
[sys.executable, "-m", "netx_topology_mcp"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
env=env,
)
assert proc.stdin and proc.stdout
try:
proc.stdin.write(json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}) + "\n")
proc.stdin.flush()
init_resp = json.loads(proc.stdout.readline())
assert init_resp["result"]["serverInfo"]["name"] == "netx-topology-mcp"
proc.stdin.write(json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}) + "\n")
proc.stdin.flush()
list_resp = json.loads(proc.stdout.readline())
tools = list_resp["result"]["tools"]
assert len(tools) == 16
names = {t["name"] for t in tools}
assert "suggestSinkHubs" in names
assert "classifyTopologyFabricNodes" in names
assert "createTopologyView" not in names
assert "listTopologyViews" not in names
assert "queryTopologyFabricNodes" in names
assert "createTopologyFolder" in names
assert "analyzeTopologyViewLayout" in names
assert "layoutTopologyView" in names
assert "sinkTopologyDualUnits" in names
assert "copyTopologyViewNodes" in names
finally:
proc.terminate()
proc.wait(timeout=5)
def test_layout_metrics_crossing_and_spacing() -> None:
from netx_topology_mcp.layout_metrics import analyze_positions, grade_layout, segments_properly_intersect
assert segments_properly_intersect((0, 0), (2, 2), (0, 2), (2, 0))
assert not segments_properly_intersect((0, 0), (1, 0), (1, 0), (2, 0)) # shared endpoint
# X crossing
nodes = [
{"fabric_node_id": "a", "name": "A", "x": 0, "y": 0},
{"fabric_node_id": "b", "name": "B", "x": 200, "y": 200},
{"fabric_node_id": "c", "name": "C", "x": 0, "y": 200},
{"fabric_node_id": "d", "name": "D", "x": 200, "y": 0},
]
edges = [
{"a_node_id": "a", "b_node_id": "b"},
{"a_node_id": "c", "b_node_id": "d"},
]
m = analyze_positions(nodes, edges)
assert m["edge_crossings"] == 1
assert m["link_count"] == 2
g = grade_layout(m)
assert g["overall"] in {"ok", "warn", "fail"}
def test_analyze_topology_view_layout_single() -> None:
graph = {
"view": {"id": "v1", "name": "demo", "folder_id": "f1"},
"nodes": [
{"fabric_node_id": "a", "name": "N-A", "x": 0, "y": 0},
{"fabric_node_id": "b", "name": "N-B", "x": 200, "y": 0},
{"fabric_node_id": "c", "name": "N-C", "x": 0, "y": 200},
{"fabric_node_id": "d", "name": "N-D", "x": 200, "y": 200},
],
"edges": [
{"a_node_id": "a", "b_node_id": "b"},
{"a_node_id": "b", "b_node_id": "d"},
{"a_node_id": "d", "b_node_id": "c"},
{"a_node_id": "c", "b_node_id": "a"},
],
}
with patch("netx_topology_mcp.http_tools.http_json", return_value={"ok": True, "data": graph}):
out = call_http_tool("analyzeTopologyViewLayout", {"view_id": "v1"})
assert out.get("isError") is not True
payload = json.loads(out["content"][0]["text"])
assert payload["ok"] is True
assert payload["size"]["links"] == 4
assert payload["crossing"]["edge_crossings"] == 0
assert payload["overlap"]["status"] == "ok"
assert "sparsity" in payload
assert "verdict" in payload
assert "total" in payload["verdict"]
def test_update_positions_omits_label_and_locked_unless_given() -> None:
from netx_topology_mcp.http_tools import _update_topology_view_positions
with patch("netx_topology_mcp.http_tools.http_json") as mock_http:
mock_http.return_value = {"ok": True, "data": {"ok": True, "updated": 3}}
_update_topology_view_positions(
{
"view_id": "v1",
"positions": [
{"fabric_node_id": "a", "x": 1, "y": 2},
{"fabric_node_id": "b", "x": 3, "y": 4, "label": "", "locked": False},
{"fabric_node_id": "c", "x": 5, "y": 6, "label": "Core-1", "locked": True},
],
}
)
sent = mock_http.call_args[1]["body"]["positions"]
assert sent[0] == {"fabric_node_id": "a", "x": 1.0, "y": 2.0}
assert sent[1]["label"] == "" and sent[1]["locked"] is False
assert sent[2]["label"] == "Core-1" and sent[2]["locked"] is True
Reference in a new issue View git blame Copy permalink