fix: reject non-VPCS nodes in the VPCS config tool

VPCS syntax typed into another node's console is silently discarded
(IOS answers % Invalid input) while the tool still reports success.
get_device_ports_from_topology now carries the GNS3 node type through
to callers, and VPCSCommands fails device preparation with a per-device
error unless the node type is vpcs.
This commit is contained in:
YueGuobin 2026-08-26 00:36:47 +08:00
parent 8a8314ab29
commit 2951af6eab
No known key found for this signature in database
4 changed files with 93 additions and 0 deletions

View File

@ -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,

View File

@ -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": {

View File

@ -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 <address>/<mask> <gateway> Set IP and gateway
- save Save config to startup.vpc

View File

@ -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"