feat(copilot): add VPCS Telnet driver with Netmiko and ANSI code stripping

Implement custom VPCS driver and unified tool architecture:

   - Add VPCSTelnet custom driver (vpcs_telnet.py)
     - No authentication (direct console access like VPCS behavior)
     - Simple prompt pattern matching (PC\d+>)
     - Automatic ANSI escape code stripping for clean output
     - Config mode methods return empty (VPCS has no config modes)

   - Replace vpcs_tools_telnetlib3.py with vpcs_tools_netmiko.py
     - Migrate from telnetlib3 to Netmiko + Nornir architecture
     - Unified tool architecture matching config/display tools
     - Improved code consistency and maintainability

   - Add comprehensive test coverage (test_vpcs_telnet.py)
     - 30 unit tests covering all VPCS driver functionality
     - Tests for ANSI code stripping, telnet_login, send_command
     - Tests for device registration and initialization

   - Update VPCS built-in template (services/templates.py)
     - Add platform:vpcs and device_type:gns3_vpcs_telnet tags
     - Automatic driver selection without manual configuration

   - Update documentation (docs/)
     - multi-vendor-device-support.md: VPCS driver documentation
     - netmiko_devices.md: Add VPCS to supported devices list
     - README.md: Update multi-vendor support description
This commit is contained in:
YueGuobin committed 2026-03-14 14:25:37 +08:00
1 parent bbe57f34b9
commit e2e2e23cf7
12 files changed
+1767 -521

No files matched your search

@@ -33,6 +33,7 @@ authentication or behavior patterns.
Supported Drivers:
- huawei_ce: GNS3HuaweiTelnetCE for CloudEngine devices (no authentication)
- ruijie_telnet: RuijieTelnetEnhanced for Ruijie devices (interactive prompt handling)
- vpcs_telnet: VPCSTelnet for VPCS virtual PC simulator devices (no authentication)
All drivers use 'gns3_' prefix to clearly distinguish them from Netmiko's
built-in drivers.
@@ -69,4 +70,9 @@ try:
except Exception as e:
logger.warning(f"Failed to import Ruijie driver: {e}", exc_info=True)
__all__ = ["huawei_ce", "ruijie_telnet"]
try:
from . import vpcs_telnet # noqa: F401
except Exception as e:
logger.warning(f"Failed to import VPCS driver: {e}", exc_info=True)
__all__ = ["huawei_ce", "ruijie_telnet", "vpcs_telnet"]
@@ -0,0 +1,495 @@
#!/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 VPCSTelnet driver
#
"""
Unit test script for VPCSTelnet custom device driver.
This script tests:
1. Device type registration
2. Inheritance from BaseConnection
3. VPCS-specific methods (telnet_login, session_preparation, send_command)
4. Config mode methods (check_config_mode, config_mode, exit_config_mode)
5. disable_paging method
6. Default parameters (default_enter, global_delay_factor, fast_cli)
Run with: python test_vpcs_telnet.py
"""
import sys
import unittest
from unittest.mock import Mock, patch, MagicMock
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 TestVPCSTelnetDriver(unittest.TestCase):
"""Test suite for VPCSTelnet 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 (
vpcs_telnet,
)
cls.vpcs_telnet = vpcs_telnet
cls.VPCSTelnet = vpcs_telnet.VPCSTelnet
def test_device_type_registered(self):
"""Test that gns3_vpcs_telnet is registered in Netmiko."""
from netmiko.ssh_dispatcher import CLASS_MAPPER, CLASS_MAPPER_BASE
# Check CLASS_MAPPER
self.assertIn("gns3_vpcs_telnet", CLASS_MAPPER)
self.assertEqual(
CLASS_MAPPER["gns3_vpcs_telnet"],
self.VPCSTelnet
)
# Check CLASS_MAPPER_BASE
self.assertIn("gns3_vpcs_telnet", CLASS_MAPPER_BASE)
self.assertEqual(
CLASS_MAPPER_BASE["gns3_vpcs_telnet"],
self.VPCSTelnet
)
def test_inheritance_from_base_connection(self):
"""Test that VPCSTelnet inherits from BaseConnection."""
from netmiko.base_connection import BaseConnection
# Verify inheritance
self.assertIsInstance(self.VPCSTelnet, type)
# Check if VPCSTelnet is a subclass of BaseConnection
self.assertTrue(issubclass(self.VPCSTelnet, BaseConnection))
def test_vpcs_methods_available(self):
"""Test that VPCS-specific methods are available."""
# These methods should be available in VPCSTelnet
vpcs_methods = [
"telnet_login",
"session_preparation",
"send_command",
"send_command_timing",
"check_config_mode",
"config_mode",
"exit_config_mode",
"disable_paging",
]
for method_name in vpcs_methods:
self.assertTrue(
hasattr(self.VPCSTelnet, method_name),
f"Method {method_name} not found in VPCSTelnet",
)
def test_connect_handler_accepts_device_type(self):
"""Test that ConnectHandler accepts gns3_vpcs_telnet."""
from netmiko.ssh_dispatcher import CLASS_MAPPER
# Get platforms list
platforms = list(CLASS_MAPPER.keys())
# Verify gns3_vpcs_telnet is in platforms
self.assertIn("gns3_vpcs_telnet", platforms)
# Verify it's in telnet platforms
telnet_platforms = [x for x in platforms if "telnet" in x]
self.assertIn("gns3_vpcs_telnet", telnet_platforms)
def test_telnet_platforms_list(self):
"""Test that gns3_vpcs_telnet is in telnet_platforms list."""
from netmiko.ssh_dispatcher import telnet_platforms
self.assertIn("gns3_vpcs_telnet", telnet_platforms)
class TestVPCSTelnetInit(unittest.TestCase):
"""Test VPCSTelnet initialization and default parameters."""
@classmethod
def setUpClass(cls):
"""Set up test fixtures."""
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501
VPCSTelnet,
)
cls.VPCSTelnet = VPCSTelnet
def test_default_enter_parameter(self):
"""Test that default_enter is set to \\r\\n for VPCS."""
import inspect
# Check that default_enter is handled in __init__
init_source = inspect.getsource(self.VPCSTelnet.__init__)
self.assertIn("default_enter", init_source)
self.assertIn("\\r\\n", init_source)
def test_default_parameters_when_none(self):
"""Test default parameters when not provided."""
# Mock the BaseConnection.__init__ to avoid actual connection
# But also set device_type since VPCSTelnet.__init__ sets it
def mock_init(self, *args, **kwargs):
# Simulate what VPCSTelnet.__init__ does
kwargs.setdefault("device_type", "gns3_vpcs_telnet")
self.device_type = kwargs.get("device_type")
with patch.object(
self.VPCSTelnet, "__init__", mock_init
):
instance = self.VPCSTelnet(host="127.0.0.1")
self.assertEqual(instance.device_type, "gns3_vpcs_telnet")
def test_device_type_set(self):
"""Test that device_type is set to gns3_vpcs_telnet."""
# Mock the __init__ to set device_type
def mock_init(self, *args, **kwargs):
# Simulate what VPCSTelnet.__init__ does
kwargs.setdefault("device_type", "gns3_vpcs_telnet")
self.device_type = kwargs.get("device_type")
with patch.object(
self.VPCSTelnet, "__init__", mock_init
):
instance = self.VPCSTelnet(host="127.0.0.1")
self.assertEqual(instance.device_type, "gns3_vpcs_telnet")
class TestVPCSTelnetMethods(unittest.TestCase):
"""Test VPCSTelnet method implementations."""
@classmethod
def setUpClass(cls):
"""Set up test fixtures."""
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501
VPCSTelnet,
)
cls.VPCSTelnet = VPCSTelnet
def test_check_config_mode_always_false(self):
"""Test that check_config_mode always returns False."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
result = instance.check_config_mode()
self.assertFalse(result)
def test_config_mode_returns_empty(self):
"""Test that config_mode returns empty string."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
result = instance.config_mode()
self.assertEqual(result, "")
def test_exit_config_mode_returns_empty(self):
"""Test that exit_config_mode returns empty string."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
result = instance.exit_config_mode()
self.assertEqual(result, "")
def test_disable_paging_returns_empty(self):
"""Test that disable_paging returns empty string."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
result = instance.disable_paging()
self.assertEqual(result, "")
def test_session_preparation_logs_debug(self):
"""Test that session_preparation completes without errors."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
# Should not raise any exception
instance.session_preparation()
class TestVPCSTelnetSendCommand(unittest.TestCase):
"""Test VPCSTelnet send_command method."""
@classmethod
def setUpClass(cls):
"""Set up test fixtures."""
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501
VPCSTelnet,
)
cls.VPCSTelnet = VPCSTelnet
def test_send_command_writes_bytes(self):
"""Test that send_command writes command as bytes."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
# Mock the necessary attributes
instance.remote_conn = MagicMock()
instance.remote_conn.write = MagicMock()
# Test command encoding
command_string = "ip 192.168.1.1 255.255.255.0"
expected_bytes = command_string.encode("ascii") + b"\n"
# Call send_command (will fail at read_until_pattern, but we can
# check the write call)
try:
instance.send_command(command_string)
except Exception:
pass
# Verify write was called with correct bytes
instance.remote_conn.write.assert_called_once_with(expected_bytes)
def test_send_command_default_expect_string(self):
"""Test that send_command uses VPCS prompt as default expect_string."""
import inspect
# Check that expect_string defaults to r"PC\d+>"
source = inspect.getsource(self.VPCSTelnet.send_command)
self.assertIn("expect_string = r\"PC\\d+>\"", source)
def test_send_command_timing_calls_send_command(self):
"""Test that send_command_timing delegates to send_command."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
# Mock send_command
instance.send_command = Mock(return_value="test output")
# Call send_command_timing
result = instance.send_command_timing(
"ip 192.168.1.1 255.255.255.0",
strip_prompt=True,
strip_command=True,
)
# Verify send_command was called
instance.send_command.assert_called_once_with(
command_string="ip 192.168.1.1 255.255.255.0",
strip_prompt=True,
strip_command=True,
)
self.assertEqual(result, "test output")
class TestVPCSTelnetTelnetLogin(unittest.TestCase):
"""Test VPCSTelnet telnet_login method."""
@classmethod
def setUpClass(cls):
"""Set up test fixtures."""
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501
VPCSTelnet,
)
cls.VPCSTelnet = VPCSTelnet
def test_telnet_login_default_prompt_pattern(self):
"""Test that telnet_login has correct default prompt pattern."""
import inspect
# Check default prompt terminator
source = inspect.getsource(self.VPCSTelnet.telnet_login)
self.assertIn('pri_prompt_terminator: str = r"PC\\d+>"', source)
def test_telnet_login_sends_newlines(self):
"""Test that telnet_login sends 4 newlines."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
# Mock necessary attributes
instance.remote_conn = MagicMock()
instance.remote_conn.write = MagicMock()
instance.read_channel = MagicMock(return_value="PC1>")
instance.select_delay_factor = Mock(return_value=1.0)
instance.read_until_pattern = MagicMock(return_value="PC1>")
# Call telnet_login
instance.telnet_login()
# Verify write was called 4 times (4 newlines)
self.assertEqual(instance.remote_conn.write.call_count, 4)
# Verify each call was with b"\n"
for call in instance.remote_conn.write.call_args_list:
self.assertEqual(call[0][0], b"\n")
def test_telnet_login_waits_for_prompt(self):
"""Test that telnet_login waits for VPCS prompt."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
# Mock necessary attributes
instance.remote_conn = MagicMock()
instance.remote_conn.write = MagicMock()
instance.read_channel = MagicMock(return_value="PC1>")
instance.select_delay_factor = Mock(return_value=1.0)
instance.read_until_pattern = MagicMock(return_value="PC1>")
# Call telnet_login
result = instance.telnet_login()
# Verify read_until_pattern was called
instance.read_until_pattern.assert_called_once()
class TestVPCSTelnetRegistration(unittest.TestCase):
"""Test VPCSTelnet registration function."""
@classmethod
def setUpClass(cls):
"""Set up test fixtures."""
from gns3server.agent.gns3_copilot.utils.custom_netmiko import (
vpcs_telnet,
)
cls.vpcs_telnet = vpcs_telnet
def test_register_function_exists(self):
"""Test that register_custom_device_type function exists."""
self.assertTrue(
hasattr(self.vpcs_telnet, "register_custom_device_type")
)
def test_register_function_is_callable(self):
"""Test that register_custom_device_type is callable."""
self.assertTrue(
callable(self.vpcs_telnet.register_custom_device_type)
)
class TestVPCSTelnetAnsiStripping(unittest.TestCase):
"""Test VPCSTelnet ANSI escape code stripping."""
@classmethod
def setUpClass(cls):
"""Set up test fixtures."""
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501
VPCSTelnet,
)
cls.VPCSTelnet = VPCSTelnet
def test_strip_ansi_codes_method_exists(self):
"""Test that _strip_ansi_codes method exists."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
self.assertTrue(hasattr(instance, "_strip_ansi_codes"))
self.assertTrue(callable(instance._strip_ansi_codes))
def test_strip_bold_codes(self):
"""Test stripping bold ANSI codes."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
text = "\x1b[1mip\x1b[0m \x1b[4mARG\x1b[0m"
result = instance._strip_ansi_codes(text)
self.assertEqual(result, "ip ARG")
def test_strip_underline_codes(self):
"""Test stripping underline ANSI codes."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
text = "\x1b[4maddress\x1b[0m [\x1b[4mmask\x1b[0m]"
result = instance._strip_ansi_codes(text)
self.assertEqual(result, "address [mask]")
def test_strip_reset_codes(self):
"""Test stripping reset ANSI codes."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
text = "\x1b[1mNAME\x1b[0m : PC2[1]"
result = instance._strip_ansi_codes(text)
# Note: [1] at the end is not a valid ANSI code, so it should remain
self.assertEqual(result, "NAME : PC2[1]")
def test_strip_color_codes(self):
"""Test stripping color ANSI codes."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
text = "\x1b[31mRed\x1b[32mGreen\x1b[0mNormal"
result = instance._strip_ansi_codes(text)
self.assertEqual(result, "RedGreenNormal")
def test_strip_complex_vpcs_output(self):
"""Test stripping ANSI codes from real VPCS output."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
# Sample VPCS output with ANSI codes (with proper ESC characters)
text = """\x1b[1mip\x1b[0m \x1b[4mARG\x1b[0m ... [[\x1b[4mOPTION\x1b[0m]
Configure the current VPC's IP settings
\x1b[1mNAME\x1b[0m : PC2[1]
IP/MASK : 192.168.1.20/24"""
result = instance._strip_ansi_codes(text)
# Verify ANSI codes are removed
self.assertNotIn("\x1b[1m", result)
self.assertNotIn("\x1b[0m", result)
self.assertNotIn("\x1b[4m", result)
self.assertIn("ip ARG", result)
self.assertIn("NAME", result)
self.assertIn("PC2[1]", result) # [1] is not a valid ANSI escape
def test_strip_empty_string(self):
"""Test stripping ANSI codes from empty string."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
result = instance._strip_ansi_codes("")
self.assertEqual(result, "")
def test_strip_no_ansi_codes(self):
"""Test that text without ANSI codes is unchanged."""
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
text = "NAME : PC2\nIP/MASK : 192.168.1.20/24"
result = instance._strip_ansi_codes(text)
self.assertEqual(result, text)
def test_send_command_calls_strip_ansi(self):
"""Test that send_command calls _strip_ansi_codes."""
from unittest.mock import patch
instance = self.VPCSTelnet.__new__(self.VPCSTelnet)
# Mock necessary attributes
instance.remote_conn = MagicMock()
instance.remote_conn.write = MagicMock()
instance.read_until_pattern = MagicMock(
return_value="\x1b[1mNAME\x1b[0m : PC2\nPC2>"
)
# Mock _strip_ansi_codes to verify it's called
instance._strip_ansi_codes = Mock(
return_value="NAME : PC2\nPC2>"
)
# Call send_command
instance.send_command("show ip")
# Verify _strip_ansi_codes was called
instance._strip_ansi_codes.assert_called_once()
call_arg = instance._strip_ansi_codes.call_args[0][0]
self.assertIn("\x1b[1m", call_arg) # Verify it received raw output
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(TestVPCSTelnetDriver))
suite.addTests(loader.loadTestsFromTestCase(TestVPCSTelnetInit))
suite.addTests(loader.loadTestsFromTestCase(TestVPCSTelnetMethods))
suite.addTests(loader.loadTestsFromTestCase(TestVPCSTelnetSendCommand))
suite.addTests(loader.loadTestsFromTestCase(TestVPCSTelnetTelnetLogin))
suite.addTests(loader.loadTestsFromTestCase(TestVPCSTelnetRegistration))
suite.addTests(loader.loadTestsFromTestCase(TestVPCSTelnetAnsiStripping))
# 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)
@@ -0,0 +1,424 @@
# 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 <https://www.gnu.org/licenses/>.
#
# Copyright (C) 2025 Yue Guobin (岳国宾)
# Author: Yue Guobin (岳国宾)
#
# Project Home: https://github.com/yueguobin/gns3-copilot
#
# mypy: ignore-errors
"""
Custom Netmiko device driver for VPCS (Virtual PC Simulator) in GNS3.
VPCS is a lightweight virtual PC simulator used in GNS3 lab environments
for testing network connectivity and basic IP configuration.
Key Features:
- No authentication required (direct console access)
- Simple command-line interface (PC1>, PC2>, etc.)
- Supports basic PC commands (ip, ping, arp, version, etc.)
- Telnet-based connection
Device Type: gns3_vpcs_telnet
"""
import importlib
import logging
import re
import time
from netmiko.base_connection import BaseConnection
logger = logging.getLogger(__name__)
# ANSI escape code pattern for stripping terminal formatting codes
# Matches sequences like: [1m (bold), [0m (reset), [4m (underline), etc.
ANSI_ESCAPE_PATTERN = re.compile(r"\x1B\[[0-?]*[ -/]*[@-~]")
class VPCSTelnet(BaseConnection):
"""
Custom VPCS device driver for GNS3 emulation.
VPCS devices are lightweight virtual PCs that simulate basic network
functionality for lab testing. This driver handles the unique
characteristics of VPCS:
- No authentication (no username/password)
- Simple prompt pattern (PC1>, PC2>, etc.)
- No configuration modes
"""
def __init__(
self,
*args,
**kwargs,
) -> None:
"""Initialize VPCS Telnet connection."""
# Set device type to identify this as a VPCS device
kwargs.setdefault("device_type", "gns3_vpcs_telnet")
# VPCS uses carriage return + line feed for command termination
# This is critical for VPCS devices to respond to commands
default_enter = kwargs.get("default_enter")
if default_enter is None:
kwargs["default_enter"] = "\r\n"
kwargs.setdefault("global_delay_factor", 1.0)
kwargs.setdefault("fast_cli", False)
super().__init__(*args, **kwargs)
def telnet_login(
self,
pri_prompt_terminator: str = r"PC\d+>",
alt_prompt_terminator: str = r"",
username_pattern: str = r"(?:user:|username|login|user name)",
pwd_pattern: str = r"assword",
delay_factor: float = 1.0,
max_loops: int = 20,
) -> str:
"""
Telnet login for VPCS devices (no authentication).
VPCS devices connect directly to command line without username/password.
This method initializes the connection and waits for the VPCS prompt.
Strategy (matching telnetlib3 implementation):
1. Send 4 newlines with 0.5s delay between each
2. Wait for VPCS prompt pattern (PC1>, PC2>, etc.)
3. Return once prompt is detected
Args:
pri_prompt_terminator: Primary prompt pattern (default: r"PC\\d+>")
alt_prompt_terminator: Not used for VPCS
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)
return_msg = ""
# Step 1: Clear buffer - read any existing data
try:
initial_data = self.read_channel()
if initial_data:
return_msg += initial_data
except Exception:
# Ignore errors during initial read
pass
# Step 2: Send 4 newlines with delays (matching telnetlib3 exactly)
# tn.write(b"\n"); sleep(0.5) - repeated 4 times
for i in range(4):
try:
# Send newline directly as bytes
self.remote_conn.write(b"\n")
time.sleep(0.5 * delay_factor)
# Read response
new_output = self.read_channel()
if new_output:
return_msg += new_output
except Exception as e:
logger.debug("Error during VPCS connection initialization: %s", e)
# Step 3: Wait for VPCS prompt pattern
try:
# Read until we see the prompt
output = self.read_until_pattern(
pattern=pri_prompt_terminator,
read_timeout=10
)
return_msg += output
if re.search(pri_prompt_terminator, return_msg, flags=re.M):
logger.info("VPCS prompt detected")
return return_msg
except Exception as e:
logger.debug("Error waiting for VPCS prompt: %s", e)
# Step 4: Return what we have (connection might still work)
logger.warning("VPCS prompt not clearly detected, returning current output")
return return_msg
def session_preparation(self) -> None:
"""
Prepare the session after connection is established.
VPCS doesn't require special session preparation.
The telnet_login method already handles initialization with 4 newlines.
"""
# No additional preparation needed
logger.debug("VPCS session preparation completed")
def send_command(
self,
command_string: str,
expect_string: str | None = None,
read_timeout: float | None = None,
delay_factor: float | None = None,
max_loops: int | None = None,
strip_prompt: bool = False,
strip_command: bool = False,
normalize: bool = False,
use_textfsm: bool = False,
) -> str:
"""
Send a command to VPCS and return the output.
This method exactly matches the telnetlib3 implementation:
1. Encode command as ASCII and append newline
2. Write command+newline as single byte sequence
3. Sleep for 5 seconds
4. Wait for prompt pattern and capture output
5. Return captured output
Args:
command_string: The command to send
expect_string: Pattern to expect (default: VPCS prompt r"PC\\d+>")
read_timeout: Timeout for reading output
delay_factor: Not used (kept for signature compatibility)
max_loops: Not used (kept for signature compatibility)
strip_prompt: Remove prompt from output (default: False for debugging)
strip_command: Remove command from output (default: False for debugging)
normalize: Not used for VPCS
use_textfsm: Not used for VPCS
Returns:
Command output from VPCS
"""
# Use VPCS prompt as default expect_string
if expect_string is None:
expect_string = r"PC\d+>"
# CRITICAL: Match telnetlib3 behavior exactly
# tn.write(command.encode(encoding="ascii") + b"\n")
cmd_bytes = command_string.encode(encoding="ascii") + b"\n"
self.remote_conn.write(cmd_bytes)
logger.debug("VPCS: Sent command '%s' (%d bytes)", command_string, len(cmd_bytes))
# Sleep for 5 seconds (matching telnetlib3: sleep(5))
time.sleep(5)
# Wait for prompt pattern and capture output
# This matches telnetlib3: tn.expect([rb"PC\d+>"]) + tn.read_very_eager()
# read_until_pattern returns the output including the matched pattern
try:
output = self.read_until_pattern(pattern=expect_string, read_timeout=10)
logger.debug("VPCS: read_until_pattern returned %d chars", len(output))
except Exception as e:
# If pattern not found, try to read whatever is available
logger.debug("VPCS: Pattern not found, reading available output: %s", e)
output = self.read_channel()
logger.debug("VPCS: read_channel returned %d chars", len(output) if output else 0)
if isinstance(output, bytes):
output = output.decode("utf-8", errors="replace")
logger.debug("VPCS: Raw output before strip: '%s'", output)
# Strip ANSI escape codes from VPCS output
# VPCS uses terminal formatting codes (bold, underline, etc.)
# that need to be removed for clean output parsing
output = self._strip_ansi_codes(output)
# Strip command and prompt if requested (currently disabled for debugging)
if strip_command:
output = self.strip_command_packets(command_string, output)
if strip_prompt:
output = self.strip_prompt(output)
logger.debug("VPCS: Final output after strip: '%s'", output)
return output
def send_command_timing(
self,
command_string: str,
delay_factor: float | None = None,
max_loops: int | None = None,
strip_prompt: bool = True,
strip_command: bool = True,
normalize: bool = False,
) -> str:
"""
Send command using timing-based delay.
For VPCS, timing-based and pattern-based methods are identical,
both use the telnetlib3 implementation pattern with fixed 5s delay.
Args:
command_string: The command to send
delay_factor: Not used (fixed 5s delay for VPCS)
max_loops: Not used (kept for signature compatibility)
strip_prompt: Remove prompt from output
strip_command: Remove command from output
normalize: Not used for VPCS
Returns:
Command output from VPCS
"""
# For VPCS, timing and pattern-based methods are identical
return self.send_command(
command_string=command_string,
strip_prompt=strip_prompt,
strip_command=strip_command,
)
def check_config_mode(
self, check_string: str = "", pattern: str = "", force_regex: bool = False
) -> bool:
"""
VPCS has no configuration mode.
This method always returns False for VPCS devices.
Returns:
False (VPCS has no config mode)
"""
return False
def _strip_ansi_codes(self, text: str) -> str:
"""
Strip ANSI escape codes from VPCS output.
VPCS uses terminal formatting codes (bold, underline, colors, etc.)
that need to be removed for clean output parsing. This method removes
all ANSI escape sequences from the given text.
Examples of codes removed:
- [1m - Bold text
- [0m - Reset formatting
- [4m - Underline text
- [30-47m - Color codes
Args:
text: Text potentially containing ANSI escape codes
Returns:
Text with all ANSI escape codes removed
"""
return ANSI_ESCAPE_PATTERN.sub("", text)
def config_mode(
self, config_command: str = "", pattern: str = "", re_flags: int = 0
) -> str:
"""
VPCS has no configuration mode.
This method returns an empty string for VPCS devices.
Returns:
Empty string (no config mode to enter)
"""
return ""
def exit_config_mode(self, exit_config: str = "", pattern: str = "") -> str:
"""
VPCS has no configuration mode.
This method returns an empty string for VPCS devices.
Returns:
Empty string (no config mode to exit)
"""
return ""
def disable_paging(
self,
command: str = "",
delay_factor: float | None = None,
**kwargs,
) -> str:
"""
VPCS doesn't use paging.
This method returns an empty string for VPCS devices.
Returns:
Empty string (no paging to disable)
"""
return ""
# Register the custom device type with Netmiko
def register_custom_device_type() -> None:
"""
Register the custom VPCS Telnet device type with Netmiko.
This function adds 'gns3_vpcs_telnet' to Netmiko's CLASS_MAPPER
and updates the platforms lists.
IMPORTANT: This function should be called BEFORE using the VPCS device
type with Netmiko.
"""
# Use importlib to avoid namespace conflicts
sd = importlib.import_module("netmiko.ssh_dispatcher")
# Register the device type in both mappers
# CLASS_MAPPER_BASE is used for base class definitions
sd.CLASS_MAPPER_BASE["gns3_vpcs_telnet"] = VPCSTelnet
# CLASS_MAPPER is used by ConnectHandler for device type validation
sd.CLASS_MAPPER["gns3_vpcs_telnet"] = VPCSTelnet
# 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)
logger.info("Successfully registered VPCS Telnet device type with Netmiko")
# Auto-register on import
# This ensures the device type is available when the module is imported
try:
register_custom_device_type()
except Exception as e:
# Log but don't fail on import
logger.warning(
"Failed to register VPCS device type: %s",
e,
exc_info=True
)