diff --git a/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py b/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py index 221bf9bed..4eda2d3fd 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py +++ b/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py @@ -461,6 +461,25 @@ class VPCSCommands(BaseTool): port = device_ports[device_name]["port"] + node_type = device_ports[device_name].get("node_type") + if node_type != "vpcs": + # VPCS syntax typed into another node's CLI is silently + # discarded (e.g. IOS answers "% Invalid input"), so reject + # mismatched devices before a console session is opened + logger.error( + "Device '%s' is a %s node, not a VPCS node", + device_name, + node_type or "unknown-type", + ) + hosts_data[device_name] = { + "error": ( + f"Device '{device_name}' is a {node_type or 'unknown-type'} node, " + "not a VPCS node; use device_config_send / device_show_run " + "for network devices" + ) + } + continue + # VPCS devices use gns3_vpcs_telnet device type hosts_data[device_name] = { "port": port, diff --git a/gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py b/gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py index 3ef331157..b3127574e 100644 --- a/gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py +++ b/gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py @@ -59,6 +59,7 @@ def get_device_ports_from_topology( "device_name": { "port": console_port, "platform": "huawei", # Extracted from tags + "node_type": "vpcs", # GNS3 node type from the topology "groups": ["network_devices"], # For inheriting shared settings "connection_options": { "netmiko": { @@ -160,9 +161,14 @@ def get_device_ports_from_topology( # This is the Nornir best practice - each host has its own # connection configuration (device_type), while sharing common # settings (hostname, timeout) via group inheritance. + # node_type (the GNS3 node type, e.g. vpcs/iou/docker) lets callers + # reject mismatched devices before opening a console connection; + # DictInventory ignores keys it does not know, so carrying it here + # is safe for entries fed straight into Nornir. host_entry = { "port": node_info["console_port"], "platform": platform, + "node_type": node_info.get("type"), "groups": ["network_devices"], # For inheriting hostname, timeout, etc. "connection_options": { "netmiko": { diff --git a/gns3server/agent/mcp/__init__.py b/gns3server/agent/mcp/__init__.py index 2d9e39a89..780c05e62 100644 --- a/gns3server/agent/mcp/__init__.py +++ b/gns3server/agent/mcp/__init__.py @@ -1517,6 +1517,9 @@ async def vpcs_config_set( ) -> list[dict[str, Any]]: """Configure VPCS devices (set IP addresses, gateway, etc.). + Only VPCS nodes are accepted: any other node type in device_configs fails + with a per-device error instead of typing VPCS syntax into its CLI. + VPCS-specific configuration commands: - ip
/ Set IP and gateway - save Save config to startup.vpc diff --git a/tests/agent/mcp/test_device_tools.py b/tests/agent/mcp/test_device_tools.py new file mode 100644 index 000000000..fc996332e --- /dev/null +++ b/tests/agent/mcp/test_device_tools.py @@ -0,0 +1,65 @@ +""" +Device config tool tests with mocked topology and Nornir layers. + +Covers the VPCS node-type guard and the error contract shared by +device_config_send / device_show_run / vpcs_config_set. +""" +import json + +import pytest +from unittest.mock import MagicMock, patch + +VPCS_MOD = "gns3server.agent.gns3_copilot.tools_v2.vpcs_tools_netmiko" + + +def _topology_ports(node_type): + """Mocked get_device_ports_from_topology return value for one device.""" + return {"PC1": {"port": 5000, "node_type": node_type}} + + +class TestVPCSNodeTypeGuard: + + def test_non_vpcs_node_is_rejected(self): + from gns3server.agent.gns3_copilot.tools_v2.vpcs_tools_netmiko import VPCSCommands + + with patch(f"{VPCS_MOD}.get_device_ports_from_topology", + return_value=_topology_ports("iou")) as topo: + result = VPCSCommands()._run(json.dumps({ + "project_id": "0c0fde25-6ead-4413-a283-ea8fd2324291", + "device_configs": [{"device_name": "PC1", "commands": ["ip 10.0.0.1/24"]}], + })) + assert topo.called + assert len(result) == 1 + assert result[0]["device_name"] == "PC1" + assert result[0]["status"] == "failed" + assert "not a VPCS node" in result[0]["error"] + + def test_missing_node_type_is_rejected(self): + from gns3server.agent.gns3_copilot.tools_v2.vpcs_tools_netmiko import VPCSCommands + + with patch(f"{VPCS_MOD}.get_device_ports_from_topology", + return_value={"PC1": {"port": 5000}}): + result = VPCSCommands()._run(json.dumps({ + "project_id": "0c0fde25-6ead-4413-a283-ea8fd2324291", + "device_configs": [{"device_name": "PC1", "commands": ["ip 10.0.0.1/24"]}], + })) + assert result[0]["status"] == "failed" + assert "unknown-type" in result[0]["error"] + + def test_vpcs_node_passes_the_guard(self): + from gns3server.agent.gns3_copilot.tools_v2.vpcs_tools_netmiko import VPCSCommands + + tool = VPCSCommands() + nornir = MagicMock() + host_result = MagicMock(failed=False) + host_result.result = "OK" + nornir.run.return_value = {"PC1": host_result} + with patch(f"{VPCS_MOD}.get_device_ports_from_topology", + return_value=_topology_ports("vpcs")), \ + patch.object(VPCSCommands, "_initialize_nornir", return_value=nornir): + result = tool._run(json.dumps({ + "project_id": "0c0fde25-6ead-4413-a283-ea8fd2324291", + "device_configs": [{"device_name": "PC1", "commands": ["ip 10.0.0.1/24"]}], + })) + assert result[0]["status"] == "success" + assert result[0]["output"] == "OK"