mirror of
https://github.com/hansjone/netx.git
synced 2026-10-18 04:00:47 +08:00
oliver
9085cac742
- New packages/netx-bizmonitor-mcp: 8 read-only cutover/biz-state tools mirroring the netxops bizMonitor group. - Align netx-mcp (15) and netx-topology-mcp (16) with backend HTTP contracts; surface partial batch failures as tool errors. - CLI hard gate: single-line show/display only; exec job IDs, path containment, owner and ne:exec checks. - Cisco IOSv collection: stable prompt/echo handling over CR. - Remove legacy netx_api/mcp and mcp_server.py; add mcp.all.json; sync skills and docs. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
"""Tests for netx HTTP MCP (via netx-mcp package)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from netx_mcp.http_tools import HTTP_MCP_TOOLS, call_http_tool
|
|
|
|
|
|
def test_http_mcp_tool_list_has_fifteen_tools() -> None:
|
|
names = [str(t.get("name") or "") for t in HTTP_MCP_TOOLS]
|
|
assert len(names) == 15
|
|
assert "queryNmsAlarms" in names
|
|
assert "queryNmsAlarmsRaw" in names
|
|
assert "execManagedNe" in names
|
|
assert "listCliTargets" in names
|
|
|
|
|
|
def test_call_query_nms_alarms_forwards_http() -> None:
|
|
with patch("netx_mcp.http_tools.http_json") as mock_http:
|
|
mock_http.return_value = {"ok": True, "data": {"total": 0, "items": []}}
|
|
out = call_http_tool("queryNmsAlarms", {"severity": "critical", "page": 1, "page_size": 10})
|
|
mock_http.assert_called_once()
|
|
assert mock_http.call_args[0][0] == "GET"
|
|
assert mock_http.call_args[0][1] == "/v1/ume/alarms"
|
|
params = mock_http.call_args[1]["params"]
|
|
assert params["severity"] == "critical"
|
|
text = out["content"][0]["text"]
|
|
payload = json.loads(text)
|
|
assert payload["ok"] is True
|
|
|
|
|
|
def test_call_get_nms_ne_requires_id() -> None:
|
|
out = call_http_tool("getNmsNe", {})
|
|
assert out.get("isError") is True
|
|
text = out["content"][0]["text"]
|
|
payload = json.loads(text)
|
|
assert payload["error"] == "ne_id_required"
|
|
|
|
|
|
def test_call_exec_managed_ne_posts_body() -> None:
|
|
with patch("netx_mcp.http_tools.http_post_json") as mock_post:
|
|
mock_post.return_value = {"ok": True, "data": {"ok": True, "output": "ok"}}
|
|
out = call_http_tool(
|
|
"execManagedNe",
|
|
{"ne_id": "abc", "commands": ["show version"]},
|
|
)
|
|
mock_post.assert_called_once()
|
|
assert mock_post.call_args[0][0] == "/v1/managed-ne/exec"
|
|
body = mock_post.call_args[0][1]
|
|
assert body["ne_id"] == "abc"
|
|
assert body["commands"] == ["show version"]
|
|
text = out["content"][0]["text"]
|
|
payload = json.loads(text)
|
|
assert payload["ok"] is True
|
|
|
|
|
|
def test_stdio_initialize_and_tools_list() -> None:
|
|
import os
|
|
|
|
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_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
|
|
init_req = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}) + "\n"
|
|
proc.stdin.write(init_req)
|
|
proc.stdin.flush()
|
|
init_line = proc.stdout.readline()
|
|
init_resp = json.loads(init_line)
|
|
assert init_resp["result"]["serverInfo"]["mode"] == "http"
|
|
|
|
list_req = json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}) + "\n"
|
|
proc.stdin.write(list_req)
|
|
proc.stdin.flush()
|
|
list_line = proc.stdout.readline()
|
|
list_resp = json.loads(list_line)
|
|
assert "error" not in list_resp, list_resp
|
|
tools = list_resp["result"]["tools"]
|
|
names = {t["name"] for t in tools}
|
|
assert len(tools) == 15
|
|
assert "queryNmsAlarms" in names
|
|
assert "findTopologyPaths" in names
|
|
|
|
proc.terminate()
|
|
proc.wait(timeout=5)
|
|
|
|
|
|
|
|
def test_only_standalone_mcp_implementation_is_kept() -> None:
|
|
from pathlib import Path
|
|
|
|
root = Path(__file__).resolve().parents[1]
|
|
assert not (root / "netx_api/mcp_server.py").exists()
|
|
assert not list((root / "netx_api/mcp").glob("*.py"))
|