mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-08-27 12:30:13 +03:00
refactor(copilot): rename Huawei CE driver and add Ruijie tests with format fixes
- Rename huawei_telnet_ce to gns3_huawei_telnet_ce for consistent naming - Add comprehensive test suite for RuijieTelnetEnhanced driver (10 tests) - Update list_netmiko_telnet_devices.py to include gns3_ruijie_telnet - Fix all flake8 format issues (line length, unused imports, variables) - Update documentation to reflect device type rename
This commit is contained in:
YueGuobin
parent
6e88582ff1
commit
c76982b29e
@ -9,10 +9,10 @@ GNS3-Copilot supports network devices from multiple vendors through Netmiko and
|
||||
| Vendor | Platform | Device Type | Protocol | Status |
|
||||
|--------|----------|-------------|----------|--------|
|
||||
| **Cisco** | `cisco_ios` | `cisco_ios_telnet` | Telnet | ✅ Tested |
|
||||
| **Huawei** | `huawei` | `huawei_telnet_ce` | Telnet | ✅ Tested (Custom Driver) |
|
||||
| **Huawei** | `huawei` | `gns3_huawei_telnet_ce` | Telnet | ✅ Tested (Custom Driver) |
|
||||
| **Ruijie (锐捷)** | `ruijie_os` | `gns3_ruijie_telnet` | Telnet | ✅ Tested (Custom Driver) |
|
||||
|
||||
## Custom Huawei Driver (`HuaweiTelnetCE`)
|
||||
## Custom Huawei Driver (`GNS3HuaweiTelnetCE`)
|
||||
|
||||
### Problem Statement
|
||||
|
||||
@ -39,7 +39,7 @@ CiscoBaseConnection (Cisco-style base class)
|
||||
↓
|
||||
HuaweiBase (Huawei device base class) ← Inherits VRP support
|
||||
↓
|
||||
HuaweiTelnetCE (Custom GNS3 driver) ← Overrides telnet_login only
|
||||
GNS3HuaweiTelnetCE (Custom GNS3 driver) ← Overrides telnet_login only
|
||||
```
|
||||
|
||||
**Why Inherit from HuaweiBase?**
|
||||
@ -49,7 +49,7 @@ HuaweiTelnetCE (Custom GNS3 driver) ← Overrides telnet_login only
|
||||
- ✅ Huawei paging disable (`screen-length 0 temporary`)
|
||||
- ✅ Minimal code changes - only override authentication
|
||||
|
||||
### HuaweiTelnetCE Implementation
|
||||
### GNS3HuaweiTelnetCE Implementation
|
||||
|
||||
#### Location
|
||||
```
|
||||
@ -99,7 +99,7 @@ custom_netmiko/
|
||||
#### Limitations
|
||||
|
||||
**Authentication Requirement:**
|
||||
- The `huawei_telnet_ce` driver is designed for GNS3 devices **without authentication**
|
||||
- The `gns3_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
|
||||
@ -109,7 +109,7 @@ custom_netmiko/
|
||||
|
||||
| Scenario | Use Driver | Requires Credentials? |
|
||||
|----------|-----------|----------------------|
|
||||
| GNS3 Huawei (fresh, no auth) | `huawei_telnet_ce` | ❌ No |
|
||||
| GNS3 Huawei (fresh, no auth) | `gns3_huawei_telnet_ce` | ❌ No |
|
||||
| GNS3 Huawei (configured with username/password) | `huawei_telnet` | ✅ Yes |
|
||||
| Real Huawei hardware | `huawei_telnet` | ✅ Yes |
|
||||
|
||||
@ -187,12 +187,12 @@ def register_custom_device_type() -> None:
|
||||
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
|
||||
sd.CLASS_MAPPER["gns3_huawei_telnet_ce"] = GNS3HuaweiTelnetCE
|
||||
sd.CLASS_MAPPER["huawei_ce"] = GNS3HuaweiTelnetCE
|
||||
|
||||
# Register in CLASS_MAPPER_BASE (for base class definitions)
|
||||
sd.CLASS_MAPPER_BASE["huawei_telnet_ce"] = HuaweiTelnetCE
|
||||
sd.CLASS_MAPPER_BASE["huawei_ce"] = HuaweiTelnetCE
|
||||
sd.CLASS_MAPPER_BASE["gns3_huawei_telnet_ce"] = GNS3HuaweiTelnetCE
|
||||
sd.CLASS_MAPPER_BASE["huawei_ce"] = GNS3HuaweiTelnetCE
|
||||
|
||||
# CRITICAL: Rebuild static lists
|
||||
sd.platforms = list(sd.CLASS_MAPPER.keys())
|
||||
@ -369,7 +369,7 @@ with ConnectHandler(**device) as conn:
|
||||
Device type and platform are extracted from GNS3 node tags:
|
||||
|
||||
```
|
||||
device_type:huawei_telnet_ce → Netmiko device type (precise)
|
||||
device_type:gns3_huawei_telnet_ce → Netmiko device type (precise)
|
||||
platform:huawei → Nornir platform (high-level)
|
||||
```
|
||||
|
||||
@ -378,7 +378,7 @@ platform:huawei → Nornir platform (high-level)
|
||||
| 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` |
|
||||
| Huawei CE | `device_type:gns3_huawei_telnet_ce` | `platform:huawei` |
|
||||
| Ruijie | `device_type:gns3_ruijie_telnet` | `platform:ruijie_os` |
|
||||
|
||||
### Nornir Best Practice: Host-Level Connection Configuration
|
||||
@ -420,7 +420,7 @@ Defaults Level (data.location)
|
||||
# Had to create multiple groups dynamically
|
||||
groups = {
|
||||
"cisco_ios_telnet": {"device_type": "cisco_ios_telnet", ...},
|
||||
"huawei_telnet": {"device_type": "huawei_telnet_ce", ...},
|
||||
"huawei_telnet": {"device_type": "gns3_huawei_telnet_ce", ...},
|
||||
"juniper_junos": {"device_type": "juniper_junos_telnet", ...},
|
||||
}
|
||||
# Each host assigned to its vendor-specific group
|
||||
@ -548,7 +548,7 @@ groups = {
|
||||
|
||||
```python
|
||||
# Host level (highest priority)
|
||||
host["connection_options"]["netmiko"]["extras"]["device_type"] = "huawei_telnet_ce"
|
||||
host["connection_options"]["netmiko"]["extras"]["device_type"] = "gns3_huawei_telnet_ce"
|
||||
|
||||
↓ OVERRIDES
|
||||
|
||||
@ -578,7 +578,7 @@ from gns3server.agent.gns3_copilot.utils import custom_netmiko
|
||||
|
||||
# Custom driver auto-registers on import
|
||||
device = {
|
||||
"device_type": "huawei_telnet_ce",
|
||||
"device_type": "gns3_huawei_telnet_ce",
|
||||
"host": "127.0.0.1",
|
||||
"port": 5000,
|
||||
# No username/password needed!
|
||||
@ -638,7 +638,7 @@ inventory = {
|
||||
"groups": ["network_devices"],
|
||||
"connection_options": {
|
||||
"netmiko": {
|
||||
"extras": {"device_type": "huawei_telnet_ce"}
|
||||
"extras": {"device_type": "gns3_huawei_telnet_ce"}
|
||||
}
|
||||
}
|
||||
},
|
||||
@ -674,7 +674,7 @@ nr = InitNornir(inventory=inventory)
|
||||
result = nr.run(task=send_commands, commands=["display version"])
|
||||
|
||||
# Each device gets vendor-specific command handling
|
||||
# huawei-sw1 uses huawei_telnet_ce driver
|
||||
# huawei-sw1 uses gns3_huawei_telnet_ce driver
|
||||
# cisco-r1 uses cisco_ios_telnet driver
|
||||
```
|
||||
|
||||
@ -746,22 +746,22 @@ gns3server/agent/gns3_copilot/
|
||||
```python
|
||||
# test_netmiko_custom.py
|
||||
|
||||
class TestHuaweiTelnetCEDriver(unittest.TestCase):
|
||||
class TestGNS3HuaweiTelnetCEDriver(unittest.TestCase):
|
||||
def test_device_type_registered(self):
|
||||
"""Verify huawei_telnet_ce is in Netmiko CLASS_MAPPER"""
|
||||
"""Verify gns3_huawei_telnet_ce is in Netmiko CLASS_MAPPER"""
|
||||
from netmiko.ssh_dispatcher import CLASS_MAPPER
|
||||
self.assertIn("huawei_telnet_ce", CLASS_MAPPER)
|
||||
self.assertIn("gns3_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))
|
||||
self.assertTrue(issubclass(GNS3HuaweiTelnetCE, 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))
|
||||
self.assertTrue(hasattr(GNS3HuaweiTelnetCE, method))
|
||||
```
|
||||
|
||||
**Running Tests:**
|
||||
@ -787,7 +787,7 @@ python gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_huawei_ce.p
|
||||
- Precise driver type for Netmiko connection
|
||||
- Includes protocol information
|
||||
- **Actively used** to determine which Netmiko driver class to load
|
||||
- Examples: `huawei_telnet_ce`, `cisco_ios_telnet`
|
||||
- Examples: `gns3_huawei_telnet_ce`, `cisco_ios_telnet`
|
||||
|
||||
### Why Keep `platform` Field?
|
||||
|
||||
@ -807,7 +807,7 @@ python gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_huawei_ce.p
|
||||
|
||||
| Platform | Device Type | Netmiko Usage | Notes |
|
||||
|----------|-------------|---------------|-------|
|
||||
| `huawei` | `huawei_telnet_ce` | ✅ Active | Custom driver for GNS3 |
|
||||
| `huawei` | `gns3_huawei_telnet_ce` | ✅ Active | Custom driver for GNS3 |
|
||||
| `cisco_ios` | `cisco_ios_telnet` | ✅ Active | Standard Netmiko driver |
|
||||
|
||||
**Important:** For nornir_netmiko, only `device_type` in `connection_options` matters. The `platform` field is informational only.
|
||||
|
||||
@ -207,7 +207,7 @@
|
||||
| HP | `hp_procurve_telnet` | Netmiko |
|
||||
| Huawei | `huawei_olt_telnet` | Netmiko |
|
||||
| Huawei | `huawei_telnet` | Netmiko |
|
||||
| Huawei | `huawei_telnet_ce` | Custom ✨ (GNS3) |
|
||||
| Huawei | `gns3_huawei_telnet_ce` | Custom ✨ (GNS3) |
|
||||
| Infinera | `infinera_packet_telnet` | Netmiko |
|
||||
| Ipinfusion | `ipinfusion_ocnos_telnet` | Netmiko |
|
||||
| Juniper | `juniper_junos_telnet` | Netmiko |
|
||||
|
||||
@ -50,7 +50,7 @@ from gns3server.agent.gns3_copilot.utils.command_filter import (
|
||||
)
|
||||
|
||||
# Import custom Netmiko device types for GNS3 emulation
|
||||
# This registers huawei_telnet_ce and other custom device types
|
||||
# This registers gns3_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
|
||||
|
||||
|
||||
@ -50,7 +50,7 @@ from gns3server.agent.gns3_copilot.utils.command_filter import (
|
||||
)
|
||||
|
||||
# Import custom Netmiko device types for GNS3 emulation
|
||||
# This registers huawei_telnet_ce and other custom device types
|
||||
# This registers gns3_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
|
||||
|
||||
|
||||
@ -11,8 +11,11 @@ This package contains Netmiko drivers optimized for GNS3 emulation environments
|
||||
```
|
||||
custom_netmiko/
|
||||
├── __init__.py # Package initialization, auto-registers all drivers
|
||||
├── huawei_ce.py # Huawei CloudEngine custom driver
|
||||
├── huawei_ce.py # Huawei CloudEngine custom driver (GNS3HuaweiTelnetCE)
|
||||
├── ruijie_telnet.py # Ruijie OS custom driver (RuijieTelnetEnhanced)
|
||||
├── README.md # This file
|
||||
├── scripts/ # Utility scripts
|
||||
│ └── list_netmiko_telnet_devices.py
|
||||
└── tests/ # Unit tests
|
||||
├── __init__.py
|
||||
└── test_huawei_ce.py # Huawei CE driver tests
|
||||
@ -22,11 +25,11 @@ custom_netmiko/
|
||||
|
||||
### Huawei (`huawei_ce.py`)
|
||||
|
||||
**Driver Name:** `HuaweiTelnetCE`
|
||||
**Driver Name:** `GNS3HuaweiTelnetCE`
|
||||
|
||||
**Device Types:**
|
||||
- `huawei_telnet_ce` - Primary type
|
||||
- `huawei_ce` - Alias
|
||||
- `gns3_huawei_telnet_ce` - Primary type (GNS3 custom driver)
|
||||
- `huawei_telnet` - Standard Netmiko type (for devices with authentication)
|
||||
|
||||
**Features:**
|
||||
- Skip authentication (for GNS3 devices without username/password)
|
||||
@ -47,7 +50,7 @@ from netmiko import ConnectHandler
|
||||
from gns3server.agent.gns3_copilot.utils import custom_netmiko
|
||||
|
||||
device = {
|
||||
"device_type": "huawei_telnet_ce",
|
||||
"device_type": "gns3_huawei_telnet_ce",
|
||||
"host": "127.0.0.1",
|
||||
"port": 5000,
|
||||
}
|
||||
|
||||
@ -31,8 +31,11 @@ emulation environments where devices may have non-standard
|
||||
authentication or behavior patterns.
|
||||
|
||||
Supported Drivers:
|
||||
- huawei_ce: HuaweiTelnetCE for CloudEngine devices (no authentication)
|
||||
- ruijie_telnet: RuijieTelnet for Ruijie devices (interactive prompt handling)
|
||||
- huawei_ce: GNS3HuaweiTelnetCE for CloudEngine devices (no authentication)
|
||||
- ruijie_telnet: RuijieTelnetEnhanced for Ruijie devices (interactive prompt handling)
|
||||
|
||||
All drivers use 'gns3_' prefix to clearly distinguish them from Netmiko's
|
||||
built-in drivers.
|
||||
|
||||
Usage:
|
||||
from gns3server.agent.gns3_copilot.utils import custom_netmiko
|
||||
@ -40,8 +43,9 @@ Usage:
|
||||
# Auto-registers all drivers on import
|
||||
from netmiko import ConnectHandler
|
||||
|
||||
# Huawei device
|
||||
device = {
|
||||
"device_type": "huawei_telnet_ce",
|
||||
"device_type": "gns3_huawei_telnet_ce",
|
||||
"host": "127.0.0.1",
|
||||
"port": 5000,
|
||||
}
|
||||
|
||||
@ -57,7 +57,7 @@
|
||||
"""
|
||||
Custom Netmiko device driver for Huawei devices in GNS3 emulation environment.
|
||||
|
||||
This module provides a custom device type 'huawei_telnet_ce' for Huawei
|
||||
This module provides a custom device type 'gns3_huawei_telnet_ce' for Huawei
|
||||
network devices that connect via console without requiring authentication
|
||||
(username/password).
|
||||
|
||||
@ -80,7 +80,7 @@ import time
|
||||
from netmiko.huawei.huawei import HuaweiBase
|
||||
|
||||
|
||||
class HuaweiTelnetCE(HuaweiBase):
|
||||
class GNS3HuaweiTelnetCE(HuaweiBase):
|
||||
"""
|
||||
Custom Huawei device driver for GNS3 emulation.
|
||||
|
||||
@ -98,7 +98,7 @@ class HuaweiTelnetCE(HuaweiBase):
|
||||
*args,
|
||||
**kwargs,
|
||||
) -> None:
|
||||
"""Initialize HuaweiTelnetCE connection."""
|
||||
"""Initialize GNS3HuaweiTelnetCE connection."""
|
||||
# Set default device type for proper initialization
|
||||
# The '_telnet' suffix in device_type tells Netmiko to use
|
||||
# Telnet protocol
|
||||
@ -375,9 +375,9 @@ class HuaweiTelnetCE(HuaweiBase):
|
||||
# Register the custom device type with Netmiko
|
||||
def register_custom_device_type() -> None:
|
||||
"""
|
||||
Register the custom HuaweiTelnetCE device type with Netmiko.
|
||||
Register the custom GNS3HuaweiTelnetCE device type with Netmiko.
|
||||
|
||||
This function adds 'huawei_telnet_ce' to both Netmiko's CLASS_MAPPER
|
||||
This function adds 'gns3_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.
|
||||
|
||||
@ -398,10 +398,10 @@ def register_custom_device_type() -> None:
|
||||
|
||||
# Register the device type in both mappers
|
||||
# CLASS_MAPPER_BASE is used for base class definitions
|
||||
sd.CLASS_MAPPER_BASE["huawei_telnet_ce"] = HuaweiTelnetCE
|
||||
sd.CLASS_MAPPER_BASE["gns3_huawei_telnet_ce"] = GNS3HuaweiTelnetCE
|
||||
|
||||
# CLASS_MAPPER is used by ConnectHandler for device type validation
|
||||
sd.CLASS_MAPPER["huawei_telnet_ce"] = HuaweiTelnetCE
|
||||
sd.CLASS_MAPPER["gns3_huawei_telnet_ce"] = GNS3HuaweiTelnetCE
|
||||
|
||||
# CRITICAL: Update the static platforms lists
|
||||
# These lists are computed at module import time and won't
|
||||
|
||||
@ -58,12 +58,12 @@ class RuijieTelnetEnhanced(RuijieOSBase):
|
||||
# Interactive command patterns that trigger [yes/no] prompts
|
||||
# These are commands that commonly require confirmation
|
||||
INTERACTIVE_PATTERNS = [
|
||||
re.compile(r'^router-id\s+', re.IGNORECASE), # OSPF/EIGRP/BGP router-id
|
||||
re.compile(r'^erase\s+', re.IGNORECASE), # erase startup-config
|
||||
re.compile(r'^delete\s+', re.IGNORECASE), # delete files
|
||||
re.compile(r'^format\s+', re.IGNORECASE), # format filesystem
|
||||
re.compile(r'^reload\b', re.IGNORECASE), # reload/reboot
|
||||
re.compile(r'^boot\s+system\s+', re.IGNORECASE), # change boot image
|
||||
re.compile(r'^router-id\s+', re.IGNORECASE), # OSPF router-id
|
||||
re.compile(r'^erase\s+', re.IGNORECASE), # erase startup-config
|
||||
re.compile(r'^delete\s+', re.IGNORECASE), # delete files
|
||||
re.compile(r'^format\s+', re.IGNORECASE), # format filesystem
|
||||
re.compile(r'^reload\b', re.IGNORECASE), # reload/reboot
|
||||
re.compile(r'^boot\s+system\s+', re.IGNORECASE), # change boot image
|
||||
]
|
||||
|
||||
def __init__(
|
||||
@ -74,7 +74,8 @@ class RuijieTelnetEnhanced(RuijieOSBase):
|
||||
"""Initialize RuijieTelnetEnhanced connection."""
|
||||
# Set default_enter for telnet (like Netmiko's RuijieOSTelnet)
|
||||
default_enter = kwargs.get("default_enter")
|
||||
kwargs["default_enter"] = "\r\n" if default_enter is None else default_enter
|
||||
if default_enter is None:
|
||||
kwargs["default_enter"] = "\r\n"
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
def _preprocess_interactive_commands(
|
||||
@ -127,7 +128,8 @@ class RuijieTelnetEnhanced(RuijieOSBase):
|
||||
|
||||
Args:
|
||||
config_commands: Configuration commands to send
|
||||
**kwargs: Additional arguments (exit_config_mode, read_timeout, etc.)
|
||||
**kwargs: Additional arguments (exit_config_mode, read_timeout,
|
||||
etc.)
|
||||
|
||||
Returns:
|
||||
Output from configuration commands
|
||||
@ -160,7 +162,8 @@ class RuijieTelnetEnhanced(RuijieOSBase):
|
||||
)
|
||||
except Exception as batch_error:
|
||||
logging.warning(
|
||||
"Ruijie device: Batch send failed, falling back to one-by-one: %s",
|
||||
"Ruijie device: Batch send failed, "
|
||||
"falling back to one-by-one: %s",
|
||||
batch_error,
|
||||
)
|
||||
# Fallback to one-by-one send with real-time detection
|
||||
@ -229,8 +232,8 @@ class RuijieTelnetEnhanced(RuijieOSBase):
|
||||
for pattern in interactive_patterns:
|
||||
if re.search(pattern, new_output, re.IGNORECASE):
|
||||
logging.info(
|
||||
"Ruijie device: Detected interactive prompt after '%s', "
|
||||
"sending 'yes'",
|
||||
"Ruijie device: Detected interactive prompt "
|
||||
"after '%s', sending 'yes'",
|
||||
cmd,
|
||||
)
|
||||
# Send 'yes' to confirm
|
||||
|
||||
@ -72,7 +72,7 @@ def register_custom_drivers() -> set:
|
||||
import huawei_ce
|
||||
|
||||
# Store custom device types before registration
|
||||
custom_devices.add("huawei_telnet_ce")
|
||||
custom_devices.add("gns3_huawei_telnet_ce")
|
||||
|
||||
# The driver auto-registers on import, but we can call it
|
||||
# explicitly
|
||||
@ -84,7 +84,29 @@ def register_custom_drivers() -> set:
|
||||
except Exception as e:
|
||||
# Log but don't fail
|
||||
print(
|
||||
f"Warning: Failed to register custom drivers: {e}",
|
||||
f"Warning: Failed to register Huawei CE driver: {e}",
|
||||
file=sys.stderr
|
||||
)
|
||||
|
||||
try:
|
||||
# Import and register Ruijie Telnet driver
|
||||
from gns3server.agent.gns3_copilot.utils.custom_netmiko \
|
||||
import ruijie_telnet
|
||||
|
||||
# Store custom device types before registration
|
||||
custom_devices.add("gns3_ruijie_telnet")
|
||||
|
||||
# The driver auto-registers on import, but we can call it
|
||||
# explicitly
|
||||
if hasattr(ruijie_telnet, 'register_custom_device_type'):
|
||||
ruijie_telnet.register_custom_device_type()
|
||||
except ImportError:
|
||||
# Silently skip if custom drivers are not available
|
||||
pass
|
||||
except Exception as e:
|
||||
# Log but don't fail
|
||||
print(
|
||||
f"Warning: Failed to register Ruijie driver: {e}",
|
||||
file=sys.stderr
|
||||
)
|
||||
|
||||
|
||||
@ -3,11 +3,11 @@
|
||||
#
|
||||
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
|
||||
#
|
||||
# Unit test script for custom Netmiko HuaweiTelnetCE driver
|
||||
# Unit test script for custom Netmiko GNS3HuaweiTelnetCE driver
|
||||
#
|
||||
|
||||
"""
|
||||
Unit test script for HuaweiTelnetCE custom device driver.
|
||||
Unit test script for GNS3HuaweiTelnetCE custom device driver.
|
||||
|
||||
This script tests:
|
||||
1. Device type registration
|
||||
@ -20,47 +20,56 @@ Run with: python test_huawei_ce.py
|
||||
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import Mock, patch, MagicMock
|
||||
from unittest.mock import Mock, patch
|
||||
import os
|
||||
|
||||
# Add project root to path using relative path
|
||||
test_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(test_dir)))))
|
||||
project_root = os.path.dirname(
|
||||
os.path.dirname(os.path.dirname(
|
||||
os.path.dirname(os.path.dirname(test_dir)))))
|
||||
sys.path.insert(0, project_root)
|
||||
|
||||
|
||||
class TestHuaweiTelnetCEDriver(unittest.TestCase):
|
||||
"""Test suite for HuaweiTelnetCE custom driver."""
|
||||
"""Test suite for GNS3HuaweiTelnetCE 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
|
||||
# Import the custom driver module (triggers registration)
|
||||
from gns3server.agent.gns3_copilot.utils.custom_netmiko import (
|
||||
huawei_ce,
|
||||
)
|
||||
|
||||
cls.huawei_ce = huawei_ce
|
||||
cls.HuaweiTelnetCE = huawei_ce.HuaweiTelnetCE
|
||||
cls.HuaweiTelnetCE = huawei_ce.GNS3HuaweiTelnetCE
|
||||
|
||||
def test_device_type_registered(self):
|
||||
"""Test that huawei_telnet_ce is registered in Netmiko."""
|
||||
"""Test that gns3_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)
|
||||
self.assertIn("gns3_huawei_telnet_ce", CLASS_MAPPER)
|
||||
self.assertEqual(
|
||||
CLASS_MAPPER["gns3_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)
|
||||
self.assertIn("gns3_huawei_telnet_ce", CLASS_MAPPER_BASE)
|
||||
self.assertEqual(
|
||||
CLASS_MAPPER_BASE["gns3_huawei_telnet_ce"],
|
||||
self.HuaweiTelnetCE
|
||||
)
|
||||
|
||||
def test_inheritance_from_huawei_base(self):
|
||||
"""Test that HuaweiTelnetCE inherits from HuaweiBase."""
|
||||
"""Test that GNS3HuaweiTelnetCE inherits from HuaweiBase."""
|
||||
from netmiko.huawei.huawei import HuaweiBase
|
||||
|
||||
# Verify inheritance
|
||||
self.assertIsInstance(self.HuaweiTelnetCE, type)
|
||||
# Check if HuaweiTelnetCE is a subclass of HuaweiBase
|
||||
# Check if GNS3HuaweiTelnetCE is a subclass of HuaweiBase
|
||||
self.assertTrue(issubclass(self.HuaweiTelnetCE, HuaweiBase))
|
||||
|
||||
def test_huawei_base_methods_available(self):
|
||||
@ -78,7 +87,7 @@ class TestHuaweiTelnetCEDriver(unittest.TestCase):
|
||||
for method_name in vrp_methods:
|
||||
self.assertTrue(
|
||||
hasattr(self.HuaweiTelnetCE, method_name),
|
||||
f"Method {method_name} not found in HuaweiTelnetCE",
|
||||
f"Method {method_name} not found in GNS3HuaweiTelnetCE",
|
||||
)
|
||||
|
||||
def test_telnet_login_method_exists(self):
|
||||
@ -86,23 +95,24 @@ class TestHuaweiTelnetCEDriver(unittest.TestCase):
|
||||
# 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)
|
||||
# Check if method is in GNS3HuaweiTelnetCE's __dict__
|
||||
# (means it's defined there, not inherited)
|
||||
self.assertIn(
|
||||
"telnet_login",
|
||||
self.HuaweiTelnetCE.__dict__,
|
||||
"telnet_login should be defined in HuaweiTelnetCE",
|
||||
"telnet_login should be defined in GNS3HuaweiTelnetCE",
|
||||
)
|
||||
|
||||
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)
|
||||
with patch.object(
|
||||
self.HuaweiTelnetCE, "__init__",
|
||||
lambda self, *args, **kwargs: None
|
||||
):
|
||||
instance = self.HuaweiTelnetCE.__new__(
|
||||
self.HuaweiTelnetCE
|
||||
)
|
||||
|
||||
# Mock the necessary attributes
|
||||
instance.protocol = "telnet"
|
||||
@ -113,18 +123,18 @@ class TestHuaweiTelnetCEDriver(unittest.TestCase):
|
||||
self.assertEqual(instance.device_type, "huawei_telnet")
|
||||
|
||||
def test_connect_handler_accepts_device_type(self):
|
||||
"""Test that ConnectHandler accepts huawei_telnet_ce device type."""
|
||||
"""Test that ConnectHandler accepts gns3_huawei_telnet_ce."""
|
||||
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 gns3_huawei_telnet_ce is in platforms
|
||||
self.assertIn("gns3_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)
|
||||
self.assertIn("gns3_huawei_telnet_ce", telnet_platforms)
|
||||
|
||||
def test_prompt_pattern_constants(self):
|
||||
"""Test that Huawei prompt patterns are correctly defined."""
|
||||
@ -153,14 +163,18 @@ class TestHuaweiTelnetCEDriver(unittest.TestCase):
|
||||
|
||||
|
||||
class TestHuaweiTelnetCEIntegration(unittest.TestCase):
|
||||
"""Integration tests for HuaweiTelnetCE driver."""
|
||||
"""Integration tests for GNS3HuaweiTelnetCE 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
|
||||
from gns3server.agent.gns3_copilot.utils.custom_netmiko.huawei_ce import ( # noqa: E501
|
||||
GNS3HuaweiTelnetCE,
|
||||
)
|
||||
|
||||
# Create a mock instance
|
||||
instance = HuaweiTelnetCE.__new__(HuaweiTelnetCE)
|
||||
instance = GNS3HuaweiTelnetCE.__new__(
|
||||
GNS3HuaweiTelnetCE
|
||||
)
|
||||
|
||||
# Mock the necessary attributes and methods
|
||||
instance.host = "127.0.0.1"
|
||||
@ -192,8 +206,12 @@ def run_tests():
|
||||
suite = unittest.TestSuite()
|
||||
|
||||
# Add test cases
|
||||
suite.addTests(loader.loadTestsFromTestCase(TestHuaweiTelnetCEDriver))
|
||||
suite.addTests(loader.loadTestsFromTestCase(TestHuaweiTelnetCEIntegration))
|
||||
suite.addTests(
|
||||
loader.loadTestsFromTestCase(TestHuaweiTelnetCEDriver)
|
||||
)
|
||||
suite.addTests(
|
||||
loader.loadTestsFromTestCase(TestHuaweiTelnetCEIntegration)
|
||||
)
|
||||
|
||||
# Run tests
|
||||
runner = unittest.TextTestRunner(verbosity=2)
|
||||
@ -203,7 +221,10 @@ def run_tests():
|
||||
print("\n" + "=" * 100)
|
||||
print("Test Summary:")
|
||||
print(f" Run: {result.testsRun}")
|
||||
print(f" Success: {result.testsRun - len(result.failures) - len(result.errors)}")
|
||||
success_count = (
|
||||
result.testsRun - len(result.failures) - len(result.errors)
|
||||
)
|
||||
print(f" Success: {success_count}")
|
||||
print(f" Failed: {len(result.failures)}")
|
||||
print(f" Errors: {len(result.errors)}")
|
||||
print("=" * 100)
|
||||
|
||||
@ -0,0 +1,307 @@
|
||||
#!/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 RuijieTelnetEnhanced driver
|
||||
#
|
||||
|
||||
"""
|
||||
Unit test script for RuijieTelnetEnhanced custom device driver.
|
||||
|
||||
This script tests:
|
||||
1. Device type registration
|
||||
2. Inheritance from RuijieOSBase
|
||||
3. Interactive command preprocessing
|
||||
4. Batch send and fallback logic (mocked)
|
||||
|
||||
Run with: python test_ruijie_telnet.py
|
||||
"""
|
||||
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import Mock, patch
|
||||
import os
|
||||
|
||||
# Add project root to path using relative path
|
||||
test_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
project_root = os.path.dirname(
|
||||
os.path.dirname(os.path.dirname(
|
||||
os.path.dirname(os.path.dirname(test_dir)))))
|
||||
sys.path.insert(0, project_root)
|
||||
|
||||
|
||||
class TestRuijieTelnetEnhancedDriver(unittest.TestCase):
|
||||
"""Test suite for RuijieTelnetEnhanced custom driver."""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
"""Set up test fixtures - import and register custom driver."""
|
||||
# Import the custom driver module (triggers registration)
|
||||
from gns3server.agent.gns3_copilot.utils.custom_netmiko import (
|
||||
ruijie_telnet,
|
||||
)
|
||||
|
||||
cls.ruijie_telnet = ruijie_telnet
|
||||
cls.RuijieTelnetEnhanced = ruijie_telnet.RuijieTelnetEnhanced
|
||||
|
||||
def test_device_type_registered(self):
|
||||
"""Test that gns3_ruijie_telnet is registered in Netmiko."""
|
||||
from netmiko.ssh_dispatcher import CLASS_MAPPER, CLASS_MAPPER_BASE
|
||||
|
||||
# Check CLASS_MAPPER
|
||||
self.assertIn("gns3_ruijie_telnet", CLASS_MAPPER)
|
||||
self.assertEqual(
|
||||
CLASS_MAPPER["gns3_ruijie_telnet"],
|
||||
self.RuijieTelnetEnhanced
|
||||
)
|
||||
|
||||
# Check CLASS_MAPPER_BASE
|
||||
self.assertIn("gns3_ruijie_telnet", CLASS_MAPPER_BASE)
|
||||
self.assertEqual(
|
||||
CLASS_MAPPER_BASE["gns3_ruijie_telnet"],
|
||||
self.RuijieTelnetEnhanced
|
||||
)
|
||||
|
||||
def test_inheritance_from_ruijie_os_base(self):
|
||||
"""Test that RuijieTelnetEnhanced inherits from RuijieOSBase."""
|
||||
from netmiko.ruijie.ruijie_os import RuijieOSBase
|
||||
|
||||
# Verify inheritance
|
||||
self.assertIsInstance(self.RuijieTelnetEnhanced, type)
|
||||
# Check if RuijieTelnetEnhanced is a subclass of RuijieOSBase
|
||||
self.assertTrue(issubclass(self.RuijieTelnetEnhanced, RuijieOSBase))
|
||||
|
||||
def test_ruijie_base_methods_available(self):
|
||||
"""Test that Ruijie-specific methods are available."""
|
||||
# These methods should be inherited from RuijieOSBase
|
||||
ruijie_methods = [
|
||||
"config_mode",
|
||||
"check_config_mode",
|
||||
"exit_config_mode",
|
||||
"send_config_set",
|
||||
"send_command",
|
||||
"disable_paging",
|
||||
]
|
||||
|
||||
for method_name in ruijie_methods:
|
||||
self.assertTrue(
|
||||
hasattr(self.RuijieTelnetEnhanced, method_name),
|
||||
f"Method {method_name} not found in RuijieTelnetEnhanced",
|
||||
)
|
||||
|
||||
def test_interactive_patterns_defined(self):
|
||||
"""Test that INTERACTIVE_PATTERNS is correctly defined."""
|
||||
import re
|
||||
|
||||
# Should have INTERACTIVE_PATTERNS as a class attribute
|
||||
self.assertTrue(
|
||||
hasattr(self.RuijieTelnetEnhanced, "INTERACTIVE_PATTERNS")
|
||||
)
|
||||
|
||||
# Check it's a list
|
||||
patterns = self.RuijieTelnetEnhanced.INTERACTIVE_PATTERNS
|
||||
self.assertIsInstance(patterns, list)
|
||||
|
||||
# Check it contains expected patterns (compiled regex)
|
||||
self.assertGreater(len(patterns), 0)
|
||||
for pattern in patterns:
|
||||
self.assertIsInstance(pattern, re.Pattern)
|
||||
|
||||
def test_preprocess_interactive_commands(self):
|
||||
"""Test that interactive commands are preprocessed correctly."""
|
||||
# Create a mock instance (without actual connection)
|
||||
with patch.object(
|
||||
self.RuijieTelnetEnhanced, "__init__",
|
||||
lambda self, *args, **kwargs: None
|
||||
):
|
||||
instance = self.RuijieTelnetEnhanced.__new__(
|
||||
self.RuijieTelnetEnhanced
|
||||
)
|
||||
|
||||
# Set INTERACTIVE_PATTERNS from class
|
||||
instance.INTERACTIVE_PATTERNS = (
|
||||
self.RuijieTelnetEnhanced.INTERACTIVE_PATTERNS
|
||||
)
|
||||
|
||||
# Test commands with router-id (should trigger 'yes')
|
||||
commands = [
|
||||
"router ospf 1",
|
||||
"router-id 1.1.1.1",
|
||||
"network 10.0.0.0 0.0.0.255 area 0",
|
||||
]
|
||||
|
||||
processed = instance._preprocess_interactive_commands(
|
||||
commands
|
||||
)
|
||||
|
||||
# Should have inserted 'yes' after router-id command
|
||||
self.assertEqual(len(processed), 4)
|
||||
self.assertEqual(processed[0], "router ospf 1")
|
||||
self.assertEqual(processed[1], "router-id 1.1.1.1")
|
||||
self.assertEqual(processed[2], "yes")
|
||||
self.assertEqual(
|
||||
processed[3], "network 10.0.0.0 0.0.0.255 area 0"
|
||||
)
|
||||
|
||||
def test_preprocess_non_interactive_commands(self):
|
||||
"""Test that non-interactive commands are not modified."""
|
||||
# Create a mock instance
|
||||
with patch.object(
|
||||
self.RuijieTelnetEnhanced, "__init__",
|
||||
lambda self, *args, **kwargs: None
|
||||
):
|
||||
instance = self.RuijieTelnetEnhanced.__new__(
|
||||
self.RuijieTelnetEnhanced
|
||||
)
|
||||
|
||||
# Set INTERACTIVE_PATTERNS from class
|
||||
instance.INTERACTIVE_PATTERNS = (
|
||||
self.RuijieTelnetEnhanced.INTERACTIVE_PATTERNS
|
||||
)
|
||||
|
||||
# Test commands without interactive prompts
|
||||
commands = [
|
||||
"interface GigabitEthernet 0/1",
|
||||
"description Test Interface",
|
||||
"ip address 192.168.1.1 255.255.255.0",
|
||||
]
|
||||
|
||||
processed = instance._preprocess_interactive_commands(commands)
|
||||
|
||||
# Should not have inserted any 'yes' commands
|
||||
self.assertEqual(len(processed), 3)
|
||||
self.assertEqual(processed, commands)
|
||||
|
||||
def test_connect_handler_accepts_device_type(self):
|
||||
"""Test that ConnectHandler accepts gns3_ruijie_telnet."""
|
||||
from netmiko.ssh_dispatcher import CLASS_MAPPER
|
||||
|
||||
# Get platforms list
|
||||
platforms = list(CLASS_MAPPER.keys())
|
||||
|
||||
# Verify gns3_ruijie_telnet is in platforms
|
||||
self.assertIn("gns3_ruijie_telnet", platforms)
|
||||
|
||||
# Verify it's in telnet platforms
|
||||
telnet_platforms = [x for x in platforms if "telnet" in x]
|
||||
self.assertIn("gns3_ruijie_telnet", telnet_platforms)
|
||||
|
||||
def test_default_enter_parameter(self):
|
||||
"""Test that default_enter is set to \\r\\n for telnet."""
|
||||
import inspect
|
||||
|
||||
# Check that default_enter is handled in __init__
|
||||
# (The actual logic is in the __init__ method body)
|
||||
init_source = inspect.getsource(
|
||||
self.RuijieTelnetEnhanced.__init__
|
||||
)
|
||||
self.assertIn("default_enter", init_source)
|
||||
self.assertIn("\\r\\n", init_source)
|
||||
|
||||
|
||||
class TestRuijieTelnetEnhancedIntegration(unittest.TestCase):
|
||||
"""Integration tests for RuijieTelnetEnhanced driver."""
|
||||
|
||||
def test_send_config_set_calls_preprocess(self):
|
||||
"""Test that send_config_set calls preprocessing."""
|
||||
from gns3server.agent.gns3_copilot.utils.custom_netmiko.ruijie_telnet import ( # noqa: E501
|
||||
RuijieTelnetEnhanced,
|
||||
)
|
||||
|
||||
# Create a mock instance
|
||||
instance = RuijieTelnetEnhanced.__new__(
|
||||
RuijieTelnetEnhanced
|
||||
)
|
||||
|
||||
# Mock the necessary attributes and methods
|
||||
instance.RETURN = "\r\n"
|
||||
instance.global_delay_factor = 1.0
|
||||
|
||||
# Mock config_mode and exit_config_mode
|
||||
instance.config_mode = Mock(return_value="")
|
||||
instance.exit_config_mode = Mock(return_value="")
|
||||
instance._send_config_batch = Mock(return_value="Output")
|
||||
|
||||
# Test commands
|
||||
commands = ["interface GigabitEthernet 0/1", "description Test"]
|
||||
|
||||
# Call send_config_set
|
||||
instance.send_config_set(
|
||||
commands, enter_config_mode=False, exit_config_mode=False
|
||||
)
|
||||
|
||||
# Verify _send_config_batch was called
|
||||
instance._send_config_batch.assert_called_once()
|
||||
|
||||
def test_multiple_interactive_patterns(self):
|
||||
"""Test that multiple interactive commands are detected."""
|
||||
from gns3server.agent.gns3_copilot.utils.custom_netmiko.ruijie_telnet import ( # noqa: E501
|
||||
RuijieTelnetEnhanced,
|
||||
)
|
||||
|
||||
# Create a mock instance
|
||||
instance = RuijieTelnetEnhanced.__new__(
|
||||
RuijieTelnetEnhanced
|
||||
)
|
||||
instance.INTERACTIVE_PATTERNS = (
|
||||
RuijieTelnetEnhanced.INTERACTIVE_PATTERNS
|
||||
)
|
||||
|
||||
# Test commands with multiple interactive prompts
|
||||
commands = [
|
||||
"router-id 1.1.1.1", # Should trigger 'yes'
|
||||
"erase startup-config", # Should trigger 'yes'
|
||||
"interface GigabitEthernet 0/1", # Should NOT trigger
|
||||
]
|
||||
|
||||
processed = instance._preprocess_interactive_commands(commands)
|
||||
|
||||
# Should have inserted 'yes' after first two commands
|
||||
# processed[0] = "router-id 1.1.1.1", processed[1] = "yes"
|
||||
# processed[2] = "erase startup-config", processed[3] = "yes"
|
||||
# processed[4] = "interface GigabitEthernet 0/1"
|
||||
self.assertEqual(len(processed), 5)
|
||||
self.assertEqual(processed[0], "router-id 1.1.1.1")
|
||||
self.assertEqual(processed[1], "yes")
|
||||
self.assertEqual(processed[2], "erase startup-config")
|
||||
self.assertEqual(processed[3], "yes")
|
||||
self.assertEqual(processed[4], "interface GigabitEthernet 0/1")
|
||||
|
||||
|
||||
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(TestRuijieTelnetEnhancedDriver)
|
||||
)
|
||||
suite.addTests(
|
||||
loader.loadTestsFromTestCase(TestRuijieTelnetEnhancedIntegration)
|
||||
)
|
||||
|
||||
# Run tests
|
||||
runner = unittest.TextTestRunner(verbosity=2)
|
||||
result = runner.run(suite)
|
||||
|
||||
# Print summary
|
||||
print("\n" + "=" * 100)
|
||||
print("Test Summary:")
|
||||
print(f" Run: {result.testsRun}")
|
||||
success_count = (
|
||||
result.testsRun - len(result.failures) - len(result.errors)
|
||||
)
|
||||
print(f" Success: {success_count}")
|
||||
print(f" Failed: {len(result.failures)}")
|
||||
print(f" Errors: {len(result.errors)}")
|
||||
print("=" * 100)
|
||||
|
||||
return result.wasSuccessful()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
success = run_tests()
|
||||
sys.exit(0 if success else 1)
|
||||
Loading…
x
Reference in New Issue
Block a user