From db164e9d419110bea797f1bf32761657643590ef Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Thu, 12 Mar 2026 16:43:03 +0800 Subject: [PATCH] feat(copilot): add multi-vendor device support with custom Huawei driver Add comprehensive multi-vendor support for GNS3 network automation, including a custom Netmiko driver for Huawei CloudEngine devices. Features: - Custom HuaweiTelnetCE driver for GNS3 emulation (no authentication) - Auto-commit before exit to prevent [Y/N/C] prompts - Dynamic device type detection from GNS3 node tags - Support for both Cisco IOS and Huawei devices - Proper VRP command handling (system-view, return confirmation) Implementation: - New package: utils/custom_netmiko/ - huawei_ce.py: Huawei CloudEngine driver - tests/test_huawei_ce.py: Unit tests (9/9 passing) - README.md: Driver development guide - Updated tools for multi-vendor support: - display_tools_nornir.py: Dynamic group generation - config_tools_nornir.py: Multi-vendor config commands - get_gns3_device_port.py: Device port extraction - Documentation: multi-vendor-device-support.md Limitations: - huawei_telnet_ce driver requires devices without authentication - For devices with username/password, use standard huawei_telnet driver Co-Authored-By: Yue Guobin " --- docs/gns3-copilot/README.md | 19 +- .../multi-vendor-device-support.md | 472 ++++++++++++++++++ .../tools_v2/config_tools_nornir.py | 46 +- .../tools_v2/display_tools_nornir.py | 46 +- .../utils/custom_netmiko/README.md | 183 +++++++ .../utils/custom_netmiko/__init__.py | 62 +++ .../utils/custom_netmiko/huawei_ce.py | 402 +++++++++++++++ .../utils/custom_netmiko/tests/__init__.py | 28 ++ .../custom_netmiko/tests/test_huawei_ce.py | 215 ++++++++ .../utils/get_gns3_device_port.py | 14 +- 10 files changed, 1462 insertions(+), 25 deletions(-) create mode 100644 docs/gns3-copilot/implemented/multi-vendor-device-support.md create mode 100644 gns3server/agent/gns3_copilot/utils/custom_netmiko/README.md create mode 100644 gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py create mode 100644 gns3server/agent/gns3_copilot/utils/custom_netmiko/huawei_ce.py create mode 100644 gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/__init__.py create mode 100644 gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_huawei_ce.py diff --git a/docs/gns3-copilot/README.md b/docs/gns3-copilot/README.md index 0ecf00f4c..eecf82159 100644 --- a/docs/gns3-copilot/README.md +++ b/docs/gns3-copilot/README.md @@ -12,7 +12,8 @@ docs/gns3-copilot/ ├── llm-model-configs.md # LLM model configuration system ├── command-security.md # Command security and filtering ├── context-window-management.md # Context window optimization - └── node-control-tools.md # Node start/stop/suspend tools for lab automation + ├── node-control-tools.md # Node start/stop/suspend tools for lab automation + └── multi-vendor-device-support.md # Multi-vendor device support (Cisco, Huawei) ``` ## Implemented Features @@ -75,6 +76,22 @@ Tools for controlling network device lifecycle in GNS3 projects. **Status:** ✅ Implemented +### Multi-Vendor Device Support (`implemented/multi-vendor-device-support.md`) +Multi-vendor network device support with custom Netmiko driver for Huawei devices. + +**Key Features:** +- Custom HuaweiTelnetCE driver for GNS3 emulation (no authentication) +- Cisco IOS Telnet support +- Dynamic device type detection from GNS3 tags +- Automatic Nornir group generation +- VRP-specific command handling (system-view, return confirmation) + +**Tested Vendors:** +- Cisco IOS (Telnet) +- Huawei CloudEngine (Telnet, custom driver) + +**Status:** ✅ Implemented + ## Future Enhancements The following features are currently under consideration or development: diff --git a/docs/gns3-copilot/implemented/multi-vendor-device-support.md b/docs/gns3-copilot/implemented/multi-vendor-device-support.md new file mode 100644 index 000000000..c862a148b --- /dev/null +++ b/docs/gns3-copilot/implemented/multi-vendor-device-support.md @@ -0,0 +1,472 @@ +# Multi-Vendor Network Device Support + +## Overview + +GNS3-Copilot supports network devices from multiple vendors through Netmiko and Nornir integration. The system includes a custom Netmiko driver for Huawei devices in GNS3 emulation environments and supports dynamic device type detection. + +## Supported Vendors + +| Vendor | Platform | Device Type | Protocol | Status | +|--------|----------|-------------|----------|--------| +| **Cisco** | `cisco_ios` | `cisco_ios_telnet` | Telnet | ✅ Tested | +| **Huawei** | `huawei` | `huawei_telnet_ce` | Telnet | ✅ Tested (Custom Driver) | + +## Custom Huawei Driver (`HuaweiTelnetCE`) + +### Problem Statement + +GNS3-emulated Huawei devices connect via console **without requiring authentication**. Standard Netmiko drivers attempt username/password authentication, causing connection timeouts. + +**Standard Driver Behavior:** +``` +Telnet Connection → Wait for username prompt → Send username → Wait for password → Send password → Access + ^ Times out after 20 seconds +``` + +**GNS3 Huawei Device:** +``` +Telnet Connection → Direct access to command line (no login prompts) + +``` + +### Solution: Custom Driver Architecture + +``` +BaseConnection (Netmiko base class) + ↓ +CiscoBaseConnection (Cisco-style base class) + ↓ +HuaweiBase (Huawei device base class) ← Inherits VRP support + ↓ +HuaweiTelnetCE (Custom GNS3 driver) ← Overrides telnet_login only +``` + +**Why Inherit from HuaweiBase?** +- ✅ Built-in VRP (Versatile Routing Platform) command handling +- ✅ Huawei-specific configuration mode (`system-view`) +- ✅ Huawei prompt patterns (`<...>`, `[...]`) +- ✅ Huawei paging disable (`screen-length 0 temporary`) +- ✅ Minimal code changes - only override authentication + +### HuaweiTelnetCE Implementation + +#### Location +``` +gns3server/agent/gns3_copilot/utils/custom_netmiko/huawei_ce.py +``` + +**Package Structure:** +``` +custom_netmiko/ +├── __init__.py # Package initialization, auto-registers all drivers +├── huawei_ce.py # Huawei CloudEngine custom driver +├── README.md # Driver development guide +└── tests/ # Unit tests + ├── __init__.py + └── test_huawei_ce.py # Huawei CE driver tests +``` + +#### Key Features + +1. **Skip Authentication** + - Directly detect Huawei prompt patterns + - No username/password prompts + - Connection ready in < 1 second + +2. **VRP Prompt Recognition** + ``` + User view: + System view: [HUAWEI] + Interface: [HUAWEI-GigabitEthernet0/0/1] + ``` + +3. **Automatic Confirmation Handling** + - Detects and responds to `[y/n]` prompts + - Example: `return` command asks "Return to user view? [y/n]:" + - Automatically sends `y` to confirm + +4. **Proper Output Collection** + - Uses Netmiko's `read_channel_timing()` for reliable output + - Waits for command completion (2s no new data = done) + - 30-second absolute timeout prevents hanging + +5. **Auto-Commit Before Exit** + - Automatically sends `commit` command before exiting config mode + - Prevents "Uncommitted configurations [Y/N/C]" prompt + - Ensures configuration changes are saved + +#### Limitations + +**Authentication Requirement:** +- The `huawei_telnet_ce` driver is designed for GNS3 devices **without authentication** +- If your Huawei device has been configured with a username/password: + - **Option 1**: Use the standard `huawei_telnet` driver (requires username/password) + - **Option 2**: Remove authentication from the device for GNS3 testing +- The driver does **not** currently auto-detect authentication requirements + +**When to Use Each Driver:** + +| Scenario | Use Driver | Requires Credentials? | +|----------|-----------|----------------------| +| GNS3 Huawei (fresh, no auth) | `huawei_telnet_ce` | ❌ No | +| GNS3 Huawei (configured with username/password) | `huawei_telnet` | ✅ Yes | +| Real Huawei hardware | `huawei_telnet` | ✅ Yes | + +#### Method Overrides + +**1. `telnet_login` - Skip Authentication** +```python +def telnet_login(self, pri_prompt_terminator=r"<\S+>|>\s*$", + alt_prompt_terminator=r"\[\S+\]", ...) -> str: + # Clear buffer + self.read_channel() + + # Send returns until prompt detected + for i in range(max_loops): + self.write_channel(self.RETURN) + output = self.read_channel() + + # Check for Huawei prompts + if re.search(pri_prompt_terminator, output): + return output # Success! + + return output # Best effort +``` + +**2. `send_config_set` - Configuration Commands** +```python +def send_config_set(self, config_commands, **kwargs) -> str: + # Enter config mode + output += self.config_mode(config_command="system-view") + + # Send all commands + for cmd in config_commands: + self.write_channel(f"{cmd}{self.RETURN}") + time.sleep(delay_factor * 0.05) + + # Collect output using Netmiko standard method + output += self.read_channel_timing(read_timeout=30, last_read=2.0) + + # Auto-commit before exit (prevents [Y/N/C] prompt) + self.write_channel(f"commit{self.RETURN}") + time.sleep(0.5 * self.global_delay_factor) + output += self.read_channel() + + # Exit config mode + output += self.exit_config_mode() + + return output +``` + +**3. `exit_config_mode` - Handle Confirmation** +```python +def exit_config_mode(self, exit_config="return", pattern=r"<\S+>|>\s*$") -> str: + self.write_channel(f"return{self.RETURN}") + + # Look for confirmation prompt + for _ in range(20): + new_output = self.read_channel() + + if re.search(r"\[y/n\]", new_output): + self.write_channel(f"y{self.RETURN}") # Auto-confirm + + if re.search(pattern, new_output): + return output # Back to user view + + return output +``` + +### Device Type Registration + +The custom driver must be registered with Netmiko's global mappings: + +```python +def register_custom_device_type() -> None: + import importlib + sd = importlib.import_module("netmiko.ssh_dispatcher") + + # Register in CLASS_MAPPER (for ConnectHandler) + sd.CLASS_MAPPER["huawei_telnet_ce"] = HuaweiTelnetCE + sd.CLASS_MAPPER["huawei_ce"] = HuaweiTelnetCE + + # Register in CLASS_MAPPER_BASE (for base class definitions) + sd.CLASS_MAPPER_BASE["huawei_telnet_ce"] = HuaweiTelnetCE + sd.CLASS_MAPPER_BASE["huawei_ce"] = HuaweiTelnetCE + + # CRITICAL: Rebuild static lists + sd.platforms = list(sd.CLASS_MAPPER.keys()) + sd.platforms.sort() + sd.telnet_platforms = [x for x in sd.platforms if "telnet" in x] +``` + +**Important: Static List Problem** +- `ssh_dispatcher.platforms` is computed at module import time +- Modifying `CLASS_MAPPER` doesn't automatically update `platforms` +- Must manually rebuild the list after registration + +**Auto-Registration** +```python +# Automatically runs on module import +try: + register_custom_device_type() +except Exception as e: + logger.warning(f"Failed to register custom device type: {e}") +``` + +## Dynamic Device Type Detection + +### GNS3 Node Tags + +Device type and platform are extracted from GNS3 node tags: + +``` +device_type:huawei_telnet_ce → Netmiko device type (precise) +platform:huawei → Nornir platform (high-level) +``` + +**Tag Examples:** + +| Vendor | Device Type Tag | Platform Tag | +|--------|----------------|--------------| +| Cisco IOS | `device_type:cisco_ios_telnet` | `platform:cisco_ios` | +| Huawei CE | `device_type:huawei_telnet_ce` | `platform:huawei` | + +### Automatic Group Generation + +Nornir groups are dynamically generated based on device type: + +```python +# From get_gns3_device_port.py +if device_type and "_telnet" in device_type: + group_name = f"{platform}_telnet" # e.g., "huawei_telnet" +else: + group_name = platform # e.g., "cisco_ios" + +hosts_data[device_name] = { + "port": console_port, + "groups": [group_name], + "device_type": device_type, + "platform": platform, +} +``` + +**Why Dynamic Groups?** +- Single Nornir instance can handle multiple vendors +- Each device gets correct connection parameters +- Vendor-specific commands work automatically + +## Usage Examples + +### Direct Netmiko Usage + +**Huawei Device (Custom Driver):** +```python +from netmiko import ConnectHandler +from gns3server.agent.gns3_copilot.utils import custom_netmiko + +# Custom driver auto-registers on import +device = { + "device_type": "huawei_telnet_ce", + "host": "127.0.0.1", + "port": 5000, + # No username/password needed! +} + +with ConnectHandler(**device) as conn: + # Execute display command + output = conn.send_command("display version") + + # Execute configuration commands + config = [ + "interface GE1/0/1", + "description Uplink-to-Core", + "undo shutdown" + ] + output = conn.send_config_set(config) +``` + +**Cisco IOS Device (Standard Driver):** +```python +from netmiko import ConnectHandler + +device = { + "device_type": "cisco_ios_telnet", + "host": "127.0.0.1", + "port": 5001, + "username": "cisco", + "password": "cisco", +} + +with ConnectHandler(**device) as conn: + output = conn.send_command("show version") + config = ["interface GigabitEthernet0/0", "description Test"] + output = conn.send_config_set(config) +``` + +### Nornir Multi-Vendor Automation + +```python +from nornir import InitNornir +from gns3server.agent.gns3_copilot.utils import custom_netmiko + +# Auto-register custom driver (happens automatically on import) +from gns3server.agent.gns3_copilot.utils.custom_netmiko import huawei_ce +huawei_ce.register_custom_device_type() + +# Initialize Nornir with mixed-vendor inventory +inventory = { + "options": { + "hosts": { + "huawei-sw1": { + "hostname": "127.0.0.1", + "port": 5001, + "platform": "huawei", + "device_type": "huawei_telnet_ce", + "groups": ["huawei_telnet"] + }, + "cisco-r1": { + "hostname": "127.0.0.1", + "port": 5002, + "platform": "cisco_ios", + "device_type": "cisco_ios_telnet", + "groups": ["cisco_ios_telnet"] + } + }, + "groups": { + "huawei_telnet": { + "platform": "huawei", + "device_type": "huawei_telnet_ce" + }, + "cisco_ios_telnet": { + "platform": "cisco_ios", + "device_type": "cisco_ios_telnet" + } + } + } +} + +nr = InitNornir(inventory=inventory) + +# Execute commands on all devices (multi-vendor) +result = nr.run(task=send_commands, commands=["display version"]) + +# Each device gets vendor-specific command handling +``` + +### GNS3 Copilot Tool Usage + +```python +from gns3server.agent.gns3_copilot.tools_v2 import DisplayToolNornir + +tool = DisplayToolNornir() +result = tool._run(json.dumps({ + "device_names": ["huawei-sw1", "cisco-r1"], + "commands": ["display version", "show version"], + "project_id": "project-uuid" +})) + +# Returns: +# { +# "huawei-sw1": { +# "display version": "", +# "status": "success" +# }, +# "cisco-r1": { +# "show version": "", +# "status": "success" +# } +# } +``` + +## Module Structure + +``` +gns3server/agent/gns3_copilot/ +├── utils/ +│ ├── custom_netmiko/ # Custom Netmiko drivers package +│ │ ├── __init__.py # Package initialization +│ │ ├── huawei_ce.py # Huawei CloudEngine driver +│ │ ├── README.md # Driver development guide +│ │ └── tests/ # Unit tests +│ │ ├── __init__.py +│ │ └── test_huawei_ce.py # Huawei CE driver tests +│ └── get_gns3_device_port.py # Device port extraction +├── tools_v2/ +│ ├── display_tools_nornir.py # Multi-vendor display commands +│ └── config_tools_nornir.py # Multi-vendor config commands +``` + +## Unit Testing + +### Test Coverage + +```python +# test_netmiko_custom.py + +class TestHuaweiTelnetCEDriver(unittest.TestCase): + def test_device_type_registered(self): + """Verify huawei_telnet_ce is in Netmiko CLASS_MAPPER""" + from netmiko.ssh_dispatcher import CLASS_MAPPER + self.assertIn("huawei_telnet_ce", CLASS_MAPPER) + + def test_inheritance_from_huawei_base(self): + """Verify inherits from HuaweiBase""" + from netmiko.huawei.huawei import HuaweiBase + self.assertTrue(issubclass(HuaweiTelnetCE, HuaweiBase)) + + def test_vrp_methods_available(self): + """Verify VRP-specific methods are available""" + methods = ["config_mode", "check_config_mode", "exit_config_mode"] + for method in methods: + self.assertTrue(hasattr(HuaweiTelnetCE, method)) +``` + +**Running Tests:** +```bash +source venv/bin/activate +python gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_huawei_ce.py +``` + +**Current Test Status:** ✅ All 9 tests passing + +## Platform vs Device Type + +### Key Concepts + +**Platform (Nornir):** +- High-level vendor identifier +- Used for inventory grouping +- Examples: `huawei`, `cisco_ios` + +**Device Type (Netmiko):** +- Precise driver type +- Includes protocol information +- Examples: `huawei_telnet_ce`, `cisco_ios_telnet` + +### Mapping + +| Platform | Device Type | Notes | +|----------|-------------|-------| +| `huawei` | `huawei_telnet_ce` | Custom driver for GNS3 | +| `cisco_ios` | `cisco_ios_telnet` | Standard Netmiko driver | + +## Related Documentation + +- [Custom Netmiko README](../../../../../gns3server/agent/gns3_copilot/utils/custom_netmiko/README.md) - Driver development guide +- [Node Control Tools](./node-control-tools.md) - Device lifecycle management +- [Command Security](./command-security.md) - Command filtering and validation +- [Chat API](./chat-api.md) - Session management and SSE + +## References + +- [Netmiko Documentation](https://ktbyers.github.io/netmiko/) +- [Netmiko PLATFORMS.md](https://github.com/ktbyers/netmiko/blob/master/PLATFORMS.md) +- [Nornir Documentation](https://nornir.readthedocs.io/) + +--- + +_Implementation Date: 2026-03-12_ + +_Status: ✅ Implemented - Custom Huawei driver for GNS3 emulation, multi-vendor support with Cisco IOS and Huawei tested_ + +_Unit Tests: ✅ 9/9 passing_ diff --git a/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py b/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py index cff664baf..a925bd76a 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py @@ -49,6 +49,22 @@ from gns3server.agent.gns3_copilot.utils.command_filter import ( filter_forbidden_commands, ) +# Import custom Netmiko device types for GNS3 emulation +# This registers huawei_telnet_ce and other custom device types +# NOTE: Must be imported BEFORE any Nornir operations to ensure device types are registered +from gns3server.agent.gns3_copilot.utils import custom_netmiko # noqa: F401 + +# Explicitly register custom device types to ensure they are available +# This is a safety measure in case the auto-registration on import doesn't work +try: + from gns3server.agent.gns3_copilot.utils.custom_netmiko import huawei_ce + + # Re-register to ensure device types are available + huawei_ce.register_custom_device_type() +except Exception: + # Fail silently - the import-time registration should have worked + pass + # config log logger = logging.getLogger(__name__) @@ -60,9 +76,9 @@ logging.getLogger("nornir.core").setLevel(logging.WARNING) logging.getLogger("nornir").setLevel(logging.WARNING) -# Local Nornir configuration functions for Cisco IOS Telnet devices +# Local Nornir configuration functions for network devices def _get_nornir_defaults() -> dict[str, Any]: - """Get Nornir default configuration for Cisco IOS.""" + """Get Nornir default configuration.""" return {"data": {"location": "gns3"}} @@ -70,11 +86,11 @@ def _get_nornir_groups_config( device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios" ) -> dict[str, Any]: """ - Get Nornir group configuration for Cisco IOS Telnet devices. + Get Nornir group configuration for network devices. Args: - device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet') - platform: Platform type for Nornir (e.g., 'cisco_ios') + device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet', 'huawei_telnet') + platform: Platform type for Nornir (e.g., 'cisco_ios', 'huawei') Returns: Dictionary containing Nornir group configuration @@ -92,7 +108,7 @@ def _get_nornir_groups_config( def _get_nornir_group( - group_name: str = "cisco_IOSv_telnet", + group_name: str = "network_devices_telnet", device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios", ) -> dict[str, Any]: @@ -101,8 +117,8 @@ def _get_nornir_group( Args: group_name: Name of the group - device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet') - platform: Platform type for Nornir (e.g., 'cisco_ios') + device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet', 'huawei_telnet') + platform: Platform type for Nornir (e.g., 'cisco_ios', 'huawei') Returns: Dictionary containing group configuration @@ -529,13 +545,23 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): ) defaults = _get_nornir_defaults() + # Dynamically generate group name based on platform and device type + # e.g., "huawei_telnet", "cisco_ios_telnet", "juniper_junos" + actual_platform = platform or "cisco_ios" + if device_type and "_telnet" in device_type: + group_name = f"{actual_platform}_telnet" + else: + group_name = actual_platform + # Log nornir account information gns3_host = get_gns3_server_host() logger.info( - "Initializing Nornir: host=%s, platform=%s, timeout=%d", + "Initializing Nornir: host=%s, platform=%s, device_type=%s, group=%s, timeout=%d", gns3_host, groups_data.get("platform"), + device_type, + group_name, groups_data.get("timeout"), ) @@ -544,7 +570,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): "plugin": "DictInventory", "options": { "hosts": hosts_data, - "groups": {"cisco_IOSv_telnet": groups_data}, + "groups": {group_name: groups_data}, "defaults": defaults, }, }, diff --git a/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py b/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py index 77f1369b3..53edb46c1 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py @@ -49,6 +49,22 @@ from gns3server.agent.gns3_copilot.utils.command_filter import ( filter_forbidden_commands, ) +# Import custom Netmiko device types for GNS3 emulation +# This registers huawei_telnet_ce and other custom device types +# NOTE: Must be imported BEFORE any Nornir operations to ensure device types are registered +from gns3server.agent.gns3_copilot.utils import custom_netmiko # noqa: F401 + +# Explicitly register custom device types to ensure they are available +# This is a safety measure in case the auto-registration on import doesn't work +try: + from gns3server.agent.gns3_copilot.utils.custom_netmiko import huawei_ce + + # Re-register to ensure device types are available + huawei_ce.register_custom_device_type() +except Exception: + # Fail silently - the import-time registration should have worked + pass + # config log logger = logging.getLogger(__name__) @@ -60,9 +76,9 @@ logging.getLogger("nornir.core").setLevel(logging.WARNING) logging.getLogger("nornir").setLevel(logging.WARNING) -# Local Nornir configuration functions for Cisco IOS Telnet devices +# Local Nornir configuration functions for network devices def _get_nornir_defaults() -> dict[str, Any]: - """Get Nornir default configuration for Cisco IOS.""" + """Get Nornir default configuration.""" return {"data": {"location": "gns3"}} @@ -70,11 +86,11 @@ def _get_nornir_groups_config( device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios" ) -> dict[str, Any]: """ - Get Nornir group configuration for Cisco IOS Telnet devices. + Get Nornir group configuration for network devices. Args: - device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet') - platform: Platform type for Nornir (e.g., 'cisco_ios') + device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet', 'huawei_telnet') + platform: Platform type for Nornir (e.g., 'cisco_ios', 'huawei') Returns: Dictionary containing Nornir group configuration @@ -92,7 +108,7 @@ def _get_nornir_groups_config( def _get_nornir_group( - group_name: str = "cisco_IOSv_telnet", + group_name: str = "network_devices_telnet", device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios", ) -> dict[str, Any]: @@ -101,8 +117,8 @@ def _get_nornir_group( Args: group_name: Name of the group - device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet') - platform: Platform type for Nornir (e.g., 'cisco_ios') + device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet', 'huawei_telnet') + platform: Platform type for Nornir (e.g., 'cisco_ios', 'huawei') Returns: Dictionary containing group configuration @@ -533,13 +549,23 @@ class ExecuteMultipleDeviceCommands(BaseTool): ) defaults = _get_nornir_defaults() + # Dynamically generate group name based on platform and device type + # e.g., "huawei_telnet", "cisco_ios_telnet", "juniper_junos" + actual_platform = platform or "cisco_ios" + if device_type and "_telnet" in device_type: + group_name = f"{actual_platform}_telnet" + else: + group_name = actual_platform + # Log nornir account information gns3_host = get_gns3_server_host() logger.info( - "Initializing Nornir: host=%s, platform=%s, timeout=%d", + "Initializing Nornir: host=%s, platform=%s, device_type=%s, group=%s, timeout=%d", gns3_host, groups_data.get("platform"), + device_type, + group_name, groups_data.get("timeout"), ) @@ -548,7 +574,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): "plugin": "DictInventory", "options": { "hosts": hosts_data, - "groups": {"cisco_IOSv_telnet": groups_data}, + "groups": {group_name: groups_data}, "defaults": defaults, }, }, diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/README.md b/gns3server/agent/gns3_copilot/utils/custom_netmiko/README.md new file mode 100644 index 000000000..3aade1359 --- /dev/null +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/README.md @@ -0,0 +1,183 @@ +# Custom Netmiko Drivers + +Custom Netmiko drivers for GNS3-emulated network devices. + +## Overview + +This package contains Netmiko drivers optimized for GNS3 emulation environments where devices may have non-standard authentication or behavior patterns. + +## Directory Structure + +``` +custom_netmiko/ +├── __init__.py # Package initialization, auto-registers all drivers +├── huawei_ce.py # Huawei CloudEngine custom driver +├── README.md # This file +└── tests/ # Unit tests + ├── __init__.py + └── test_huawei_ce.py # Huawei CE driver tests +``` + +## Supported Drivers + +### Huawei (`huawei_ce.py`) + +**Driver Name:** `HuaweiTelnetCE` + +**Device Types:** +- `huawei_telnet_ce` - Primary type +- `huawei_ce` - Alias + +**Features:** +- Skip authentication (for GNS3 devices without username/password) +- VRP prompt recognition (``, `[HUAWEI]`) +- Auto-commit before exit (prevents [Y/N/C] prompts) +- Proper output collection using `read_channel_timing()` + +**Limitations:** +- Does not support username/password authentication +- If your device requires authentication, use standard `huawei_telnet` driver + +## Usage + +### Direct Usage + +```python +from netmiko import ConnectHandler +from gns3server.agent.gns3_copilot.utils import custom_netmiko + +device = { + "device_type": "huawei_telnet_ce", + "host": "127.0.0.1", + "port": 5000, +} + +with ConnectHandler(**device) as conn: + output = conn.send_command("display version") + config = ["interface GE1/0/1", "description Test"] + output = conn.send_config_set(config) +``` + +### In GNS3 Copilot Tools + +The drivers are auto-registered when the tools are imported: + +```python +from gns3server.agent.gns3_copilot.tools_v2 import DisplayToolNornir + +tool = DisplayToolNornir() +result = tool._run(json.dumps({ + "device_names": ["huawei-sw1"], + "commands": ["display version"], + "project_id": "project-uuid" +})) +``` + +## Running Tests + +```bash +# Run all tests +source venv/bin/activate +python -m unittest discover -s gns3server/agent/gns3_copilot/utils/custom_netmiko/tests + +# Run Huawei CE tests only +python gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_huawei_ce.py +``` + +## Adding New Drivers + +### 1. Create Driver File + +Create a new file in `custom_netmiko/` (e.g., `cisco.py`): + +```python +# cisco.py +from netmiko.cisco.cisco_ios import CiscoIosTelnet + +class CustomCiscoDriver(CiscoIosTelnet): + """Custom Cisco driver for GNS3.""" + + def telnet_login(self, ...): + # Override login logic + pass + +def register_custom_device_type() -> None: + """Register this driver with Netmiko.""" + import importlib + sd = importlib.import_module("netmiko.ssh_dispatcher") + + sd.CLASS_MAPPER["cisco_custom"] = CustomCiscoDriver + sd.CLASS_MAPPER_BASE["cisco_custom"] = CustomCiscoDriver + + # Rebuild static lists + sd.platforms = list(sd.CLASS_MAPPER.keys()) + sd.platforms.sort() + +# Auto-register on import +try: + register_custom_device_type() +except Exception as e: + import logging + logging.getLogger(__name__).warning(f"Failed to register: {e}") +``` + +### 2. Update `__init__.py` + +Add import to `__init__.py`: + +```python +try: + from . import cisco # noqa: F401 +except Exception as e: + logger.warning(f"Failed to import Cisco driver: {e}", exc_info=True) + +__all__ = ["huawei_ce", "cisco"] +``` + +### 3. Create Tests + +Create `tests/test_cisco.py`: + +```python +import unittest +from gns3server.agent.gns3_copilot.utils.custom_netmiko import cisco + +class TestCustomCiscoDriver(unittest.TestCase): + def test_device_type_registered(self): + from netmiko.ssh_dispatcher import CLASS_MAPPER + self.assertIn("cisco_custom", CLASS_MAPPER) + +if __name__ == "__main__": + unittest.main() +``` + +## Troubleshooting + +### Driver Not Found Error + +``` +ValueError: Unsupported 'device_type' +``` + +**Solution:** Ensure the custom_netmiko package is imported before using the device: + +```python +from gns3server.agent.gns3_copilot.utils import custom_netmiko +# Now use the device +``` + +### Import Errors + +If you get import errors, check: +1. Virtual environment is activated +2. Python path includes project root +3. Dependencies are installed (`pip install netmiko nornir nornir-netmiko`) + +## Related Documentation + +- [Multi-Vendor Device Support](../../../docs/gns3-copilot/implemented/multi-vendor-device-support.md) +- [Netmiko Documentation](https://ktbyers.github.io/netmiko/) + +--- + +_Last updated: 2026-03-12_ diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py new file mode 100644 index 000000000..a81a3ec24 --- /dev/null +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py @@ -0,0 +1,62 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +# +# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3 +# +# This file is part of GNS3-Copilot project. +# +# GNS3-Copilot 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. +# +# GNS3-Copilot 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 GNS3-Copilot. If not, see . +# +# Copyright (C) 2025 Yue Guobin (岳国宾) +# Author: Yue Guobin (岳国宾) +# +# Project Home: https://github.com/yueguobin/gns3-copilot +# + +""" +Custom Netmiko drivers for GNS3-emulated network devices. + +This package contains custom Netmiko drivers optimized for GNS3 +emulation environments where devices may have non-standard +authentication or behavior patterns. + +Supported Drivers: +- huawei_ce: HuaweiTelnetCE for CloudEngine devices (no authentication) + +Usage: + from gns3server.agent.gns3_copilot.utils import custom_netmiko + + # Auto-registers all drivers on import + from netmiko import ConnectHandler + + device = { + "device_type": "huawei_telnet_ce", + "host": "127.0.0.1", + "port": 5000, + } + + with ConnectHandler(**device) as conn: + output = conn.send_command("display version") +""" + +import logging + +logger = logging.getLogger(__name__) + +# Import all custom drivers (auto-registers them with Netmiko) +try: + from . import huawei_ce # noqa: F401 +except Exception as e: + logger.warning(f"Failed to import Huawei CE driver: {e}", exc_info=True) + +__all__ = ["huawei_ce"] diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/huawei_ce.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/huawei_ce.py new file mode 100644 index 000000000..c8543a22b --- /dev/null +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/huawei_ce.py @@ -0,0 +1,402 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +# +# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3 +# +# This file is part of GNS3-Copilot project. +# +# GNS3-Copilot 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. +# +# GNS3-Copilot 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 GNS3-Copilot. If not, see . +# +# Copyright (C) 2025 Yue Guobin (岳国宾) +# Author: Yue Guobin (岳国宾) +# +# Project Home: https://github.com/yueguobin/gns3-copilot +# + +""" +Custom Netmiko device driver for Huawei devices in GNS3 emulation environment. + +This module provides a custom device type 'huawei_telnet_ce' for Huawei network +devices that connect via console without requiring authentication (username/password). + +This is specifically designed for GNS3 emulated Huawei devices (e.g., CloudEngine +series) where the console connection directly enters the system view without +login prompts. + +Key Features: +- Inherits from HuaweiBase for proper VRP command handling +- Skips authentication (no username/password required) +- Supports Huawei-specific config mode (system-view) +- Handles Huawei prompt patterns (<>, [], >) +""" + +import re +import time +from typing import Optional + +from netmiko.huawei.huawei import HuaweiBase + + +class HuaweiTelnetCE(HuaweiBase): + """ + Custom Huawei device driver for GNS3 emulation. + + Inherits from HuaweiBase to leverage existing VRP-specific functionality: + - system-view configuration mode + - Huawei prompt patterns + - Config mode detection and exit + + This driver overrides telnet_login to handle GNS3 devices that + don't require authentication. + """ + + def __init__( + self, + *args, + **kwargs, + ) -> None: + """Initialize HuaweiTelnetCE connection.""" + # Set default device type for proper initialization + # The '_telnet' suffix in device_type tells Netmiko to use Telnet protocol + kwargs.setdefault("device_type", "huawei_telnet") + + # Huawei prompt patterns (inherited from HuaweiBase) + # User view: + # System view: [HUAWEI] + # Interface view: [HUAWEI-GigabitEthernet0/0/1] + + super().__init__(*args, **kwargs) + + def telnet_login( + self, + pri_prompt_terminator: str = r"<\S+>|>\s*$", + alt_prompt_terminator: str = r"\[\S+\]", + username_pattern: str = r"(?:user:|username|login|user name)", + pwd_pattern: str = r"assword", + delay_factor: float = 1.0, + max_loops: int = 10, + ) -> str: + """ + Telnet login for GNS3 Huawei devices (no authentication). + + Simplified login logic for devices that connect directly to + command line without username/password prompts. + + Strategy: + 1. Clear any existing buffer data + 2. Send carriage returns to trigger prompt + 3. Wait for Huawei prompt pattern + 4. Return once prompt is detected + + Args: + pri_prompt_terminator: Primary prompt pattern (e.g., ) + alt_prompt_terminator: Alternate prompt pattern (e.g., [HUAWEI]) + username_pattern: Not used (kept for signature compatibility) + pwd_pattern: Not used (kept for signature compatibility) + delay_factor: Delay factor for timing + max_loops: Maximum wait loops + + Returns: + Output from the connection process + """ + delay_factor = self.select_delay_factor(delay_factor) + output = "" + return_msg = "" + + # Step 1: Clear buffer - read any existing data to avoid interference + try: + initial_data = self.read_channel() + if initial_data: + return_msg += initial_data + except Exception: + # Ignore errors during initial read + pass + + # Step 2: Send carriage returns and wait for prompt + for i in range(max_loops): + try: + # Send return to trigger prompt + self.write_channel(self.RETURN) + time.sleep(0.5 * delay_factor) + + # Read response + new_output = self.read_channel() + output = new_output + return_msg += new_output + + # Check for Huawei prompt patterns + # User view: , System view: [HUAWEI], or just > + if re.search(pri_prompt_terminator, output, flags=re.M): + return return_msg + if re.search(alt_prompt_terminator, output, flags=re.M): + return return_msg + + except EOFError: + self.remote_conn.close() + msg = f"Connection failed (EOF): {self.host}" + raise self.connection_error(msg) from None + except Exception: + # Continue trying on other exceptions + pass + + # Step 3: Final attempt - return what we have + # Even if we couldn't detect the prompt clearly, the connection + # might still be usable + return return_msg + + def session_preparation(self) -> None: + """ + Prepare the session after connection is established. + + Inherited from HuaweiBase, but ensures proper initialization + for GNS3 emulation environment. + """ + # Wait for prompt to stabilize + time.sleep(0.5 * self.global_delay_factor) + + # Disable paging using Huawei-specific command + try: + self.disable_paging(command="screen-length 0 temporary") + except Exception: + # If disable_paging fails, try the parent implementation + super().disable_paging() + + # Ensure we're in a clean state + try: + if hasattr(self, 'base_prompt') and self.base_prompt: + self._test_channel_read(pattern=self.base_prompt) + else: + # If base_prompt is not set yet, just read to clear buffer + self.read_channel() + except Exception: + pass + + def disable_paging( + self, + command: str = "screen-length 0 temporary", + **kwargs, + ) -> str: + """ + Disable paging for Huawei devices. + + Uses Huawei-specific command 'screen-length 0 temporary' + which disables paging for the current session only. + + Args: + command: Command to disable paging + + Returns: + Output from the disable paging command + """ + return super().disable_paging(command=command, **kwargs) + + def send_config_set( + self, + config_commands: str | list[str], + **kwargs, + ) -> str: + """ + Send configuration commands to Huawei device. + + Overrides the parent method to handle Huawei-specific behavior: + - Uses Huawei-specific prompts (<...> for user view, [...] for system view) + - Handles the 'return' confirmation prompt + - Uses read_channel_timing for proper output collection + + Args: + config_commands: Configuration commands to send + **kwargs: Additional arguments (exit_config_mode, read_timeout, etc.) + + Returns: + Output from configuration commands + """ + # Convert string to list + if isinstance(config_commands, str): + config_commands = [config_commands] + + # Get parameters with Huawei-specific defaults + exit_config_mode = kwargs.get("exit_config_mode", True) + read_timeout = kwargs.get("read_timeout", 30) # Longer timeout for GNS3 emulation + delay_factor = self.global_delay_factor + strip_prompt = kwargs.get("strip_prompt", False) + strip_command = kwargs.get("strip_command", False) + config_mode_command = kwargs.get("config_command", "system-view") + cmd_verify = kwargs.get("cmd_verify", False) # Disable cmd_verify for Huawei + + output = "" + + # Enter config mode if needed + if kwargs.get("enter_config_mode", True): + output += self.config_mode(config_command=config_mode_command) + + # Send all configuration commands + # Use cmd_verify=False mode: send all commands, then read all output at once + for cmd in config_commands: + self.write_channel(f"{cmd}{self.RETURN}") + # Small delay between commands + time.sleep(delay_factor * 0.05) + + # Use read_channel_timing to collect all output + # This method keeps reading until there is no new data for 'last_read' seconds + # This is the proper Netmiko way to handle command output + output += self.read_channel_timing(read_timeout=read_timeout, last_read=2.0) + + # Exit config mode if requested + if exit_config_mode: + # For Huawei devices, commit configuration before exiting + # This avoids the "Uncommitted configurations" [Y/N/C] prompt + try: + # Send commit command (in system view [HUAWEI]) + self.write_channel(f"commit{self.RETURN}") + time.sleep(0.5 * self.global_delay_factor) + commit_output = self.read_channel() + output += commit_output + except Exception: + # If commit fails, continue with exit (might not support commit) + pass + + # Now exit config mode + output += self.exit_config_mode() + + if strip_prompt: + output = self.strip_prompt(output) + + if strip_command: + output = self.strip_command(config_commands, output) + + return output + + def exit_config_mode( + self, exit_config: str = "return", pattern: str = r"<\S+>|>\s*$" + ) -> str: + r""" + Exit configuration mode for Huawei devices. + + Huawei devices display a confirmation prompt when using 'return': + Return to user view? [y/n]: + + This method automatically answers 'y' to the prompt. + + Args: + exit_config: Command to exit config mode (default: "return") + pattern: Pattern to detect user view prompt (default: r"<\S+>|>\s*$") + + Returns: + Output from exiting config mode + """ + # Check if we're currently in config mode + if not self.check_config_mode(): + return "" + + output = "" + # Send the exit command (write_channel returns None, don't concatenate) + self.write_channel(f"{exit_config}{self.RETURN}") + time.sleep(0.5 * self.global_delay_factor) + + # Look for the confirmation prompt + # Huawei prompt: "Return to user view? [y/n]:" + prompt_pattern = r"\[y/n\]" + max_loops = 20 # More loops for slower devices + + for _ in range(max_loops): + new_output = self.read_channel() + output += new_output + + # If we see the confirmation prompt, send 'y' + if re.search(prompt_pattern, new_output): + self.write_channel(f"y{self.RETURN}") # Returns None + time.sleep(0.5 * self.global_delay_factor) + # Clear the confirmation response + new_output = self.read_channel() + output += new_output + + # Check if we've exited to user view + if re.search(pattern, new_output): + return output + + # Final read to get remaining output + new_output = self.read_channel() + output += new_output + + return output + + +# Register the custom device type with Netmiko +def register_custom_device_type() -> None: + """ + Register the custom HuaweiTelnetCE device type with Netmiko. + + This function adds 'huawei_telnet_ce' to both Netmiko's CLASS_MAPPER + and CLASS_MAPPER_BASE so it can be used like any other built-in device type. + + Additionally, it updates the static 'platforms' and 'telnet_platforms' lists + which are used by ConnectHandler for device type validation. + + IMPORTANT: This function should be called BEFORE initializing Nornir or + running any Netmiko tasks. Call it explicitly at the appropriate time. + + Returns: + None + """ + # Use importlib to avoid namespace conflicts + import importlib + + # Import the module using importlib to ensure we get the module, not a function + sd = importlib.import_module("netmiko.ssh_dispatcher") + + # Add to both mappers + # CLASS_MAPPER_BASE is used for base class definitions + sd.CLASS_MAPPER_BASE["huawei_telnet_ce"] = HuaweiTelnetCE + sd.CLASS_MAPPER_BASE["huawei_ce"] = HuaweiTelnetCE + + # CLASS_MAPPER is used by ConnectHandler for device type validation + sd.CLASS_MAPPER["huawei_telnet_ce"] = HuaweiTelnetCE + sd.CLASS_MAPPER["huawei_ce"] = HuaweiTelnetCE + + # CRITICAL: Netmiko may also look for types with _telnet suffix + # Add this variant to ensure compatibility + sd.CLASS_MAPPER["huawei_telnet_ce_telnet"] = HuaweiTelnetCE + sd.CLASS_MAPPER_BASE["huawei_telnet_ce_telnet"] = HuaweiTelnetCE + + # CRITICAL: Update the static platforms lists + # These lists are computed at module import time and won't automatically update + # when CLASS_MAPPER is modified. We need to manually rebuild them. + + # Recalculate platforms list + sd.platforms = list(sd.CLASS_MAPPER.keys()) + sd.platforms.sort() + + # Recalculate platforms_base list + sd.platforms_base = list(sd.CLASS_MAPPER_BASE.keys()) + sd.platforms_base.sort() + + # Recalculate telnet_platforms list + sd.telnet_platforms = [x for x in sd.platforms if "telnet" in x] + + # Rebuild the platform strings used in error messages + sd.platforms_str = "\n" + "\n".join(sd.platforms_base) + sd.telnet_platforms_str = "\n" + "\n".join(sd.telnet_platforms) + + +# Auto-register on import +# This ensures the device type is available when the module is imported +# NOTE: For Nornir scenarios, you may need to call this explicitly +# before InitNornir to ensure proper timing +try: + register_custom_device_type() +except Exception as e: + # Log but don't fail on import + import logging + + logger = logging.getLogger(__name__) + logger.warning(f"Failed to register custom device type: {e}", exc_info=True) diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/__init__.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/__init__.py new file mode 100644 index 000000000..c7e50c78f --- /dev/null +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/__init__.py @@ -0,0 +1,28 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +# +# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3 +# +# This file is part of GNS3-Copilot project. +# +# GNS3-Copilot 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. +# +# GNS3-Copilot 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 GNS3-Copilot. If not, see . +# +# Copyright (C) 2025 Yue Guobin (岳国宾) +# Author: Yue Guobin (岳国宾) +# +# Project Home: https://github.com/yueguobin/gns3-copilot +# + +""" +Unit tests for custom Netmiko drivers. +""" diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_huawei_ce.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_huawei_ce.py new file mode 100644 index 000000000..74d272e89 --- /dev/null +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_huawei_ce.py @@ -0,0 +1,215 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: GPL-3.0-or-later +# +# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3 +# +# Unit test script for custom Netmiko HuaweiTelnetCE driver +# + +""" +Unit test script for HuaweiTelnetCE custom device driver. + +This script tests: +1. Device type registration +2. Inheritance from HuaweiBase +3. VRP-specific command handling +4. Telnet login logic (mocked) + +Run with: python test_huawei_ce.py +""" + +import sys +import unittest +from unittest.mock import Mock, patch, MagicMock + +# Add project root to path +sys.path.insert(0, "/home/yueguobin/myCode/GNS3/gns3-server") + + +class TestHuaweiTelnetCEDriver(unittest.TestCase): + """Test suite for HuaweiTelnetCE custom driver.""" + + @classmethod + def setUpClass(cls): + """Set up test fixtures - import and register custom driver.""" + # Import the custom driver module (this triggers registration) + from gns3server.agent.gns3_copilot.utils.custom_netmiko import huawei_ce + + cls.huawei_ce = huawei_ce + cls.HuaweiTelnetCE = huawei_ce.HuaweiTelnetCE + + def test_device_type_registered(self): + """Test that huawei_telnet_ce is registered in Netmiko.""" + from netmiko.ssh_dispatcher import CLASS_MAPPER, CLASS_MAPPER_BASE + + # Check CLASS_MAPPER + self.assertIn("huawei_telnet_ce", CLASS_MAPPER) + self.assertEqual(CLASS_MAPPER["huawei_telnet_ce"], self.HuaweiTelnetCE) + + # Check CLASS_MAPPER_BASE + self.assertIn("huawei_telnet_ce", CLASS_MAPPER_BASE) + self.assertEqual(CLASS_MAPPER_BASE["huawei_telnet_ce"], self.HuaweiTelnetCE) + + # Check alias + self.assertIn("huawei_ce", CLASS_MAPPER) + + def test_inheritance_from_huawei_base(self): + """Test that HuaweiTelnetCE inherits from HuaweiBase.""" + from netmiko.huawei.huawei import HuaweiBase + + # Verify inheritance + self.assertIsInstance(self.HuaweiTelnetCE, type) + # Check if HuaweiTelnetCE is a subclass of HuaweiBase + self.assertTrue(issubclass(self.HuaweiTelnetCE, HuaweiBase)) + + def test_huawei_base_methods_available(self): + """Test that VRP-specific methods are available.""" + # These methods should be inherited from HuaweiBase + vrp_methods = [ + "config_mode", + "check_config_mode", + "exit_config_mode", + "send_config_set", + "send_command", + "disable_paging", + ] + + for method_name in vrp_methods: + self.assertTrue( + hasattr(self.HuaweiTelnetCE, method_name), + f"Method {method_name} not found in HuaweiTelnetCE", + ) + + def test_telnet_login_method_exists(self): + """Test that telnet_login method is overridden.""" + # Should have overridden telnet_login + self.assertTrue(hasattr(self.HuaweiTelnetCE, "telnet_login")) + + # Get the method and check if it's defined in HuaweiTelnetCE + import inspect + + telnet_login_method = getattr(self.HuaweiTelnetCE, "telnet_login") + + # Check if method is in HuaweiTelnetCE's __dict__ (means it's defined there, not inherited) + self.assertIn( + "telnet_login", + self.HuaweiTelnetCE.__dict__, + "telnet_login should be defined in HuaweiTelnetCE", + ) + + def test_initialization_parameters(self): + """Test that initialization sets correct parameters.""" + # Create a mock instance (without actual connection) + with patch.object(self.HuaweiTelnetCE, "__init__", lambda self, *args, **kwargs: None): + instance = self.HuaweiTelnetCE.__new__(self.HuaweiTelnetCE) + + # Mock the necessary attributes + instance.protocol = "telnet" + instance.device_type = "huawei_telnet" + + # Verify protocol is set to telnet + self.assertEqual(instance.protocol, "telnet") + self.assertEqual(instance.device_type, "huawei_telnet") + + def test_connect_handler_accepts_device_type(self): + """Test that ConnectHandler accepts huawei_telnet_ce device type.""" + from netmiko.ssh_dispatcher import CLASS_MAPPER + + # Get platforms list + platforms = list(CLASS_MAPPER.keys()) + + # Verify huawei_telnet_ce is in platforms + self.assertIn("huawei_telnet_ce", platforms) + + # Verify it's in telnet platforms + telnet_platforms = [x for x in platforms if "telnet" in x] + self.assertIn("huawei_telnet_ce", telnet_platforms) + + def test_prompt_pattern_constants(self): + """Test that Huawei prompt patterns are correctly defined.""" + import inspect + + # Get the telnet_login method signature + sig = inspect.signature(self.HuaweiTelnetCE.telnet_login) + + # Check default prompt patterns + pri_prompt = sig.parameters["pri_prompt_terminator"].default + alt_prompt = sig.parameters["alt_prompt_terminator"].default + + # Should match Huawei prompt patterns + self.assertIn("<", pri_prompt) + self.assertIn("[", alt_prompt) + + def test_disable_paging_command(self): + """Test that disable_paging uses correct Huawei command.""" + import inspect + + sig = inspect.signature(self.HuaweiTelnetCE.disable_paging) + command_default = sig.parameters["command"].default + + # Should use Huawei-specific command + self.assertEqual(command_default, "screen-length 0 temporary") + + +class TestHuaweiTelnetCEIntegration(unittest.TestCase): + """Integration tests for HuaweiTelnetCE driver.""" + + def test_mock_telnet_connection(self): + """Test telnet_login logic with mocked connection.""" + from gns3server.agent.gns3_copilot.utils.custom_netmiko.huawei_ce import HuaweiTelnetCE + + # Create a mock instance + instance = HuaweiTelnetCE.__new__(HuaweiTelnetCE) + + # Mock the necessary attributes and methods + instance.host = "127.0.0.1" + instance.RETURN = "\r\n" + instance.global_delay_factor = 1.0 + instance.remote_conn = Mock() + + # Mock select_delay_factor + instance.select_delay_factor = Mock(return_value=1.0) + + # Mock read_channel and write_channel + test_outputs = ["", ""] + instance.read_channel = Mock(side_effect=test_outputs) + instance.write_channel = Mock() + + # Call telnet_login + result = instance.telnet_login() + + # Verify behavior + self.assertIn("", result) + instance.write_channel.assert_called() + self.assertEqual(instance.read_channel.call_count, 2) + + +def run_tests(): + """Run all tests and print results.""" + # Create test suite + loader = unittest.TestLoader() + suite = unittest.TestSuite() + + # Add test cases + suite.addTests(loader.loadTestsFromTestCase(TestHuaweiTelnetCEDriver)) + suite.addTests(loader.loadTestsFromTestCase(TestHuaweiTelnetCEIntegration)) + + # Run tests + runner = unittest.TextTestRunner(verbosity=2) + result = runner.run(suite) + + # Print summary + print("\n" + "=" * 100) + print("测试摘要:") + print(f" 运行: {result.testsRun}") + print(f" 成功: {result.testsRun - len(result.failures) - len(result.errors)}") + print(f" 失败: {len(result.failures)}") + print(f" 错误: {len(result.errors)}") + print("=" * 100) + + return result.wasSuccessful() + + +if __name__ == "__main__": + success = run_tests() + sys.exit(0 if success else 1) 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 7d9e8d49d..1f345489c 100644 --- a/gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py +++ b/gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py @@ -49,9 +49,9 @@ def get_device_ports_from_topology( { "device_name": { "port": console_port, - "groups": ["cisco_IOSv_telnet"], - "device_type": "cisco_ios_telnet", # Extracted from tags - "platform": "cisco_ios" # Extracted from tags + "groups": ["platform_telnet"], # Dynamically generated from platform/device_type + "device_type": "huawei_telnet", # Extracted from tags + "platform": "huawei" # Extracted from tags } } Devices that don't exist or missing console_port will not be included @@ -129,9 +129,15 @@ def get_device_ports_from_topology( ) # Add device to hosts_data + # Dynamically generate group name based on platform and device_type + if device_type and "_telnet" in device_type: + group_name = f"{platform}_telnet" + else: + group_name = platform + hosts_data[device_name] = { "port": node_info["console_port"], - "groups": ["cisco_IOSv_telnet"], + "groups": [group_name], "device_type": device_type, "platform": platform, }