mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-10-12 21:45:41 +03:00
feat(mcp): accept project/node names as tool locators
LLM agents transcribe UUIDs unreliably — a 2-character slip invents an id that never existed and is invisible to the eye. Tool parameters now accept names wherever the server guarantees uniqueness: project names among live projects, node names within a project. Links, drawings, appliances and snapshots have no names and stay UUID-only. Resolution sits in the single dispatch choke point (_run_handler_sync): UUID-shaped values pass through untouched, names resolve with one list call, and an unknown name fails fast listing the known names instead of a confusing downstream 404. As a backstop for residual UUID typos, error replies carrying "<UUID> doesn't exist" are matched against the in-scope ids; a single candidate within edit distance 2 earns a "Did you mean ...?" hint.
This commit is contained in:
YueGuobin
parent
935c00802f
commit
5ac1521217
@ -122,6 +122,7 @@ from .images import (
|
||||
install_images_handler,
|
||||
prune_images_handler,
|
||||
)
|
||||
from .locators import add_did_you_mean, resolve_locators
|
||||
from .projects import (
|
||||
close_project_handler,
|
||||
create_project_handler,
|
||||
@ -339,7 +340,19 @@ def _run_handler_sync(handler, params: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
"jwt_username": _jwt_username_var.get(),
|
||||
"jwt_token_version": _jwt_token_version_var.get(),
|
||||
}
|
||||
result = handler(params, ctx)
|
||||
|
||||
def run(h, p):
|
||||
return h(p, ctx)
|
||||
|
||||
# Name-shaped project_id/node_id values become UUIDs before the handler
|
||||
# runs; a name that matches nothing fails fast with the known names.
|
||||
params, resolution_error = resolve_locators(params, run)
|
||||
if resolution_error is not None:
|
||||
result = resolution_error
|
||||
else:
|
||||
result = handler(params, ctx)
|
||||
# backstop for UUID transcription slips: "Did you mean ...?"
|
||||
result = add_did_you_mean(result, params, run)
|
||||
return [{"type": "text", "text": json.dumps(result, ensure_ascii=False, default=str)}]
|
||||
|
||||
|
||||
@ -351,7 +364,7 @@ async def project_list() -> list[dict[str, Any]]:
|
||||
|
||||
@mcp.tool()
|
||||
async def project_get(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Get detailed information about a specific project."""
|
||||
return await asyncio.to_thread(_run_handler_sync, get_project_handler, {"project_id": project_id})
|
||||
@ -368,7 +381,7 @@ async def project_create(
|
||||
|
||||
@mcp.tool()
|
||||
async def project_delete(
|
||||
project_id: Annotated[str, Field(description="UUID of the project to delete")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID to delete")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Delete a GNS3 project permanently."""
|
||||
return await asyncio.to_thread(_run_handler_sync, delete_project_handler, {"project_id": project_id})
|
||||
@ -376,7 +389,7 @@ async def project_delete(
|
||||
|
||||
@mcp.tool()
|
||||
async def project_open(
|
||||
project_id: Annotated[str, Field(description="UUID of the project to open")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID to open")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Open a closed GNS3 project."""
|
||||
return await asyncio.to_thread(_run_handler_sync, open_project_handler, {"project_id": project_id})
|
||||
@ -384,7 +397,7 @@ async def project_open(
|
||||
|
||||
@mcp.tool()
|
||||
async def project_close(
|
||||
project_id: Annotated[str, Field(description="UUID of the project to close")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID to close")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Close an open GNS3 project."""
|
||||
return await asyncio.to_thread(_run_handler_sync, close_project_handler, {"project_id": project_id})
|
||||
@ -392,7 +405,7 @@ async def project_close(
|
||||
|
||||
@mcp.tool()
|
||||
async def project_stats(
|
||||
project_id: Annotated[str, Field(description="UUID of the project to get statistics for")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID to get statistics for")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Get statistics (nodes, links, snapshots, drawings) for a project."""
|
||||
return await asyncio.to_thread(_run_handler_sync, get_project_stats_handler, {"project_id": project_id})
|
||||
@ -400,7 +413,7 @@ async def project_stats(
|
||||
|
||||
@mcp.tool()
|
||||
async def project_update(
|
||||
project_id: Annotated[str, Field(description="UUID of the project to update")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID to update")],
|
||||
name: Annotated[str | None, Field(description="New project name")] = None,
|
||||
auto_close: Annotated[bool | None, Field(description="Close project when last client leaves")] = None,
|
||||
auto_open: Annotated[bool | None, Field(description="Project opens when GNS3 starts")] = None,
|
||||
@ -442,7 +455,7 @@ async def project_update(
|
||||
|
||||
@mcp.tool()
|
||||
async def project_duplicate(
|
||||
project_id: Annotated[str, Field(description="UUID of the project to duplicate")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID to duplicate")],
|
||||
name: Annotated[str, Field(description="New project name")],
|
||||
reset_mac_addresses: Annotated[bool, Field(description="Reset MAC addresses for this project")] = False,
|
||||
) -> list[dict[str, Any]]:
|
||||
@ -455,7 +468,7 @@ async def project_duplicate(
|
||||
|
||||
@mcp.tool()
|
||||
async def project_readme_get(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Get the content of a project's README.md file — the project documentation (Markdown format)."""
|
||||
return await asyncio.to_thread(_run_handler_sync, get_project_readme_handler, {"project_id": project_id})
|
||||
@ -463,7 +476,7 @@ async def project_readme_get(
|
||||
|
||||
@mcp.tool()
|
||||
async def project_readme_update(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
content: Annotated[str, Field(description="Content to write to README.md (Markdown format)")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Update or create a project's README.md file — the project documentation (Markdown format)."""
|
||||
@ -477,7 +490,7 @@ async def project_readme_update(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_list(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
fields: Annotated[
|
||||
list[str] | None,
|
||||
Field(
|
||||
@ -491,8 +504,8 @@ async def node_list(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_get(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
node_id: Annotated[str, Field(description="Node name or UUID")],
|
||||
fields: Annotated[
|
||||
list[str] | None,
|
||||
Field(
|
||||
@ -514,8 +527,8 @@ async def node_get(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_start(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str | None, Field(description="Node UUID (single mode)")] = None,
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
node_id: Annotated[str | None, Field(description="Node name or UUID (single mode)")] = None,
|
||||
node_ids: Annotated[
|
||||
list[str] | None, Field(description='Batch mode: ["uuid1","uuid2"] — start multiple nodes in parallel')
|
||||
] = None,
|
||||
@ -531,8 +544,8 @@ async def node_start(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_stop(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str | None, Field(description="Node UUID (single mode)")] = None,
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
node_id: Annotated[str | None, Field(description="Node name or UUID (single mode)")] = None,
|
||||
node_ids: Annotated[
|
||||
list[str] | None, Field(description='Batch mode: ["uuid1","uuid2"] — stop multiple nodes in parallel')
|
||||
] = None,
|
||||
@ -548,8 +561,8 @@ async def node_stop(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_suspend(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str | None, Field(description="Node UUID (single mode)")] = None,
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
node_id: Annotated[str | None, Field(description="Node name or UUID (single mode)")] = None,
|
||||
node_ids: Annotated[
|
||||
list[str] | None, Field(description='Batch mode: ["uuid1","uuid2"] — suspend multiple nodes in parallel')
|
||||
] = None,
|
||||
@ -565,7 +578,7 @@ async def node_suspend(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_create(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
template_id: Annotated[
|
||||
str | None, Field(description="Template UUID (required for single mode; used as default in batch mode)")
|
||||
] = None,
|
||||
@ -626,8 +639,8 @@ async def node_create(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_delete(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str | None, Field(description="Node UUID (single mode)")] = None,
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
node_id: Annotated[str | None, Field(description="Node name or UUID (single mode)")] = None,
|
||||
node_ids: Annotated[
|
||||
list[str] | None, Field(description='Batch mode: ["uuid1","uuid2"] — delete multiple nodes in parallel')
|
||||
] = None,
|
||||
@ -643,8 +656,8 @@ async def node_delete(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_update(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node to update")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
node_id: Annotated[str, Field(description="Node name or UUID to update")],
|
||||
**kwargs: Any,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Update a node's properties (name, position, etc.)."""
|
||||
@ -654,8 +667,8 @@ async def node_update(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_console(
|
||||
project_id: Annotated[str, Field(description="UUID of the project containing the node")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node to get console info for")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID containing the node")],
|
||||
node_id: Annotated[str, Field(description="Node name or UUID to get console info for")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Get WebSocket console connection info for a node.
|
||||
|
||||
@ -693,7 +706,7 @@ async def node_console(
|
||||
|
||||
@mcp.tool()
|
||||
async def link_list(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
fields: Annotated[
|
||||
list[str] | None,
|
||||
Field(
|
||||
@ -707,7 +720,7 @@ async def link_list(
|
||||
|
||||
@mcp.tool()
|
||||
async def link_get(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
link_id: Annotated[str, Field(description="UUID of the link")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Get detailed information about a specific link."""
|
||||
@ -716,7 +729,7 @@ async def link_get(
|
||||
|
||||
@mcp.tool()
|
||||
async def link_create(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
nodes: Annotated[
|
||||
list | None,
|
||||
Field(description="Single mode: [{node_id, adapter_number, port_number}] or compact [id, ad, pt, id, ad, pt]"),
|
||||
@ -762,7 +775,7 @@ async def link_create(
|
||||
|
||||
@mcp.tool()
|
||||
async def link_delete(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
link_id: Annotated[str | None, Field(description="Link UUID (single mode)")] = None,
|
||||
link_ids: Annotated[
|
||||
list[str] | None, Field(description='Batch mode: ["uuid1","uuid2"] — delete multiple links in parallel')
|
||||
@ -779,7 +792,7 @@ async def link_delete(
|
||||
|
||||
@mcp.tool()
|
||||
async def link_update(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
link_id: Annotated[str, Field(description="UUID of the link to update")],
|
||||
**kwargs: Any,
|
||||
) -> list[dict[str, Any]]:
|
||||
@ -841,7 +854,7 @@ async def link_update(
|
||||
|
||||
@mcp.tool()
|
||||
async def link_available_filters(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
link_id: Annotated[str, Field(description="UUID of the link")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""List the packet filter types available for a link with their parameters
|
||||
@ -1002,8 +1015,8 @@ async def compute_images(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_file_list(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
node_id: Annotated[str, Field(description="Node name or UUID")],
|
||||
path: Annotated[str, Field(description="Subdirectory path within node directory (optional)")] = "",
|
||||
recursive: Annotated[bool, Field(description="Recursively list all files (optional, default: false)")] = False,
|
||||
) -> list[dict[str, Any]]:
|
||||
@ -1026,8 +1039,8 @@ async def node_file_list(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_file_get(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
node_id: Annotated[str, Field(description="Node name or UUID")],
|
||||
file_path: Annotated[str, Field(description="Path to the file within the node directory")],
|
||||
offset: Annotated[int, Field(description="Line offset to start reading from (optional, default: 0)")] = 0,
|
||||
limit: Annotated[int, Field(description="Maximum number of lines to return (optional, default: 200)")] = 200,
|
||||
@ -1056,8 +1069,8 @@ async def node_file_get(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_file_write(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
node_id: Annotated[str, Field(description="Node name or UUID")],
|
||||
file_path: Annotated[str, Field(description="Path to the file within the node directory")],
|
||||
content: Annotated[str, Field(description="Content to write to the file")],
|
||||
) -> list[dict[str, Any]]:
|
||||
@ -1076,8 +1089,8 @@ async def node_file_write(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_file_delete(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
node_id: Annotated[str, Field(description="Node name or UUID")],
|
||||
file_path: Annotated[str, Field(description="Path to the file within the node directory")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Delete a file from a node directory. Cannot be undone."""
|
||||
@ -1097,7 +1110,7 @@ async def node_file_delete(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_start_all(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Start all nodes in a project."""
|
||||
return await asyncio.to_thread(
|
||||
@ -1111,7 +1124,7 @@ async def node_start_all(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_stop_all(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Stop all nodes in a project."""
|
||||
return await asyncio.to_thread(
|
||||
@ -1125,7 +1138,7 @@ async def node_stop_all(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_suspend_all(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Suspend all nodes in a project."""
|
||||
return await asyncio.to_thread(
|
||||
@ -1139,8 +1152,8 @@ async def node_suspend_all(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_duplicate(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node to duplicate")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
node_id: Annotated[str, Field(description="Node name or UUID to duplicate")],
|
||||
x: Annotated[int, Field(description="X coordinate for the new node")] = 0,
|
||||
y: Annotated[int, Field(description="Y coordinate for the new node")] = 0,
|
||||
z: Annotated[int, Field(description="Z layer for the new node")] = 0,
|
||||
@ -1161,8 +1174,8 @@ async def node_duplicate(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_isolate(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node to isolate")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
node_id: Annotated[str, Field(description="Node name or UUID to isolate")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Isolate a node by suspending all its attached links (network isolation)."""
|
||||
return await asyncio.to_thread(
|
||||
@ -1177,8 +1190,8 @@ async def node_isolate(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_unisolate(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node to unisolate")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
node_id: Annotated[str, Field(description="Node name or UUID to unisolate")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Un-isolate a node by resuming all its suspended links."""
|
||||
return await asyncio.to_thread(
|
||||
@ -1193,8 +1206,8 @@ async def node_unisolate(
|
||||
|
||||
@mcp.tool()
|
||||
async def node_links(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
node_id: Annotated[str, Field(description="Node name or UUID")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""List all links connected to a specific node."""
|
||||
return await asyncio.to_thread(
|
||||
@ -1212,7 +1225,7 @@ async def node_links(
|
||||
|
||||
@mcp.tool()
|
||||
async def link_reset(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
link_id: Annotated[str | None, Field(description="Link UUID (single mode)")] = None,
|
||||
link_ids: Annotated[
|
||||
list[str] | None, Field(description='Batch mode: ["uuid1","uuid2"] — reset multiple links in parallel')
|
||||
@ -1239,7 +1252,7 @@ async def link_reset(
|
||||
|
||||
@mcp.tool()
|
||||
async def link_capture_start(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
link_id: Annotated[str | None, Field(description="Link UUID (single mode)")] = None,
|
||||
data_link_type: Annotated[str, Field(description="Data link type (default: DLT_EN10MB)")] = "DLT_EN10MB",
|
||||
capture_file_name: Annotated[str | None, Field(description="Capture file name (optional)")] = None,
|
||||
@ -1265,7 +1278,7 @@ async def link_capture_start(
|
||||
|
||||
@mcp.tool()
|
||||
async def link_capture_stop(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
link_id: Annotated[str | None, Field(description="Link UUID (single mode)")] = None,
|
||||
link_ids: Annotated[
|
||||
list[str] | None,
|
||||
@ -1283,7 +1296,7 @@ async def link_capture_stop(
|
||||
|
||||
@mcp.tool()
|
||||
async def link_capture_download(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
link_id: Annotated[str | None, Field(description="Link UUID (single mode)")] = None,
|
||||
link_ids: Annotated[
|
||||
list[str] | None, Field(description='Batch mode: ["uuid1","uuid2"] — get download URLs for multiple captures')
|
||||
@ -1314,7 +1327,7 @@ async def link_capture_download(
|
||||
|
||||
@mcp.tool()
|
||||
async def link_marker(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
link_id: Annotated[str, Field(description="UUID of the link")],
|
||||
action: Annotated[str, Field(description="Action: create, update, or delete")],
|
||||
bpf: Annotated[
|
||||
@ -1382,7 +1395,7 @@ async def link_marker(
|
||||
|
||||
@mcp.tool()
|
||||
async def marker_definition(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
action: Annotated[str, Field(description="Action: create, update, delete, or list")],
|
||||
bpf: Annotated[
|
||||
str | None, Field(description="BPF expression, e.g. 'arp', 'ospf', 'tcp port 22' (required for create)")
|
||||
@ -1433,7 +1446,7 @@ async def marker_definition(
|
||||
|
||||
@mcp.tool()
|
||||
async def snapshot_list(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""List all snapshots of a project."""
|
||||
return await asyncio.to_thread(
|
||||
@ -1447,7 +1460,7 @@ async def snapshot_list(
|
||||
|
||||
@mcp.tool()
|
||||
async def snapshot_create(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
name: Annotated[str, Field(description="Name for the new snapshot")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Create a new snapshot of a project.
|
||||
@ -1468,7 +1481,7 @@ async def snapshot_create(
|
||||
|
||||
@mcp.tool()
|
||||
async def snapshot_delete(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
snapshot_id: Annotated[str, Field(description="UUID of the snapshot to delete")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Delete a snapshot from a project. Cannot be undone."""
|
||||
@ -1484,7 +1497,7 @@ async def snapshot_delete(
|
||||
|
||||
@mcp.tool()
|
||||
async def snapshot_restore(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
snapshot_id: Annotated[str, Field(description="UUID of the snapshot to restore")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Restore a project to a previous snapshot state. The project may be closed and reopened."""
|
||||
@ -1503,7 +1516,7 @@ async def snapshot_restore(
|
||||
|
||||
@mcp.tool()
|
||||
async def drawing_list(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""List all drawings (labels, shapes, images) on a project canvas."""
|
||||
return await asyncio.to_thread(
|
||||
@ -1517,7 +1530,7 @@ async def drawing_list(
|
||||
|
||||
@mcp.tool()
|
||||
async def drawing_create(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
svg: Annotated[str, Field(description="SVG content for the drawing")],
|
||||
x: Annotated[int, Field(description="X coordinate (default: 0)")] = 0,
|
||||
y: Annotated[int, Field(description="Y coordinate (default: 0)")] = 0,
|
||||
@ -1557,7 +1570,7 @@ async def drawing_create(
|
||||
|
||||
@mcp.tool()
|
||||
async def drawing_get(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
drawing_id: Annotated[str, Field(description="UUID of the drawing")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Get detailed information about a specific drawing."""
|
||||
@ -1573,7 +1586,7 @@ async def drawing_get(
|
||||
|
||||
@mcp.tool()
|
||||
async def drawing_update(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
drawing_id: Annotated[str, Field(description="UUID of the drawing")],
|
||||
svg: Annotated[str | None, Field(description="New SVG content")] = None,
|
||||
locked: Annotated[bool | None, Field(description="Lock or unlock the drawing")] = None,
|
||||
@ -1593,7 +1606,7 @@ async def drawing_update(
|
||||
|
||||
@mcp.tool()
|
||||
async def drawing_delete(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
drawing_id: Annotated[str, Field(description="UUID of the drawing to delete")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Delete a drawing from a project canvas. Cannot be undone."""
|
||||
@ -1612,7 +1625,7 @@ async def drawing_delete(
|
||||
|
||||
@mcp.tool()
|
||||
async def project_lock(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Lock all drawings and nodes in a project to prevent accidental changes."""
|
||||
return await asyncio.to_thread(
|
||||
@ -1626,7 +1639,7 @@ async def project_lock(
|
||||
|
||||
@mcp.tool()
|
||||
async def project_unlock(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Unlock a project to allow editing of drawings and nodes."""
|
||||
return await asyncio.to_thread(
|
||||
@ -1640,7 +1653,7 @@ async def project_unlock(
|
||||
|
||||
@mcp.tool()
|
||||
async def project_locked(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Check whether a project is locked (preventing edits to drawings and nodes)."""
|
||||
return await asyncio.to_thread(
|
||||
@ -1868,7 +1881,7 @@ async def image_install() -> list[dict[str, Any]]:
|
||||
|
||||
@mcp.tool()
|
||||
async def device_config_send(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
device_configs: Annotated[
|
||||
list,
|
||||
Field(
|
||||
@ -1905,7 +1918,7 @@ async def device_config_send(
|
||||
|
||||
@mcp.tool()
|
||||
async def device_show_run(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
device_configs: Annotated[
|
||||
list,
|
||||
Field(
|
||||
@ -1947,7 +1960,7 @@ async def device_show_run(
|
||||
|
||||
@mcp.tool()
|
||||
async def vpcs_config_set(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
project_id: Annotated[str, Field(description="Project name or UUID")],
|
||||
device_configs: Annotated[
|
||||
list,
|
||||
Field(
|
||||
|
||||
223
gns3server/agent/mcp/locators.py
Normal file
223
gns3server/agent/mcp/locators.py
Normal file
@ -0,0 +1,223 @@
|
||||
#
|
||||
# Copyright (C) 2026 GNS3 Technologies Inc.
|
||||
# Author: Yue Guobin
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
"""
|
||||
Name locators for MCP tools.
|
||||
|
||||
LLM agents transcribe UUIDs unreliably (a 2-character slip invents an id
|
||||
that never existed and is invisible to the eye), so MCP tool parameters
|
||||
accept names wherever the server guarantees uniqueness: project names are
|
||||
unique among live projects (the controller rejects duplicates) and node
|
||||
names are unique within a project (auto-suffixed on collision). Links,
|
||||
drawings, appliances and snapshots have no names and stay UUID-only.
|
||||
|
||||
Resolution happens in the single dispatch choke point (_run_handler_sync):
|
||||
UUID-shaped values pass through untouched, names are resolved with one
|
||||
list call, and a name that matches nothing fails fast with the known
|
||||
names instead of a confusing downstream 404. As a backstop for residual
|
||||
UUID typos, error replies carrying "<UUID> doesn't exist" are matched
|
||||
against the ids visible in scope; a single candidate within edit
|
||||
distance 2 earns a "Did you mean ..." hint.
|
||||
"""
|
||||
|
||||
import re
|
||||
from typing import Any, Callable
|
||||
|
||||
from .projects import list_projects_handler
|
||||
from gns3server.agent.gns3_copilot.gns3_client.api_handlers import get_links_handler, get_nodes_handler
|
||||
|
||||
_UUID_RE = re.compile(r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$")
|
||||
_ID_IN_TEXT_RE = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}")
|
||||
|
||||
# how many names to list when a locator matches nothing
|
||||
_MAX_HINT_NAMES = 15
|
||||
|
||||
|
||||
def looks_like_uuid(value: Any) -> bool:
|
||||
"""
|
||||
Whether *value* is a UUID-shaped string (anything else is treated as a name).
|
||||
"""
|
||||
|
||||
return isinstance(value, str) and bool(_UUID_RE.match(value))
|
||||
|
||||
|
||||
def resolve_locators(
|
||||
params: dict[str, Any], run: Callable[[Callable, dict], Any]
|
||||
) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
|
||||
"""
|
||||
Rewrite name-shaped ``project_id``/``node_id`` values into UUIDs.
|
||||
|
||||
:param params: tool params (top-level keys only; nested structures are
|
||||
left alone — link_create's node tuples require ids)
|
||||
:param run: callback running (handler, params) through the same context
|
||||
:returns: (resolved params, None) on success, (None, error dict) on failure
|
||||
"""
|
||||
|
||||
params = dict(params)
|
||||
|
||||
project_ref = params.get("project_id")
|
||||
if project_ref is not None and not looks_like_uuid(project_ref):
|
||||
result = run(list_projects_handler, {})
|
||||
projects = _as_items(result, "projects")
|
||||
if projects is None:
|
||||
return None, {"error": f"Could not resolve project '{project_ref}': {_as_error(result)}"}
|
||||
match = _match_by_name(projects, project_ref)
|
||||
if match is None:
|
||||
return None, {"error": f"Project '{project_ref}' not found. Known projects: {_known_names(projects)}"}
|
||||
params["project_id"] = match["project_id"]
|
||||
|
||||
node_ref = params.get("node_id")
|
||||
if node_ref is not None and not looks_like_uuid(node_ref) and params.get("project_id"):
|
||||
result = run(get_nodes_handler, {"project_id": params["project_id"]})
|
||||
nodes = _as_items(result, "nodes")
|
||||
if nodes is None:
|
||||
return None, {"error": f"Could not resolve node '{node_ref}': {_as_error(result)}"}
|
||||
match = _match_by_name(nodes, node_ref)
|
||||
if match is None:
|
||||
return None, {"error": f"Node '{node_ref}' not found. Known nodes: {_known_names(nodes)}"}
|
||||
params["node_id"] = match["node_id"]
|
||||
|
||||
return params, None
|
||||
|
||||
|
||||
def add_did_you_mean(
|
||||
result: Any, params: dict[str, Any], run: Callable[[Callable, dict], Any]
|
||||
) -> Any:
|
||||
"""
|
||||
Append a "Did you mean ...?" hint to an error reply that names a
|
||||
non-existent UUID, when exactly one in-scope id is within edit
|
||||
distance 2 of it (the signature of a transcription slip).
|
||||
"""
|
||||
|
||||
if not isinstance(result, dict):
|
||||
return result
|
||||
error = result.get("error")
|
||||
if not isinstance(error, str) or ("doesn't exist" not in error and "not found" not in error.lower()):
|
||||
return result
|
||||
bad_ids = set(_ID_IN_TEXT_RE.findall(error))
|
||||
if not bad_ids:
|
||||
return result
|
||||
|
||||
candidates = _collect_candidates(params, run)
|
||||
if not candidates:
|
||||
return result
|
||||
|
||||
hints = []
|
||||
for bad in bad_ids:
|
||||
close = [(cid, label) for cid, label in candidates if bad != cid and _edit_distance_at_most(bad.lower(), cid.lower(), 2)]
|
||||
if len(close) == 1:
|
||||
cid, label = close[0]
|
||||
hints.append(f"Did you mean {label} ({cid})?")
|
||||
if not hints:
|
||||
return result
|
||||
result = dict(result)
|
||||
result["error"] = error.rstrip() + " " + " ".join(hints)
|
||||
return result
|
||||
|
||||
|
||||
def _as_items(result: Any, key: str) -> list | None:
|
||||
"""
|
||||
Normalize a handler reply into a list of dicts (None = failure).
|
||||
"""
|
||||
|
||||
if isinstance(result, list):
|
||||
return result
|
||||
if isinstance(result, dict):
|
||||
items = result.get(key)
|
||||
if isinstance(items, list):
|
||||
return items
|
||||
return None
|
||||
|
||||
|
||||
def _as_error(result: Any) -> Any:
|
||||
return result.get("error", result) if isinstance(result, dict) else result
|
||||
|
||||
|
||||
def _match_by_name(items: list, name: str) -> dict | None:
|
||||
"""
|
||||
The unique item whose name equals *name* (exact first, then a unique
|
||||
case-insensitive match); None when absent or ambiguous.
|
||||
"""
|
||||
|
||||
exact = [item for item in items if item.get("name") == name]
|
||||
if len(exact) == 1:
|
||||
return exact[0]
|
||||
if not exact:
|
||||
folded = [item for item in items if isinstance(item.get("name"), str) and item["name"].lower() == name.lower()]
|
||||
if len(folded) == 1:
|
||||
return folded[0]
|
||||
return None
|
||||
|
||||
|
||||
def _known_names(items: list) -> str:
|
||||
names = sorted(str(item.get("name", "?")) for item in items)
|
||||
if len(names) > _MAX_HINT_NAMES:
|
||||
names = [*names[:_MAX_HINT_NAMES], f"... ({len(names) - _MAX_HINT_NAMES} more)"]
|
||||
return ", ".join(names)
|
||||
|
||||
|
||||
def _collect_candidates(params: dict[str, Any], run: Callable[[Callable, dict], Any]) -> list[tuple[str, str]]:
|
||||
"""
|
||||
(id, label) pairs in scope for a hint: every project, plus the nodes
|
||||
and links of the project the call targeted. Best-effort — any failure
|
||||
just yields what was gathered so far.
|
||||
"""
|
||||
|
||||
candidates = []
|
||||
try:
|
||||
projects = _as_items(run(list_projects_handler, {}), "projects") or []
|
||||
candidates.extend((p["project_id"], f"project '{p.get('name', '?')}'") for p in projects if p.get("project_id"))
|
||||
except Exception: # best-effort hint enrichment
|
||||
pass
|
||||
project_id = params.get("project_id")
|
||||
if looks_like_uuid(project_id):
|
||||
try:
|
||||
nodes = _as_items(run(get_nodes_handler, {"project_id": project_id}), "nodes") or []
|
||||
candidates.extend((n["node_id"], f"node '{n.get('name', '?')}'") for n in nodes if n.get("node_id"))
|
||||
except Exception: # best-effort hint enrichment
|
||||
pass
|
||||
try:
|
||||
links = _as_items(run(get_links_handler, {"project_id": project_id}), "links") or []
|
||||
candidates.extend((link["link_id"], "link") for link in links if link.get("link_id"))
|
||||
except Exception: # best-effort hint enrichment
|
||||
pass
|
||||
return candidates
|
||||
|
||||
|
||||
def _edit_distance_at_most(a: str, b: str, cap: int) -> bool:
|
||||
"""
|
||||
Levenshtein distance with an early bail above *cap* (ids are short;
|
||||
the cap keeps the DP band trivial).
|
||||
"""
|
||||
|
||||
if abs(len(a) - len(b)) > cap:
|
||||
return False
|
||||
if a == b:
|
||||
return True
|
||||
previous = list(range(len(b) + 1))
|
||||
for i, char_a in enumerate(a, 1):
|
||||
current = [i]
|
||||
best = i
|
||||
for j, char_b in enumerate(b, 1):
|
||||
substitution = previous[j - 1] + (char_a != char_b)
|
||||
value = min(previous[j] + 1, current[j - 1] + 1, substitution)
|
||||
current.append(value)
|
||||
best = min(best, value)
|
||||
if best > cap:
|
||||
return False
|
||||
previous = current
|
||||
return previous[-1] <= cap
|
||||
170
tests/agent/mcp/test_locators.py
Normal file
170
tests/agent/mcp/test_locators.py
Normal file
@ -0,0 +1,170 @@
|
||||
#
|
||||
# Copyright (C) 2026 GNS3 Technologies Inc.
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
"""
|
||||
Name locators: name->UUID resolution and the UUID-typo "Did you mean" hint.
|
||||
"""
|
||||
|
||||
from gns3server.agent.mcp import locators
|
||||
from gns3server.agent.mcp.locators import add_did_you_mean, looks_like_uuid, resolve_locators
|
||||
|
||||
PROJECT_ID = "d1ab2349-2e53-4e41-a9ae-b5ffcd912408"
|
||||
NODE_ID = "b78a2630-2c9e-4b43-9397-c855c8053d1a"
|
||||
NODE_ID_TYPO = "b78a2630-3b9e-4b43-9397-c855c8053d1a" # 2c9e -> 3b9e (distance 2)
|
||||
|
||||
|
||||
def _run_factory(projects=None, nodes=None):
|
||||
"""A run() callback answering the two list handlers from canned data."""
|
||||
|
||||
calls = []
|
||||
|
||||
def run(handler, params):
|
||||
calls.append((handler, dict(params)))
|
||||
if handler is locators.list_projects_handler:
|
||||
if isinstance(projects, Exception):
|
||||
raise projects
|
||||
return {"projects": projects if projects is not None else []}
|
||||
if handler is locators.get_nodes_handler:
|
||||
if isinstance(nodes, Exception):
|
||||
raise nodes
|
||||
return nodes if nodes is not None else []
|
||||
raise AssertionError(f"unexpected handler {handler}")
|
||||
|
||||
return run, calls
|
||||
|
||||
|
||||
def test_looks_like_uuid():
|
||||
|
||||
assert looks_like_uuid(PROJECT_ID) is True
|
||||
assert looks_like_uuid(PROJECT_ID.upper()) is True
|
||||
assert looks_like_uuid("FRRDocker-4") is False
|
||||
assert looks_like_uuid("") is False
|
||||
assert looks_like_uuid(None) is False
|
||||
assert looks_like_uuid(1234) is False
|
||||
|
||||
|
||||
def test_uuid_params_pass_through_without_calls():
|
||||
"""UUID-shaped values never trigger resolution lookups."""
|
||||
|
||||
run, calls = _run_factory(projects=[{"project_id": PROJECT_ID, "name": "lab"}])
|
||||
params, error = resolve_locators({"project_id": PROJECT_ID, "node_id": NODE_ID}, run)
|
||||
assert error is None
|
||||
assert params == {"project_id": PROJECT_ID, "node_id": NODE_ID}
|
||||
assert calls == []
|
||||
|
||||
|
||||
def test_project_name_resolved():
|
||||
"""A name-shaped project_id becomes the project's UUID."""
|
||||
|
||||
run, calls = _run_factory(projects=[{"project_id": "11111111-1111-1111-1111-111111111111", "name": "other"},
|
||||
{"project_id": PROJECT_ID, "name": "my-lab"}])
|
||||
params, error = resolve_locators({"project_id": "my-lab"}, run)
|
||||
assert error is None
|
||||
assert params["project_id"] == PROJECT_ID
|
||||
assert len(calls) == 1 # only the project list
|
||||
|
||||
|
||||
def test_project_name_not_found_lists_known_names():
|
||||
|
||||
run, _calls = _run_factory(projects=[{"project_id": PROJECT_ID, "name": "my-lab"}])
|
||||
params, error = resolve_locators({"project_id": "no-such-lab"}, run)
|
||||
assert params is None
|
||||
assert "not found" in error["error"]
|
||||
assert "my-lab" in error["error"]
|
||||
|
||||
|
||||
def test_node_name_resolved_within_project():
|
||||
|
||||
run, calls = _run_factory(projects=[{"project_id": PROJECT_ID, "name": "my-lab"}],
|
||||
nodes=[{"node_id": NODE_ID, "name": "FRRDocker-4"}])
|
||||
params, error = resolve_locators({"project_id": "my-lab", "node_id": "FRRDocker-4"}, run)
|
||||
assert error is None
|
||||
assert params["project_id"] == PROJECT_ID
|
||||
assert params["node_id"] == NODE_ID
|
||||
# node lookup went to the already-resolved project
|
||||
assert calls[1] == (locators.get_nodes_handler, {"project_id": PROJECT_ID})
|
||||
|
||||
|
||||
def test_node_name_case_insensitive_unique_match():
|
||||
|
||||
run, _calls = _run_factory(nodes=[{"node_id": NODE_ID, "name": "FRRDocker-4"}])
|
||||
params, error = resolve_locators({"project_id": PROJECT_ID, "node_id": "frrdocker-4"}, run)
|
||||
assert error is None
|
||||
assert params["node_id"] == NODE_ID
|
||||
|
||||
|
||||
def test_node_name_not_found_lists_known_nodes():
|
||||
|
||||
run, _calls = _run_factory(nodes=[{"node_id": NODE_ID, "name": "FRRDocker-4"}])
|
||||
params, error = resolve_locators({"project_id": PROJECT_ID, "node_id": "vpcs-1"}, run)
|
||||
assert params is None
|
||||
assert "FRRDocker-4" in error["error"]
|
||||
|
||||
|
||||
def test_handler_failure_during_resolution_is_reported():
|
||||
|
||||
run, _calls = _run_factory(nodes={"error": "boom"})
|
||||
params, error = resolve_locators({"project_id": PROJECT_ID, "node_id": "FRRDocker-4"}, run)
|
||||
assert params is None
|
||||
assert "Could not resolve node" in error["error"]
|
||||
|
||||
|
||||
def test_did_you_mean_appended_for_transcription_slip():
|
||||
"""The 2c9e -> 3b9e class of typo earns exactly one hint."""
|
||||
|
||||
run, _calls = _run_factory(projects=[{"project_id": PROJECT_ID, "name": "my-lab"}],
|
||||
nodes=[{"node_id": NODE_ID, "name": "FRRDocker-4"}])
|
||||
result = add_did_you_mean(
|
||||
{"error": f"Node ID {NODE_ID_TYPO} doesn't exist"}, {"project_id": PROJECT_ID}, run
|
||||
)
|
||||
assert f"Did you mean node 'FRRDocker-4' ({NODE_ID})" in result["error"]
|
||||
# original message preserved
|
||||
assert NODE_ID_TYPO in result["error"]
|
||||
|
||||
|
||||
def test_no_hint_when_too_far_or_ambiguous():
|
||||
|
||||
far = "b78a2630-9999-4b43-9397-c855c8053d1a" # distance 3
|
||||
run, _calls = _run_factory(nodes=[{"node_id": NODE_ID, "name": "n1"}])
|
||||
result = add_did_you_mean({"error": f"Node ID {far} doesn't exist"}, {"project_id": PROJECT_ID}, run)
|
||||
assert "Did you mean" not in result["error"]
|
||||
|
||||
two = [NODE_ID, "b78a2630-2b9e-4b43-9397-c855c8053d1a"] # both within distance 2 of the typo
|
||||
run2, _calls2 = _run_factory(nodes=[{"node_id": i, "name": f"n{i}"} for i in two])
|
||||
result2 = add_did_you_mean(
|
||||
{"error": f"Node ID {NODE_ID_TYPO} doesn't exist"}, {"project_id": PROJECT_ID}, run2
|
||||
)
|
||||
assert "Did you mean" not in result2["error"]
|
||||
|
||||
|
||||
def test_non_error_results_untouched():
|
||||
|
||||
run, _calls = _run_factory(nodes=[{"node_id": NODE_ID, "name": "n1"}])
|
||||
payload = [{"node_id": NODE_ID}]
|
||||
assert add_did_you_mean(payload, {"project_id": PROJECT_ID}, run) is payload
|
||||
assert add_did_you_mean({"filters": {}}, {"project_id": PROJECT_ID}, run) == {"filters": {}}
|
||||
# errors without an id carry nothing to hint against
|
||||
result = add_did_you_mean({"error": "Node not found: no id here"}, {"project_id": PROJECT_ID}, run)
|
||||
assert result == {"error": "Node not found: no id here"}
|
||||
|
||||
|
||||
def test_edit_distance_at_most():
|
||||
|
||||
assert locators._edit_distance_at_most("abcd", "abcd", 2)
|
||||
assert locators._edit_distance_at_most("abcd", "abce", 2)
|
||||
assert locators._edit_distance_at_most("2c9e", "3b9e", 2)
|
||||
assert not locators._edit_distance_at_most("abcd", "dcba", 2)
|
||||
assert not locators._edit_distance_at_most("abc", "abcdef", 2)
|
||||
Loading…
x
Reference in New Issue
Block a user