Fix MCP scopes cache sentinel that broke tools/list.

Reuse a stable _UNSET sentinel and unwrap http_json envelopes so auth scope filtering works.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-08-03 18:50:40 +08:00
parent d2fb33ec0d
commit 66f26b738d
3 changed files with 49 additions and 10 deletions

View file

@ -9,12 +9,13 @@ from unittest.mock import patch
import pytest
from netx_mcp.http_tools import HTTP_MCP_TOOLS, call_http_tool
from netx_mcp.http_tools import HTTP_MCP_TOOLS, call_http_tool, tools_for_scopes
from netx_mcp.server import _fetch_scopes
def test_http_mcp_tool_list_has_expected_tools() -> None:
names = [str(t.get("name") or "") for t in HTTP_MCP_TOOLS]
assert len(names) == 13
assert len(names) == 14
assert "queryUmeAlarms" in names
assert "queryUmeAlarmsRaw" in names
assert "execManagedNe" in names
@ -81,6 +82,28 @@ def test_call_exec_managed_ne_respects_max_commands_env(monkeypatch: pytest.Monk
assert payload.get("error_code") == "too_many_commands"
def test_fetch_scopes_unwraps_http_json_envelope() -> None:
with patch("netx_mcp.server.http_json") as mock_http:
mock_http.return_value = {
"ok": True,
"data": {"scopes": ["ne:read", "alarms:read"], "user": {"username": "mcp"}},
}
assert _fetch_scopes() == ["ne:read", "alarms:read"]
def test_fetch_scopes_returns_none_on_http_failure() -> None:
with patch("netx_mcp.server.http_json") as mock_http:
mock_http.return_value = {"ok": False, "error": "netx_http_401"}
assert _fetch_scopes() is None
def test_tools_for_scopes_filters_by_granted() -> None:
names = {str(t.get("name") or "") for t in tools_for_scopes(["ne:read"])}
assert "listManagedNe" in names
assert "queryUmeAlarms" not in names
assert tools_for_scopes(None) == list(HTTP_MCP_TOOLS)
def test_stdio_initialize_and_tools_list() -> None:
proc = subprocess.Popen(
[sys.executable, "-m", "netx_mcp"],
@ -88,6 +111,8 @@ def test_stdio_initialize_and_tools_list() -> None:
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
)
assert proc.stdin and proc.stdout
init_req = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}) + "\n"
@ -102,8 +127,9 @@ def test_stdio_initialize_and_tools_list() -> None:
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"]
assert len(tools) == 13
assert len(tools) == 14
proc.terminate()
proc.wait(timeout=5)