NgTech-LTD/gns3-server
2
0 Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-11-03 14:01:15 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor: ruff autofixes

Browse Source
This commit is contained in:
Mark Paronyan committed 2026-09-27 15:55:10 +03:00
1 parent 2ac637c662
commit a0450b6bce
455 files changed
+12322 -14762
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

gns3server/agent/gns3_copilot/utils/command_filter.py
+8 -8
View File
@@ -73,14 +73,14 @@ def _load_forbidden_commands() -> list[str]:
# Try to load from skills repository
manager = get_skills_manager()
if manager is not None:
commands = manager.load_forbidden_commands()
if commands:
logger.info(
"Loaded %d forbidden command patterns from skills repository",
len(commands),
)
_forbidden_commands_cache = commands
return _forbidden_commands_cache
commands = manager.load_forbidden_commands()
if commands:
logger.info(
"Loaded %d forbidden command patterns from skills repository",
len(commands),
)
_forbidden_commands_cache = commands
return _forbidden_commands_cache
# Fallback to defaults
logger.warning("Using default forbidden commands list")
gns3server/agent/gns3_copilot/utils/custom_netmiko/README.md
+6 -5
View File
@@ -69,11 +69,9 @@ The drivers are auto-registered when the tools are imported:
from gns3server.agent.gns3_copilot.tools_v2 import DisplayToolNornir
tool = DisplayToolNornir()
result = tool._run(json.dumps({
"device_names": ["huawei-sw1"],
"commands": ["display version"],
"project_id": "project-uuid"
}))
result = tool._run(
json.dumps({"device_names": ["huawei-sw1"], "commands": ["display version"], "project_id": "project-uuid"})
)
```
## Running Tests
@@ -145,11 +143,14 @@ Create `tests/test_cisco.py`:
import unittest
from gns3server.agent.gns3_copilot.utils.custom_netmiko import cisco
class TestCustomCiscoDriver(unittest.TestCase):
def test_device_type_registered(self):
from netmiko.ssh_dispatcher import CLASS_MAPPER
self.assertIn("cisco_custom", CLASS_MAPPER)
if __name__ == "__main__":
unittest.main()
```
gns3server/agent/gns3_copilot/utils/custom_netmiko/huawei_ce.py
+5 -11
View File
@@ -209,7 +209,7 @@ class GNS3HuaweiTelnetCE(HuaweiBase):
# Ensure we're in a clean state
try:
if hasattr(self, 'base_prompt') and self.base_prompt:
if hasattr(self, "base_prompt") and self.base_prompt:
self._test_channel_read(pattern=self.base_prompt)
else:
# If base_prompt is not set yet, just read to clear buffer
@@ -288,9 +288,7 @@ class GNS3HuaweiTelnetCE(HuaweiBase):
# This method keeps reading until there is no new data for
# 'last_read' seconds. This is the proper Netmiko way to handle
# command output
output += self.read_channel_timing(
read_timeout=read_timeout, last_read=2.0
)
output += self.read_channel_timing(read_timeout=read_timeout, last_read=2.0)
# Exit config mode if requested
if exit_config_mode:
@@ -318,9 +316,7 @@ class GNS3HuaweiTelnetCE(HuaweiBase):
return output
def exit_config_mode(
self, exit_config: str = "return", pattern: str = r"<\S+>|>\s*$"
) -> str:
def exit_config_mode(self, exit_config: str = "return", pattern: str = r"<\S+>|>\s*$") -> str:
r"""
Exit configuration mode for Huawei devices.
@@ -377,6 +373,7 @@ class GNS3HuaweiTelnetCE(HuaweiBase):
# Register the custom device type with Netmiko
_registered = False # Flag to prevent duplicate registration
def register_custom_device_type() -> None:
"""
Register the custom GNS3HuaweiTelnetCE device type with Netmiko.
@@ -451,7 +448,4 @@ try:
except Exception as e:
# Log but don't fail on import
logger = logging.getLogger(__name__)
logger.warning(
f"Failed to register custom device type: {e}",
exc_info=True
)
logger.warning(f"Failed to register custom device type: {e}", exc_info=True)
gns3server/agent/gns3_copilot/utils/custom_netmiko/ruijie_telnet.py
+20 -43
View File
@@ -60,12 +60,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 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__(
@@ -80,9 +80,7 @@ class RuijieTelnetEnhanced(RuijieOSBase):
kwargs["default_enter"] = "\r\n"
super().__init__(*args, **kwargs)
def _preprocess_interactive_commands(
self, config_commands: list[str]
) -> list[str]:
def _preprocess_interactive_commands(self, config_commands: list[str]) -> list[str]:
"""
Preprocess commands to insert 'yes' after interactive commands.
@@ -105,8 +103,7 @@ class RuijieTelnetEnhanced(RuijieOSBase):
for pattern in self.INTERACTIVE_PATTERNS:
if pattern.match(cmd.strip()):
logging.info(
"Ruijie device: Detected interactive command '%s', "
"inserting 'yes'",
"Ruijie device: Detected interactive command '%s', inserting 'yes'",
cmd.strip(),
)
# Insert 'yes' after this command
@@ -153,25 +150,18 @@ class RuijieTelnetEnhanced(RuijieOSBase):
output += self.config_mode()
# Preprocess: Insert 'yes' after known interactive commands
processed_commands = self._preprocess_interactive_commands(
config_commands
)
processed_commands = self._preprocess_interactive_commands(config_commands)
# Try batch send first (fast path)
try:
output += self._send_config_batch(
processed_commands, read_timeout, delay_factor
)
output += self._send_config_batch(processed_commands, read_timeout, delay_factor)
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
output += self._send_config_one_by_one(
config_commands, read_timeout, delay_factor
)
output += self._send_config_one_by_one(config_commands, read_timeout, delay_factor)
# Exit config mode if requested
if exit_config_mode:
@@ -179,9 +169,7 @@ class RuijieTelnetEnhanced(RuijieOSBase):
return output
def _send_config_batch(
self, commands: list[str], read_timeout: int, delay_factor: float
) -> str:
def _send_config_batch(self, commands: list[str], read_timeout: int, delay_factor: float) -> str:
"""
Send configuration commands in batch (fast).
@@ -197,15 +185,11 @@ class RuijieTelnetEnhanced(RuijieOSBase):
# Read all output at once
# Use same default as Netmiko (2.0 seconds) to ensure complete output
output += self.read_channel_timing(
read_timeout=read_timeout, last_read=2.0
)
output += self.read_channel_timing(read_timeout=read_timeout, last_read=2.0)
return output
def _send_config_one_by_one(
self, commands: list[str], read_timeout: int, delay_factor: float
) -> str:
def _send_config_one_by_one(self, commands: list[str], read_timeout: int, delay_factor: float) -> str:
"""
Send commands one-by-one with real-time prompt detection.
@@ -225,26 +209,21 @@ class RuijieTelnetEnhanced(RuijieOSBase):
time.sleep(0.05)
# Read output after this command
new_output = self.read_channel_timing(
read_timeout=10, last_read=0.5
)
new_output = self.read_channel_timing(read_timeout=10, last_read=0.5)
output += new_output
# Check if interactive prompt appeared after this command
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
self.write_channel(f"yes{self.RETURN}")
time.sleep(0.3)
# Read the confirmation response
output += self.read_channel_timing(
read_timeout=30, last_read=0.5
)
output += self.read_channel_timing(read_timeout=30, last_read=0.5)
break
return output
@@ -253,6 +232,7 @@ class RuijieTelnetEnhanced(RuijieOSBase):
# Register the custom device type with Netmiko
_registered = False # Flag to prevent duplicate registration
def register_custom_device_type() -> None:
"""
Register the custom RuijieTelnetEnhanced device type with Netmiko.
@@ -295,7 +275,4 @@ try:
register_custom_device_type()
except Exception as e:
logger = logging.getLogger(__name__)
logger.warning(
f"Failed to register Ruijie device type: {e}",
exc_info=True
)
logger.warning(f"Failed to register Ruijie device type: {e}", exc_info=True)
gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py
+44 -59
View File
@@ -39,6 +39,7 @@ def get_netmiko_version() -> str:
"""Get Netmiko library version."""
try:
import netmiko
return netmiko.__version__
except Exception:
return "Unknown"
@@ -68,47 +69,39 @@ def register_custom_drivers() -> set:
try:
# Import and register Huawei CE driver
from gns3server.agent.gns3_copilot.utils.custom_netmiko \
import huawei_ce
from gns3server.agent.gns3_copilot.utils.custom_netmiko import huawei_ce
# Store custom device types before registration
custom_devices.add("gns3_huawei_telnet_ce")
# The driver auto-registers on import, but we can call it
# explicitly
if hasattr(huawei_ce, 'register_custom_device_type'):
if hasattr(huawei_ce, "register_custom_device_type"):
huawei_ce.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 Huawei CE driver: {e}",
file=sys.stderr
)
print(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
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'):
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
)
print(f"Warning: Failed to register Ruijie driver: {e}", file=sys.stderr)
return custom_devices
@@ -124,32 +117,32 @@ def extract_brand_name(device_type: str) -> str:
Brand name in a standardized format
"""
# Remove common suffixes
name = device_type.replace('_telnet', '').replace('_ssh', '')
name = name.replace('_serial', '')
name = device_type.replace("_telnet", "").replace("_ssh", "")
name = name.replace("_serial", "")
# Map to standard brand names
brand_mapping = {
'cisco': 'Cisco',
'huawei': 'Huawei',
'juniper': 'Juniper',
'arista': 'Arista',
'hp': 'HP',
'aruba': 'Aruba',
'dell': 'Dell',
'brocade': 'Brocade',
'extreme': 'Extreme',
'ruckus': 'Ruckus',
'ruijie': 'Ruijie (锐捷)',
'zte': 'ZTE (中兴)',
'maipu': 'Maipu (迈普)',
'h3c': 'H3C (华三)',
'nokia': 'Nokia',
'paloalto': 'Palo Alto',
'f5': 'F5',
'checkpoint': 'Check Point',
'generic': 'Generic',
'huaweiyt': 'HuaweiYT',
'vsz': 'Ruckus',
"cisco": "Cisco",
"huawei": "Huawei",
"juniper": "Juniper",
"arista": "Arista",
"hp": "HP",
"aruba": "Aruba",
"dell": "Dell",
"brocade": "Brocade",
"extreme": "Extreme",
"ruckus": "Ruckus",
"ruijie": "Ruijie (锐捷)",
"zte": "ZTE (中兴)",
"maipu": "Maipu (迈普)",
"h3c": "H3C (华三)",
"nokia": "Nokia",
"paloalto": "Palo Alto",
"f5": "F5",
"checkpoint": "Check Point",
"generic": "Generic",
"huaweiyt": "HuaweiYT",
"vsz": "Ruckus",
}
# Check for exact matches first
@@ -163,7 +156,7 @@ def extract_brand_name(device_type: str) -> str:
return value
# Return capitalized version of the first part
return name.split('_')[0].capitalize()
return name.split("_")[0].capitalize()
def group_devices_by_brand(device_types: List[str]) -> Dict[str, List[str]]:
@@ -205,8 +198,8 @@ def get_devices_by_protocol() -> Tuple[List[str], List[str]]:
for device_type in CLASS_MAPPER.keys():
device_lower = device_type.lower()
has_telnet = 'telnet' in device_lower
has_ssh = 'ssh' in device_lower
has_telnet = "telnet" in device_lower
has_ssh = "ssh" in device_lower
if has_telnet and has_ssh:
both_protocols.append(device_type)
@@ -230,17 +223,14 @@ def get_total_device_count() -> int:
"""Get total number of device types in Netmiko."""
try:
from netmiko.ssh_dispatcher import CLASS_MAPPER
return len(CLASS_MAPPER)
except ImportError:
return 0
def generate_markdown(
version: str,
ssh_devices: List[str],
telnet_devices: List[str],
total_count: int,
custom_devices: set
version: str, ssh_devices: List[str], telnet_devices: List[str], total_count: int, custom_devices: set
) -> str:
"""
Generate Markdown documentation with table format.
@@ -261,7 +251,7 @@ def generate_markdown(
md_lines.append("# Netmiko Supported Devices")
md_lines.append("")
md_lines.append(f"**Netmiko Version:** {version}")
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
md_lines.append(f"**Generated:** {timestamp}")
md_lines.append("")
md_lines.append("---")
@@ -313,8 +303,7 @@ def generate_markdown(
# Footer
md_lines.append("---")
md_lines.append("")
md_lines.append("*This document was generated automatically by the "
"Netmiko device list script.*")
md_lines.append("*This document was generated automatically by the Netmiko device list script.*")
return "\n".join(md_lines)
@@ -332,10 +321,8 @@ def save_markdown(content: str, filename: str = "netmiko_devices.md") -> None:
# Go up from: .../gns3server/agent/gns3_copilot/utils/
# custom_netmiko/scripts to project root
# That's 6 levels up
project_root = os.path.dirname(os.path.dirname(
os.path.dirname(os.path.dirname(
os.path.dirname(os.path.dirname(script_dir))))
)
project_root = os.path.dirname(
os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(script_dir)))))
)
# Target directory: docs/gns3-copilot
@@ -348,17 +335,15 @@ def save_markdown(content: str, filename: str = "netmiko_devices.md") -> None:
output_path = os.path.join(target_dir, filename)
try:
with open(output_path, 'w', encoding='utf-8') as f:
with open(output_path, "w", encoding="utf-8") as f:
f.write(content)
print(f"Markdown document saved to: {output_path}")
except IOError as e:
print(f"Error: Failed to save Markdown file: {e}",
file=sys.stderr)
print(f"Error: Failed to save Markdown file: {e}", file=sys.stderr)
sys.exit(1)
def print_summary(version: str, ssh_count: int, telnet_count: int,
total_count: int, output_file: str) -> None:
def print_summary(version: str, ssh_count: int, telnet_count: int, total_count: int, output_file: str) -> None:
"""Print summary to console."""
print("=" * 80)
print("Netmiko Device List Generator")
@@ -402,7 +387,7 @@ def main() -> int:
ssh_devices=ssh_devices,
telnet_devices=telnet_devices,
total_count=total_count,
custom_devices=custom_devices
custom_devices=custom_devices,
)
# Save to file
@@ -415,7 +400,7 @@ def main() -> int:
ssh_count=len(ssh_devices),
telnet_count=len(telnet_devices),
total_count=total_count,
output_file=output_file
output_file=output_file,
)
return 0
gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_huawei_ce.py
+7 -22
View File
@@ -25,9 +25,7 @@ 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)
@@ -103,13 +101,8 @@ class TestHuaweiTelnetCEDriver(unittest.TestCase):
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"
@@ -169,9 +162,7 @@ class TestHuaweiTelnetCEIntegration(unittest.TestCase):
)
# Create a mock instance
instance = GNS3HuaweiTelnetCE.__new__(
GNS3HuaweiTelnetCE
)
instance = GNS3HuaweiTelnetCE.__new__(GNS3HuaweiTelnetCE)
# Mock the necessary attributes and methods
instance.host = "127.0.0.1"
@@ -203,12 +194,8 @@ 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)
@@ -218,9 +205,7 @@ def run_tests():
print("\n" + "=" * 100)
print("Test Summary:")
print(f" Run: {result.testsRun}")
success_count = (
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)}")
gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_ruijie_telnet.py
+20 -58
View File
@@ -25,9 +25,7 @@ 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)
@@ -92,9 +90,7 @@ class TestRuijieTelnetEnhancedDriver(unittest.TestCase):
import re
# Should have INTERACTIVE_PATTERNS as a class attribute
self.assertTrue(
hasattr(self.RuijieTelnetEnhanced, "INTERACTIVE_PATTERNS")
)
self.assertTrue(hasattr(self.RuijieTelnetEnhanced, "INTERACTIVE_PATTERNS"))
# Check it's a list
patterns = self.RuijieTelnetEnhanced.INTERACTIVE_PATTERNS
@@ -108,18 +104,11 @@ class TestRuijieTelnetEnhancedDriver(unittest.TestCase):
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
)
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
)
instance.INTERACTIVE_PATTERNS = self.RuijieTelnetEnhanced.INTERACTIVE_PATTERNS
# Test commands with router-id (should trigger 'yes')
commands = [
@@ -128,34 +117,23 @@ class TestRuijieTelnetEnhancedDriver(unittest.TestCase):
"network 10.0.0.0 0.0.0.255 area 0",
]
processed = instance._preprocess_interactive_commands(
commands
)
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"
)
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
)
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
)
instance.INTERACTIVE_PATTERNS = self.RuijieTelnetEnhanced.INTERACTIVE_PATTERNS
# Test commands without interactive prompts
commands = [
@@ -190,9 +168,7 @@ class TestRuijieTelnetEnhancedDriver(unittest.TestCase):
# Check that default_enter is handled in __init__
# (The actual logic is in the __init__ method body)
init_source = inspect.getsource(
self.RuijieTelnetEnhanced.__init__
)
init_source = inspect.getsource(self.RuijieTelnetEnhanced.__init__)
self.assertIn("default_enter", init_source)
self.assertIn("\\r\\n", init_source)
@@ -207,9 +183,7 @@ class TestRuijieTelnetEnhancedIntegration(unittest.TestCase):
)
# Create a mock instance
instance = RuijieTelnetEnhanced.__new__(
RuijieTelnetEnhanced
)
instance = RuijieTelnetEnhanced.__new__(RuijieTelnetEnhanced)
# Mock the necessary attributes and methods
instance.RETURN = "\r\n"
@@ -224,9 +198,7 @@ class TestRuijieTelnetEnhancedIntegration(unittest.TestCase):
commands = ["interface GigabitEthernet 0/1", "description Test"]
# Call send_config_set
instance.send_config_set(
commands, enter_config_mode=False, exit_config_mode=False
)
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()
@@ -238,17 +210,13 @@ class TestRuijieTelnetEnhancedIntegration(unittest.TestCase):
)
# Create a mock instance
instance = RuijieTelnetEnhanced.__new__(
RuijieTelnetEnhanced
)
instance.INTERACTIVE_PATTERNS = (
RuijieTelnetEnhanced.INTERACTIVE_PATTERNS
)
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'
"router-id 1.1.1.1", # Should trigger 'yes'
"erase startup-config", # Should trigger 'yes'
"interface GigabitEthernet 0/1", # Should NOT trigger
]
@@ -273,12 +241,8 @@ def run_tests():
suite = unittest.TestSuite()
# Add test cases
suite.addTests(
loader.loadTestsFromTestCase(TestRuijieTelnetEnhancedDriver)
)
suite.addTests(
loader.loadTestsFromTestCase(TestRuijieTelnetEnhancedIntegration)
)
suite.addTests(loader.loadTestsFromTestCase(TestRuijieTelnetEnhancedDriver))
suite.addTests(loader.loadTestsFromTestCase(TestRuijieTelnetEnhancedIntegration))
# Run tests
runner = unittest.TextTestRunner(verbosity=2)
@@ -288,9 +252,7 @@ def run_tests():
print("\n" + "=" * 100)
print("Test Summary:")
print(f" Run: {result.testsRun}")
success_count = (
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)}")
gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_vpcs_telnet.py
+17 -25
View File
@@ -27,9 +27,7 @@ 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)
@@ -121,6 +119,7 @@ class TestVPCSTelnetInit(unittest.TestCase):
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501
VPCSTelnet,
)
cls.VPCSTelnet = VPCSTelnet
def test_default_enter_parameter(self):
@@ -134,6 +133,7 @@ class TestVPCSTelnetInit(unittest.TestCase):
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):
@@ -141,23 +141,20 @@ class TestVPCSTelnetInit(unittest.TestCase):
kwargs.setdefault("device_type", "gns3_vpcs_telnet")
self.device_type = kwargs.get("device_type")
with patch.object(
self.VPCSTelnet, "__init__", mock_init
):
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
):
with patch.object(self.VPCSTelnet, "__init__", mock_init):
instance = self.VPCSTelnet(host="127.0.0.1")
self.assertEqual(instance.device_type, "gns3_vpcs_telnet")
@@ -171,6 +168,7 @@ class TestVPCSTelnetMethods(unittest.TestCase):
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):
@@ -213,6 +211,7 @@ class TestVPCSTelnetSendCommand(unittest.TestCase):
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501
VPCSTelnet,
)
cls.VPCSTelnet = VPCSTelnet
def test_send_command_writes_bytes(self):
@@ -243,7 +242,7 @@ class TestVPCSTelnetSendCommand(unittest.TestCase):
# Check that expect_string defaults to r"PC\d+>"
source = inspect.getsource(self.VPCSTelnet.send_command)
self.assertIn("expect_string = r\"PC\\d+>\"", source)
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."""
@@ -277,6 +276,7 @@ class TestVPCSTelnetTelnetLogin(unittest.TestCase):
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):
@@ -335,19 +335,16 @@ class TestVPCSTelnetRegistration(unittest.TestCase):
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")
)
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)
)
self.assertTrue(callable(self.vpcs_telnet.register_custom_device_type))
class TestVPCSTelnetAnsiStripping(unittest.TestCase):
@@ -359,6 +356,7 @@ class TestVPCSTelnetAnsiStripping(unittest.TestCase):
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):
@@ -435,14 +433,10 @@ IP/MASK : 192.168.1.20/24"""
# 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>"
)
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>"
)
instance._strip_ansi_codes = Mock(return_value="NAME : PC2\nPC2>")
# Call send_command
instance.send_command("show ip")
@@ -476,9 +470,7 @@ def run_tests():
print("\n" + "=" * 100)
print("Test Summary:")
print(f" Run: {result.testsRun}")
success_count = (
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)}")
gns3server/agent/gns3_copilot/utils/custom_netmiko/vpcs_telnet.py
+5 -15
View File
@@ -149,10 +149,7 @@ class VPCSTelnet(BaseConnection):
# 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
)
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):
@@ -291,9 +288,7 @@ class VPCSTelnet(BaseConnection):
strip_command=strip_command,
)
def check_config_mode(
self, check_string: str = "", pattern: str = "", force_regex: bool = False
) -> bool:
def check_config_mode(self, check_string: str = "", pattern: str = "", force_regex: bool = False) -> bool:
"""
VPCS has no configuration mode.
@@ -326,9 +321,7 @@ class VPCSTelnet(BaseConnection):
"""
return ANSI_ESCAPE_PATTERN.sub("", text)
def config_mode(
self, config_command: str = "", pattern: str = "", re_flags: int = 0
) -> str:
def config_mode(self, config_command: str = "", pattern: str = "", re_flags: int = 0) -> str:
"""
VPCS has no configuration mode.
@@ -370,6 +363,7 @@ class VPCSTelnet(BaseConnection):
# Register the custom device type with Netmiko
_registered = False # Flag to prevent duplicate registration
def register_custom_device_type() -> None:
"""
Register the custom VPCS Telnet device type with Netmiko.
@@ -431,8 +425,4 @@ 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
)
logger.warning("Failed to register VPCS device type: %s", e, exc_info=True)
gns3server/agent/gns3_copilot/utils/error_handler.py
+1 -4
View File
@@ -61,10 +61,7 @@ def format_error_message(error: Exception) -> str:
# Check if error message contains HTML
if _contains_html(error_str):
logger.warning(
"Detected HTML in error message, likely due to incorrect API "
"configuration or base URL"
)
logger.warning("Detected HTML in error message, likely due to incorrect API configuration or base URL")
return (
"API request failed. The error response indicates an issue with "
"your API configuration. Please check:\n"
gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py
+4 -14
View File
@@ -71,9 +71,7 @@ def get_device_ports_from_topology(
Devices that don't exist or missing console_port will not be included
"""
# Log received parameters
logger.info(
"Called with device_names=%s, project_id=%s", device_names, project_id
)
logger.info("Called with device_names=%s, project_id=%s", device_names, project_id)
try:
# Lazy import to avoid circular dependency
@@ -93,9 +91,7 @@ def get_device_ports_from_topology(
for device_name in device_names:
# Check if device exists in topology
if device_name not in topology.get("nodes", {}):
logger.warning(
"Device '%s' not found in topology", device_name
)
logger.warning("Device '%s' not found in topology", device_name)
continue
node_info = topology["nodes"][device_name]
@@ -133,9 +129,7 @@ def get_device_ports_from_topology(
f"Current tags: {tags}"
)
logger.error(error_msg)
hosts_data[device_name] = {
"error": error_msg
}
hosts_data[device_name] = {"error": error_msg}
continue
logger.debug(
@@ -170,11 +164,7 @@ def get_device_ports_from_topology(
"platform": platform,
"node_type": node_info.get("type"),
"groups": ["network_devices"], # For inheriting hostname, timeout, etc.
"connection_options": {
"netmiko": {
"extras": {"device_type": device_type}
}
},
"connection_options": {"netmiko": {"extras": {"device_type": device_type}}},
}
# Per-node default credentials (seeded from the template appliance
gns3server/agent/gns3_copilot/utils/gns3_drawing_utils.py
+15 -63
View File
@@ -135,14 +135,9 @@ def calculate_two_node_shape(
node2_center_x = node2["x"] + (node2_width / 2)
node2_center_y = node2["y"] + (node2_height / 2)
distance = math.sqrt(
(node2_center_x - node1_center_x) ** 2
+ (node2_center_y - node1_center_y) ** 2
)
distance = math.sqrt((node2_center_x - node1_center_x) ** 2 + (node2_center_y - node1_center_y) ** 2)
angle_rad = math.atan2(
node2_center_y - node1_center_y, node2_center_x - node1_center_x
)
angle_rad = math.atan2(node2_center_y - node1_center_y, node2_center_x - node1_center_x)
angle_deg = round(math.degrees(angle_rad))
angle_rad = math.radians(angle_deg)
@@ -162,16 +157,10 @@ def calculate_two_node_shape(
shape_width = rx * 2
shape_height = ry * 2
svg_x = center_x - (
rx * math.cos(angle_rad) - ry * math.sin(angle_rad)
)
svg_y = center_y - (
rx * math.sin(angle_rad) + ry * math.cos(angle_rad)
)
svg_x = center_x - (rx * math.cos(angle_rad) - ry * math.sin(angle_rad))
svg_y = center_y - (rx * math.sin(angle_rad) + ry * math.cos(angle_rad))
shape_svg = generate_ellipse_svg(
int(rx), int(ry), color_scheme, int(shape_width), int(shape_height)
)
shape_svg = generate_ellipse_svg(int(rx), int(ry), color_scheme, int(shape_width), int(shape_height))
offset_distance = ry * text_offset_ratio
@@ -188,22 +177,12 @@ def calculate_two_node_shape(
else: # rectangle
shape_width = distance
shape_height = max(
node1_width, node1_height, node2_width, node2_height
)
shape_height = max(node1_width, node1_height, node2_width, node2_height)
svg_x = center_x - (
(shape_width / 2) * math.cos(angle_rad)
- (shape_height / 2) * math.sin(angle_rad)
)
svg_y = center_y - (
(shape_width / 2) * math.sin(angle_rad)
+ (shape_height / 2) * math.cos(angle_rad)
)
svg_x = center_x - ((shape_width / 2) * math.cos(angle_rad) - (shape_height / 2) * math.sin(angle_rad))
svg_y = center_y - ((shape_width / 2) * math.sin(angle_rad) + (shape_height / 2) * math.cos(angle_rad))
shape_svg = generate_rectangle_svg(
int(shape_width), int(shape_height), color_scheme
)
shape_svg = generate_rectangle_svg(int(shape_width), int(shape_height), color_scheme)
offset_distance = (shape_height / 2) * text_offset_ratio
@@ -354,33 +333,15 @@ def _get_color_scheme(area_name: str) -> dict[str, Any]:
return COLOR_SCHEMES["NORMAL_AREA"]
# 3. Logical Isolation
if (
"VRF" in label
or "VLAN" in label
or "MSTP" in label
or "VXLAN" in label
or "MPLS" in label
):
if "VRF" in label or "VLAN" in label or "MSTP" in label or "VXLAN" in label or "MPLS" in label:
return COLOR_SCHEMES["ISOLATION"]
# 4. High Availability
if (
"VRRP" in label
or "HSRP" in label
or "HA" in label
or "STACK" in label
or "M-LAG" in label
):
if "VRRP" in label or "HSRP" in label or "HA" in label or "STACK" in label or "M-LAG" in label:
return COLOR_SCHEMES["HIGH_AVAILABILITY"]
# 5. External/Internet
if (
"INET" in label
or "OUT" in label
or "EXTERNAL" in label
or "INTERNET" in label
or "DMZ" in label
):
if "INET" in label or "OUT" in label or "EXTERNAL" in label or "INTERNET" in label or "DMZ" in label:
return COLOR_SCHEMES["EXTERNAL"]
# 6. Management
@@ -392,12 +353,7 @@ def _get_color_scheme(area_name: str) -> dict[str, Any]:
return COLOR_SCHEMES["SECURITY_TRUSTED"]
# 8. Cloud/Tunnel
if (
"TUNNEL" in label
or "CLOUD" in label
or "GRE" in label
or "IPSEC" in label
):
if "TUNNEL" in label or "CLOUD" in label or "GRE" in label or "IPSEC" in label:
return COLOR_SCHEMES["CLOUD_TUNNEL"]
# Legacy keyword matching for backward compatibility
@@ -436,9 +392,7 @@ def calculate_two_node_ellipse(
Wrapper around calculate_two_node_shape with shape_type="ellipse".
"""
result = calculate_two_node_shape(
node1, node2, area_name, "ellipse", text_offset_ratio
)
result = calculate_two_node_shape(node1, node2, area_name, "ellipse", text_offset_ratio)
return {
"ellipse": result["shape"],
"text": result["text"],
@@ -457,9 +411,7 @@ def calculate_two_node_rectangle(
Wrapper around calculate_two_node_shape with shape_type="rectangle".
"""
result = calculate_two_node_shape(
node1, node2, area_name, "rectangle", text_offset_ratio
)
result = calculate_two_node_shape(node1, node2, area_name, "rectangle", text_offset_ratio)
metadata = result["metadata"]
return {
"rectangle": result["shape"],
gns3server/agent/gns3_copilot/utils/llm_config_helper.py
+1 -3
View File
@@ -53,9 +53,7 @@ from fastapi import FastAPI
logger = logging.getLogger(__name__)
async def get_user_llm_config_with_app(
user_id: UUID, app: FastAPI
) -> Optional[Dict[str, Any]]:
async def get_user_llm_config_with_app(user_id: UUID, app: FastAPI) -> Optional[Dict[str, Any]]:
"""
Get user's default LLM model configuration with decrypted API key.
gns3server/agent/gns3_copilot/utils/parse_tool_content.py
+19 -65
View File
@@ -160,9 +160,7 @@ def parse_tool_content(
# Empty string handling
if not content.strip():
result = {}
logger.info(
"Content is empty or whitespace, returning: %s", result
)
logger.info("Content is empty or whitespace, returning: %s", result)
return result
s = content.strip()
@@ -170,9 +168,7 @@ def parse_tool_content(
# Handle empty dictionary case
if s == "{}":
result = {}
logger.info(
"Content is empty dictionary, returning: %s", result
)
logger.info("Content is empty dictionary, returning: %s", result)
return result
# Try to parse as Python literal
@@ -190,9 +186,7 @@ def parse_tool_content(
# Try to parse as JSON
try:
result = json.loads(s)
logger.info(
"Successfully parsed as JSON, returning: %s", result
)
logger.info("Successfully parsed as JSON, returning: %s", result)
return result
except json.JSONDecodeError:
pass
@@ -221,8 +215,7 @@ def parse_tool_content(
# Handle unsupported types
error_msg = ( # type: ignore[unreachable]
"Content must be str, dict, list, int, float, bool, or None, got "
f"{type(content).__name__}"
f"Content must be str, dict, list, int, float, bool, or None, got {type(content).__name__}"
)
logger.error(error_msg)
@@ -238,9 +231,7 @@ def parse_tool_content(
return result
def format_tool_response(
content: str | dict | list | int | float | bool | None, indent: int = 2
) -> str:
def format_tool_response(content: str | dict | list | int | float | bool | None, indent: int = 2) -> str:
"""
Format tool response as a beautiful JSON string for UI display.
@@ -259,9 +250,7 @@ def format_tool_response(
logger.info("format_tool_response parameter indent: %s", indent)
try:
parsed = parse_tool_content(
content, fallback_to_raw=True, strict_mode=False
)
parsed = parse_tool_content(content, fallback_to_raw=True, strict_mode=False)
# Ensure the result can be serialized to JSON
result = json.dumps(parsed, ensure_ascii=False, indent=indent)
logger.info("format_tool_response returning: %s", result)
@@ -270,9 +259,7 @@ def format_tool_response(
# If the parsed result cannot be serialized, convert to string and wrap
logger.error("Cannot serialize parsed result to JSON: %s", e)
try:
result = json.dumps(
{"raw": str(content)}, ensure_ascii=False, indent=indent
)
result = json.dumps({"raw": str(content)}, ensure_ascii=False, indent=indent)
logger.info("format_tool_response returning fallback: %s", result)
return result
except Exception:
@@ -286,16 +273,12 @@ def format_tool_response(
return result
except Exception as e:
logger.error("Error formatting tool response: %s", e)
result = json.dumps(
{"error": str(e)}, ensure_ascii=False, indent=indent
)
result = json.dumps({"error": str(e)}, ensure_ascii=False, indent=indent)
logger.info("format_tool_response returning error: %s", result)
return result
def normalize_tool_response(
response: dict | list | str, tool_name: str = "unknown"
) -> dict:
def normalize_tool_response(response: dict | list | str, tool_name: str = "unknown") -> dict:
"""
Normalize tool response to standard format for consistent frontend display.
@@ -334,11 +317,7 @@ def normalize_tool_response(
}
# Handle error responses
if (
isinstance(response, dict)
and "error" in response
and len(response) == 1
):
if isinstance(response, dict) and "error" in response and len(response) == 1:
return {
"success": False,
"total": 0,
@@ -362,34 +341,21 @@ def normalize_tool_response(
# Handle list responses (most tools return list of device results)
if isinstance(response, list):
successful = sum(
1
for item in response
if isinstance(item, dict) and item.get("status") == "success"
)
successful = sum(1 for item in response if isinstance(item, dict) and item.get("status") == "success")
failed = len(response) - successful
normalized_data = []
for item in response:
if isinstance(item, dict):
normalized_item = {
"id": item.get("device_id")
or item.get("node_id")
or item.get("id")
or "",
"id": item.get("device_id") or item.get("node_id") or item.get("id") or "",
"name": item.get("device_name") or item.get("name") or "",
"status": item.get("status", "unknown"),
}
if normalized_item["status"] == "success":
normalized_item["result"] = (
item.get("output") or item.get("result") or ""
)
normalized_item["result"] = item.get("output") or item.get("result") or ""
else:
normalized_item["error"] = (
item.get("error")
or item.get("output")
or "Unknown error"
)
normalized_item["error"] = item.get("error") or item.get("output") or "Unknown error"
normalized_data.append(normalized_item)
else:
# Non-dict items in list
@@ -426,14 +392,8 @@ def normalize_tool_response(
}
# Legacy format: extract common fields
total = (
response.get("total_nodes")
or response.get("total")
or response.get("count", 1)
)
successful = (
response.get("successful_nodes") or response.get("successful") or 0
)
total = response.get("total_nodes") or response.get("total") or response.get("count", 1)
successful = response.get("successful_nodes") or response.get("successful") or 0
failed = response.get("failed_nodes") or response.get("failed") or 0
# Extract data from various possible locations
@@ -458,13 +418,9 @@ def normalize_tool_response(
if not data and "status" in response:
data = [
{
"name": response.get("device_name")
or response.get("name")
or "",
"name": response.get("device_name") or response.get("name") or "",
"status": response["status"],
"result": response.get("output")
or response.get("result")
or "",
"result": response.get("output") or response.get("result") or "",
"error": response.get("error") or "",
}
]
@@ -564,9 +520,7 @@ def _test_parse_tool_content() -> None:
valid = "✓"
except Exception:
valid = "✗"
print(
f"Format Test {i + 1}: {valid} Input: {repr(input_data)} -> {result}"
)
print(f"Format Test {i + 1}: {valid} Input: {repr(input_data)} -> {result}")
if __name__ == "__main__":
gns3server/agent/gns3_copilot/utils/tool_call_stream.py
+5 -17
View File
@@ -98,9 +98,7 @@ class ToolCallStreamAccumulator:
"id": tc_id,
"type": "function",
"function": {
"name": self._current_tool_call[
"name"
],
"name": self._current_tool_call["name"],
"arguments": "",
},
},
@@ -122,9 +120,7 @@ class ToolCallStreamAccumulator:
# Core: string concatenation
if isinstance(args_chunk, str):
tool_data[
"args_string"
] += args_chunk
tool_data["args_string"] += args_chunk
# Send updated tool_call event with accumulated args
chunks.append(
@@ -135,9 +131,7 @@ class ToolCallStreamAccumulator:
"type": "function",
"function": {
"name": tool_data["name"],
"arguments": tool_data[
"args_string"
],
"arguments": tool_data["args_string"],
},
},
}
@@ -147,15 +141,9 @@ class ToolCallStreamAccumulator:
# complete ==========
# Check finish_reason == "tool_calls" or "STOP"
response_metadata = getattr(chunk, "response_metadata", {})
finish_reason = (
response_metadata.get("finish_reason")
if isinstance(response_metadata, dict)
else None
)
finish_reason = response_metadata.get("finish_reason") if isinstance(response_metadata, dict) else None
if (finish_reason == "tool_calls") or (
finish_reason == "stop" and self._current_tool_call is not None
):
if (finish_reason == "tool_calls") or (finish_reason == "stop" and self._current_tool_call is not None):
if self._current_tool_call:
tool_data = self._current_tool_call