Merge pull request #273 from MikeSviblov/fix/mcp-list-pending-status-filter
fix: list_pending_devices MCP tool leaked approved/hidden inventory rows
This commit is contained in:
+36
-1
@@ -134,6 +134,21 @@ def _build_tools() -> list[Tool]:
|
|||||||
"type": "object",
|
"type": "object",
|
||||||
"properties": {},
|
"properties": {},
|
||||||
}),
|
}),
|
||||||
|
Tool(name="list_inventory", description="List the full device inventory (everything scanned except user-hidden devices): both pending devices awaiting triage and already-approved devices. Each row carries a 'status' field. Use the optional 'status' filter to narrow the result.", inputSchema={
|
||||||
|
"type": "object",
|
||||||
|
"properties": {
|
||||||
|
"status": {"type": "string", "enum": ["all", "pending", "approved"], "default": "all", "description": "Filter inventory by status. 'all' returns pending + approved."},
|
||||||
|
},
|
||||||
|
}),
|
||||||
|
Tool(name="list_hidden_devices", description="List devices the user has hidden from the inventory. Hidden devices are excluded from list_pending_devices and list_inventory; use restore_device to bring one back to pending.", inputSchema={
|
||||||
|
"type": "object",
|
||||||
|
"properties": {},
|
||||||
|
}),
|
||||||
|
Tool(name="restore_device", description="Restore (un-hide) a previously hidden device, returning it to pending status so it reappears in the triage list. Use this to undo a hide_device action.", inputSchema={
|
||||||
|
"type": "object",
|
||||||
|
"required": ["id"],
|
||||||
|
"properties": {"id": {"type": "string"}},
|
||||||
|
}),
|
||||||
Tool(name="list_designs", description="List all designs (canvases) with their IDs and node/group/text counts", inputSchema={
|
Tool(name="list_designs", description="List all designs (canvases) with their IDs and node/group/text counts", inputSchema={
|
||||||
"type": "object",
|
"type": "object",
|
||||||
"properties": {},
|
"properties": {},
|
||||||
@@ -228,7 +243,27 @@ async def _dispatch(name: str, args: dict) -> dict:
|
|||||||
return await backend.get("/api/v1/nodes")
|
return await backend.get("/api/v1/nodes")
|
||||||
|
|
||||||
if name == "list_pending_devices":
|
if name == "list_pending_devices":
|
||||||
return await backend.get("/api/v1/scan/pending")
|
# Backend /scan/pending returns the whole inventory: approved rows stay
|
||||||
|
# listed so the frontend can show a canvas-presence badge. This tool
|
||||||
|
# promises only devices "not yet approved or hidden", so filter to
|
||||||
|
# actual pending rows (legacy rows without a status count as pending).
|
||||||
|
devices = await backend.get("/api/v1/scan/pending")
|
||||||
|
return [d for d in devices if d.get("status", "pending") == "pending"]
|
||||||
|
|
||||||
|
if name == "list_inventory":
|
||||||
|
# /scan/pending returns the whole inventory minus hidden rows (pending +
|
||||||
|
# approved). Legacy rows without a status field count as pending.
|
||||||
|
devices = await backend.get("/api/v1/scan/pending")
|
||||||
|
wanted = args.get("status", "all")
|
||||||
|
if wanted == "all":
|
||||||
|
return devices
|
||||||
|
return [d for d in devices if d.get("status", "pending") == wanted]
|
||||||
|
|
||||||
|
if name == "list_hidden_devices":
|
||||||
|
return await backend.get("/api/v1/scan/hidden")
|
||||||
|
|
||||||
|
if name == "restore_device":
|
||||||
|
return await backend.post(f"/api/v1/scan/pending/{args['id']}/restore", {})
|
||||||
|
|
||||||
if name == "list_designs":
|
if name == "list_designs":
|
||||||
return await backend.get("/api/v1/designs")
|
return await backend.get("/api/v1/designs")
|
||||||
|
|||||||
@@ -266,3 +266,62 @@ def test_create_design_schema_requires_name():
|
|||||||
async def test_unknown_tool():
|
async def test_unknown_tool():
|
||||||
with pytest.raises(ValueError, match="Unknown tool"):
|
with pytest.raises(ValueError, match="Unknown tool"):
|
||||||
await _dispatch("nonexistent", {})
|
await _dispatch("nonexistent", {})
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_list_pending_devices_filters_non_pending(mock_backend):
|
||||||
|
"""The tool promises devices *not yet approved or hidden*; the backend
|
||||||
|
endpoint returns the whole inventory including approved rows (they carry
|
||||||
|
the canvas-presence badge). The tool must filter to status == "pending"
|
||||||
|
and keep legacy rows that lack the field."""
|
||||||
|
mock_backend.get = AsyncMock(return_value=[
|
||||||
|
{"id": "p1", "ip": "192.168.1.50", "status": "pending"},
|
||||||
|
{"id": "a1", "ip": "192.168.1.60", "status": "approved"},
|
||||||
|
{"id": "h1", "ip": "192.168.1.70", "status": "hidden"},
|
||||||
|
{"id": "legacy", "ip": "192.168.1.80"},
|
||||||
|
])
|
||||||
|
result = await _dispatch("list_pending_devices", {})
|
||||||
|
assert [d["id"] for d in result] == ["p1", "legacy"]
|
||||||
|
|
||||||
|
|
||||||
|
_INVENTORY = [
|
||||||
|
{"id": "p1", "ip": "192.168.1.50", "status": "pending"},
|
||||||
|
{"id": "a1", "ip": "192.168.1.60", "status": "approved"},
|
||||||
|
{"id": "legacy", "ip": "192.168.1.80"},
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_list_inventory_default_returns_all(mock_backend):
|
||||||
|
"""No status filter returns the whole non-hidden inventory verbatim."""
|
||||||
|
mock_backend.get = AsyncMock(return_value=list(_INVENTORY))
|
||||||
|
result = await _dispatch("list_inventory", {})
|
||||||
|
mock_backend.get.assert_called_once_with("/api/v1/scan/pending")
|
||||||
|
assert [d["id"] for d in result] == ["p1", "a1", "legacy"]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_list_inventory_filters_approved(mock_backend):
|
||||||
|
mock_backend.get = AsyncMock(return_value=list(_INVENTORY))
|
||||||
|
result = await _dispatch("list_inventory", {"status": "approved"})
|
||||||
|
assert [d["id"] for d in result] == ["a1"]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_list_inventory_pending_keeps_legacy(mock_backend):
|
||||||
|
"""Legacy rows without a status field count as pending."""
|
||||||
|
mock_backend.get = AsyncMock(return_value=list(_INVENTORY))
|
||||||
|
result = await _dispatch("list_inventory", {"status": "pending"})
|
||||||
|
assert [d["id"] for d in result] == ["p1", "legacy"]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_list_hidden_devices(mock_backend):
|
||||||
|
await _dispatch("list_hidden_devices", {})
|
||||||
|
mock_backend.get.assert_called_once_with("/api/v1/scan/hidden")
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_restore_device(mock_backend):
|
||||||
|
await _dispatch("restore_device", {"id": "5"})
|
||||||
|
mock_backend.post.assert_called_once_with("/api/v1/scan/pending/5/restore", {})
|
||||||
|
|||||||
Reference in New Issue
Block a user