diff --git a/docs/gns3-copilot/implemented/multi-vendor-device-support.md b/docs/gns3-copilot/implemented/multi-vendor-device-support.md index f88fbb951..c8faae031 100644 --- a/docs/gns3-copilot/implemented/multi-vendor-device-support.md +++ b/docs/gns3-copilot/implemented/multi-vendor-device-support.md @@ -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. diff --git a/docs/gns3-copilot/netmiko_devices.md b/docs/gns3-copilot/netmiko_devices.md index 61f390dc8..31231965e 100644 --- a/docs/gns3-copilot/netmiko_devices.md +++ b/docs/gns3-copilot/netmiko_devices.md @@ -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 | 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 829d6d0fc..7dfcc3a02 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py @@ -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 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 e96d65a81..f30a79dee 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py @@ -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 diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/README.md b/gns3server/agent/gns3_copilot/utils/custom_netmiko/README.md index 3aade1359..2eea98e5c 100644 --- a/gns3server/agent/gns3_copilot/utils/custom_netmiko/README.md +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/README.md @@ -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, } diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py index 4dc3404c8..7cd39adbf 100644 --- a/gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py @@ -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, } diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/huawei_ce.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/huawei_ce.py index 74d3bf6a4..6d14e05a1 100644 --- a/gns3server/agent/gns3_copilot/utils/custom_netmiko/huawei_ce.py +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/huawei_ce.py @@ -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 diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/ruijie_telnet.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/ruijie_telnet.py index c71157c1e..98f41c895 100644 --- a/gns3server/agent/gns3_copilot/utils/custom_netmiko/ruijie_telnet.py +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/ruijie_telnet.py @@ -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 diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py index b04d6c116..d9edd8c34 100644 --- a/gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py @@ -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 ) 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 index 4c459527e..62a1c3a5c 100644 --- 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 @@ -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) diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_ruijie_telnet.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_ruijie_telnet.py new file mode 100644 index 000000000..682050c9e --- /dev/null +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_ruijie_telnet.py @@ -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)