NAPALM/napalm-netgear
1
0 Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

chore: adopt the NAPALM driver standard #15

Merged
christianmanivong merged 5 commits from chore/driver-standard into main 2026-10-23 10:03:35 +00:00
Conversation 0 Commits 5 Files Changed 13
+154 -167
6 changed files with 154 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit fa10d6617b - Show all commits

No files matched your search

napalm_netgear/__init__.py
+1 -1
View File
@@ -1,6 +1,6 @@
"""NAPALM driver for Netgear Smart Managed switches."""
from napalm_netgear.netgear_smart import NetgearSmartDriver
from napalm_netgear.netgear_plus import NetgearPlusDriver
from napalm_netgear.netgear_smart import NetgearSmartDriver
__all__ = ["NetgearSmartDriver", "NetgearPlusDriver"]
napalm_netgear/netgear_plus.py
+41 -60
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
@@ -26,8 +25,8 @@ import threading
import time
from typing import Any
from napalm_device_types import FingerprintRule, FirmwareOfferMixin, SwitchDriver
from napalm.base.exceptions import ConnectionException
from napalm_device_types import FingerprintRule, FirmwareOfferMixin, SwitchDriver
# ---------------------------------------------------------------------------
# Firmware update check — module-level constants and cache
@@ -63,13 +62,12 @@ def _fw_version_tuple(version: str) -> tuple[int, ...]:
def _fw_cdn_exists(model: str, version: str) -> bool:
"""Return True if a firmware zip exists on the Netgear CDN (HEAD request)."""
import urllib.request
import ssl
import urllib.request
url = f"{_FW_CDN_BASE}/{model}/{model}_{version}.zip"
ctx = ssl.create_default_context()
req = urllib.request.Request(url, method="HEAD",
headers={"User-Agent": "Mozilla/5.0"})
req = urllib.request.Request(url, method="HEAD", headers={"User-Agent": "Mozilla/5.0"})
try:
with urllib.request.urlopen(req, timeout=3, context=ctx):
return True
@@ -150,7 +148,15 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
VENDOR = "Netgear"
DRIVER_NAME = "netgear_plus"
SNMP_OBJECT_ID_PREFIX = "1.3.6.1.4.1.4526"
OUI_PREFIXES = ["00:09:5B", "20:4E:7F", "E0:91:F5", "A0:40:A0", "C4:04:15", "00:1B:2F", "00:1F:33"]
OUI_PREFIXES = [
"00:09:5B",
"20:4E:7F",
"E0:91:F5",
"A0:40:A0",
"C4:04:15",
"00:1B:2F",
"00:1F:33",
]
HTTP_FINGERPRINT = [
FingerprintRule("netgear gs1", weight=8.0),
FingerprintRule("netgear gs2", weight=8.0),
@@ -191,8 +197,7 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
from py_netgear_plus import NetgearSwitchConnector
except ImportError as exc:
raise ConnectionException(
"py-netgear-plus is not installed. "
"Add 'py-netgear-plus' to your dependencies."
"py-netgear-plus is not installed. Add 'py-netgear-plus' to your dependencies."
) from exc
try:
@@ -207,9 +212,7 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
except Exception as exc:
if isinstance(exc, ConnectionException):
raise
raise ConnectionException(
f"Cannot connect to {self.hostname}: {exc}"
) from exc
raise ConnectionException(f"Cannot connect to {self.hostname}: {exc}") from exc
def close(self) -> None:
"""Log out from the switch."""
@@ -285,9 +288,7 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
hostname = infos.get("switch_name", self.hostname)
os_version = infos.get("switch_firmware", "")
serial_number = infos.get("switch_serial_number", "")
interface_list = [
self._port_name(i) for i in range(1, self._connector.ports + 1)
]
interface_list = [self._port_name(i) for i in range(1, self._connector.ports + 1)]
return {
"vendor": self.VENDOR,
"model": str(model),
@@ -313,7 +314,9 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
name = self._port_name(port)
status = infos.get(f"port_{port}_status", "off") == "on"
speed = float(infos.get(f"port_{port}_connection_speed", 0) or 0)
description = port_names.get(port, "") or infos.get(f"port_{port}_description", "") or ""
description = (
port_names.get(port, "") or infos.get(f"port_{port}_description", "") or ""
)
interfaces[name] = {
"is_up": status,
"is_enabled": True,
@@ -372,9 +375,7 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
"""
infos = self._get_switch_infos()
power: dict[str, Any] = {}
poe_max_single = getattr(
self._connector.switch_model, "POE_MAX_POWER_SINGLE_PORT", None
)
poe_max_single = getattr(self._connector.switch_model, "POE_MAX_POWER_SINGLE_PORT", None)
for port in self._connector.poe_ports:
poe_active = infos.get(f"port_{port}_poe_power_active")
poe_status = bool(poe_active) if poe_active is not None else False
@@ -450,8 +451,8 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
# Example: <span vid-4>...<i ...></i>24</span><span vnm-4 ...>Guest</span>
vlan_names: dict[str, str] = {}
vlan_entry_re = re.compile(
r'<span\s+vid-4[^>]*>.*?(\d+)\s*</span>\s*'
r'<span\s+vnm-4[^>]*>([^<]*)</span>',
r"<span\s+vid-4[^>]*>.*?(\d+)\s*</span>\s*"
r"<span\s+vnm-4[^>]*>([^<]*)</span>",
re.DOTALL,
)
for m in vlan_entry_re.finditer(text):
@@ -463,15 +464,14 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
return {}
vlans: dict[str, Any] = {
vid: {"name": name, "tagged": [], "untagged": []}
for vid, name in vlan_names.items()
vid: {"name": name, "tagged": [], "untagged": []} for vid, name in vlan_names.items()
}
# Parse per-port PVID membership.
# Each <span pvid-str ...>1*, 24, 25,</span> corresponds to one port
# in order (port 1, port 2, ...). A VID followed by "*" is the native
# (untagged) VLAN for that port; all others are tagged.
pvid_span_re = re.compile(r'<span\s[^>]*pvid-str[^>]*>([^<]+)</span>')
pvid_span_re = re.compile(r"<span\s[^>]*pvid-str[^>]*>([^<]+)</span>")
for port_idx, pvid_str in enumerate(pvid_span_re.finditer(text)):
port_num = port_idx + 1
if port_num > self._connector.ports:
@@ -540,34 +540,24 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
return {"running": "", "startup": "", "candidate": ""}
def load_merge_candidate(self, filename: str | None = None, config: str | None = None) -> None:
raise NotImplementedError(
"Config management is not supported for Netgear Plus switches."
)
raise NotImplementedError("Config management is not supported for Netgear Plus switches.")
def load_replace_candidate(self, filename: str | None = None, config: str | None = None) -> None:
raise NotImplementedError(
"Config management is not supported for Netgear Plus switches."
)
def load_replace_candidate(
self, filename: str | None = None, config: str | None = None
) -> None:
raise NotImplementedError("Config management is not supported for Netgear Plus switches.")
def compare_config(self) -> str:
raise NotImplementedError(
"Config management is not supported for Netgear Plus switches."
)
raise NotImplementedError("Config management is not supported for Netgear Plus switches.")
def commit_config(self, message: str = "") -> None:
raise NotImplementedError(
"Config management is not supported for Netgear Plus switches."
)
raise NotImplementedError("Config management is not supported for Netgear Plus switches.")
def discard_config(self) -> None:
raise NotImplementedError(
"Config management is not supported for Netgear Plus switches."
)
raise NotImplementedError("Config management is not supported for Netgear Plus switches.")
def rollback(self) -> None:
raise NotImplementedError(
"Config management is not supported for Netgear Plus switches."
)
raise NotImplementedError("Config management is not supported for Netgear Plus switches.")
def has_pending_commit(self) -> bool:
return False
@@ -587,9 +577,7 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
Not supported on Netgear Plus switches via HTTP.
"""
raise NotImplementedError(
"ping() is not supported for Netgear Plus switches."
)
raise NotImplementedError("ping() is not supported for Netgear Plus switches.")
def traceroute(
self,
@@ -603,9 +591,7 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
Not supported on Netgear Plus switches via HTTP.
"""
raise NotImplementedError(
"traceroute() is not supported for Netgear Plus switches."
)
raise NotImplementedError("traceroute() is not supported for Netgear Plus switches.")
def cli(self, commands: list[str], encoding: str = "text") -> dict[str, Any]:
"""Send CLI commands.
@@ -689,12 +675,8 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
Reports PoE power status from switch infos.
"""
infos = self._get_switch_infos()
poe_max_all = getattr(
self._connector.switch_model, "POE_MAX_POWER_ALL_PORTS", None
)
poe_max_single = getattr(
self._connector.switch_model, "POE_MAX_POWER_SINGLE_PORT", None
)
poe_max_all = getattr(self._connector.switch_model, "POE_MAX_POWER_ALL_PORTS", None)
poe_max_single = getattr(self._connector.switch_model, "POE_MAX_POWER_SINGLE_PORT", None)
ports: dict[str, Any] = {}
for p in self._connector.poe_ports:
poe_active = infos.get(f"port_{p}_poe_power_active")
@@ -742,9 +724,7 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
Not supported via HTTP API.
"""
raise NotImplementedError(
"set_interface() is not supported for Netgear Plus switches."
)
raise NotImplementedError("set_interface() is not supported for Netgear Plus switches.")
# ── SNMP / Health ──────────────────────────────────────────────────────────
@@ -763,9 +743,10 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
decide (napalm-device-types ``FirmwareOfferMixin``).
"""
try:
model_name: str = getattr(
self._connector.switch_model, "MODEL_NAME", ""
) or self._connector.switch_model.__class__.__name__
model_name: str = (
getattr(self._connector.switch_model, "MODEL_NAME", "")
or self._connector.switch_model.__class__.__name__
)
current_fw: str = self._get_switch_infos().get("switch_firmware", "")
latest_fw = _fw_get_latest(model_name) if model_name and current_fw else None
if latest_fw is None:
napalm_netgear/netgear_smart.py
+99 -102
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
@@ -28,15 +27,11 @@ CLI notes
"""
import re
import socket
import time
from typing import Any, ClassVar, Dict, List, Optional, Union
from typing import Any, ClassVar
import napalm.base.constants as C
import netaddr
from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException
from napalm_device_types import ConfigLifecycleMixin, FingerprintRule, SnmpAgentMixin, SwitchDriver
from napalm.base import helpers as napalm_helpers
from napalm.base.exceptions import (
CommandErrorException,
@@ -46,8 +41,10 @@ from napalm.base.exceptions import (
ReplaceConfigException,
)
from napalm.base.netmiko_helpers import netmiko_args
from napalm_device_types import ConfigLifecycleMixin, FingerprintRule, SnmpAgentMixin, SwitchDriver
from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch
import napalm.base.constants as C
from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException
class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
@@ -73,7 +70,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
#: Where the switch publishes its health values over SNMP
#: (napalm-device-types ``SnmpHealthProfile``); the application polls.
#: CPU is text ("5 Secs ( 3.2%) 60 Secs ( …"); the five-second figure counts.
SNMP_HEALTH_PROFILE: ClassVar[Dict[str, Any]] = {
SNMP_HEALTH_PROFILE: ClassVar[dict[str, Any]] = {
"uptime": {"oids": ["1.3.6.1.2.1.1.3.0"]},
"cpu": {
"oids": ["1.3.6.1.4.1.4526.100.1.1.4.9.0"],
@@ -97,13 +94,13 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
username: str,
password: str,
timeout: int = 60,
optional_args: Optional[Dict] = None,
optional_args: dict | None = None,
) -> None:
self.hostname = hostname
self.username = username
self.password = password
self.timeout = timeout
self.device: Optional[ConnectHandler] = None
self.device: ConnectHandler | None = None
if optional_args is None:
optional_args = {}
@@ -131,16 +128,16 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
self._host_key = HostKeyCheck(optional_args, hostname, self.port)
# Config management state
self._candidate_config: Optional[str] = None
self._candidate_mode: Optional[str] = None # 'merge' or 'replace'
self._backup_config: Optional[str] = None
self._candidate_config: str | None = None
self._candidate_mode: str | None = None # 'merge' or 'replace'
self._backup_config: str | None = None
# CLI style detection cache: True = "v7" (Cisco-like, GigabitEthernet/lag
# interfaces, "show info"/"show interfaces ... status"), False = legacy
# FASTPATH-style (0/1, ch1, "show sysinfo"/"show port all"/"show vlan").
self._cli_v7: Optional[bool] = None
self._port_count_v7: Optional[int] = None
self._version_cache: Optional[str] = None
self._cli_v7: bool | None = None
self._port_count_v7: int | None = None
self._version_cache: str | None = None
# ------------------------------------------------------------------
# Connection management
@@ -179,20 +176,20 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
self.device.disconnect()
self.device = None
def is_alive(self) -> Dict[str, bool]:
def is_alive(self) -> dict[str, bool]:
"""Return connection liveness without writing to the channel."""
if self.device is None:
return {"is_alive": False}
try:
return {"is_alive": self.device.remote_conn.transport.is_active()}
except (socket.error, EOFError, AttributeError):
except (OSError, EOFError, AttributeError):
return {"is_alive": False}
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _send_command(self, command: Union[str, List[str]]) -> str:
def _send_command(self, command: str | list[str]) -> str:
"""Send a command (or ordered list of fallback commands) to the device.
Uses an explicit ``expect_string`` so that Netmiko never confuses a
@@ -216,7 +213,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return output
return output
return _do_send(command)
except (socket.error, EOFError) as exc:
except (OSError, EOFError) as exc:
raise ConnectionClosedException(str(exc)) from exc
# ------------------------------------------------------------------
@@ -260,10 +257,10 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
read_timeout=self.timeout,
)
def _apply_config_lines(self, config_text: str) -> List[str]:
def _apply_config_lines(self, config_text: str) -> list[str]:
"""Send config lines while in config mode; return list of rejected lines."""
ep_any = self._any_prompt()
errors: List[str] = []
errors: list[str] = []
for line in config_text.splitlines():
stripped = line.strip()
if not stripped or stripped.startswith(("!", "#")):
@@ -363,7 +360,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return full
@staticmethod
def _expand_port_list_v7(raw: str) -> List[str]:
def _expand_port_list_v7(raw: str) -> list[str]:
"""Expand a "v7" interface-list token into individual interface names.
Input examples::
@@ -374,7 +371,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
raw = raw.strip()
if not raw or raw == "---":
return []
ports: List[str] = []
ports: list[str] = []
for token in raw.split(","):
token = token.strip()
if not token:
@@ -402,7 +399,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
# NAPALM getters
# ------------------------------------------------------------------
def get_facts(self) -> Dict:
def get_facts(self) -> dict:
"""Return general device facts.
Combines ``show sysinfo`` and ``show version``.
@@ -467,7 +464,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
"interface_list": interface_list,
}
def _get_facts_v7(self) -> Dict:
def _get_facts_v7(self) -> dict:
"""``get_facts()`` implementation for the "v7" Cisco-like CLI.
Example ``show info`` output::
@@ -526,7 +523,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
self._port_count_v7 = int(m.group(1)) if m else 0
return self._port_count_v7
def _get_interface_list_v7(self) -> List[str]:
def _get_interface_list_v7(self) -> list[str]:
n = self._get_port_count_v7()
interfaces = [f"g{i}" for i in range(1, n + 1)]
try:
@@ -540,10 +537,10 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
interfaces.append(f"lag{m.group(1)}")
return interfaces
def _get_interface_list(self) -> List[str]:
def _get_interface_list(self) -> list[str]:
"""Return sorted list of slot/port interface names from ``show port all``."""
output = self._send_command("show port all")
interfaces: List[str] = []
interfaces: list[str] = []
for line in output.splitlines():
# Physical ports: "0/1 ..." and LAG channels: "ch1 ..."
m = re.match(r"^\s*(\d+/\d+|ch\d+)\s+", line)
@@ -554,7 +551,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
key=lambda s: [int(x) if x.isdigit() else x for x in re.split(r"(\d+)", s)],
)
def get_interfaces(self) -> Dict[str, Dict]:
def get_interfaces(self) -> dict[str, dict]:
"""Return per-interface details.
Combines ``show port all`` (link status, speed) with
@@ -583,7 +580,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
self._add_lag_info(interfaces)
return interfaces
def _add_lag_info(self, interfaces: Dict[str, Dict]) -> None:
def _add_lag_info(self, interfaces: dict[str, dict]) -> None:
"""Enrich ``interfaces`` with LAG/trunk membership from ``show port-channel all``.
FASTPATH-based switches (NETGEAR M4xxx / GSxxxTxx Smart Managed Pro)
@@ -618,7 +615,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
interfaces[lag]["lag_members"] = members
interfaces[lag]["lag_mode"] = lag_mode
def _get_interfaces_v7(self) -> Dict[str, Dict]:
def _get_interfaces_v7(self) -> dict[str, dict]:
"""``get_interfaces()`` implementation for the "v7" Cisco-like CLI.
Example ``show interfaces GigabitEthernet 1-10 status`` output::
@@ -629,7 +626,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
g2 Notconnct 1 -- -- Copper
"""
n = self._get_port_count_v7()
interfaces: Dict[str, Dict] = {}
interfaces: dict[str, dict] = {}
if n:
status_out = self._send_command(f"show interfaces GigabitEthernet 1-{n} status")
interfaces.update(self._parse_interface_status_v7(status_out))
@@ -638,8 +635,8 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return interfaces
@staticmethod
def _parse_interface_status_v7(output: str) -> Dict[str, Dict]:
interfaces: Dict[str, Dict] = {}
def _parse_interface_status_v7(output: str) -> dict[str, dict]:
interfaces: dict[str, dict] = {}
for line in output.splitlines():
line_s = line.rstrip()
if not line_s:
@@ -669,7 +666,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
}
return interfaces
def _add_lag_info_v7(self, interfaces: Dict[str, Dict]) -> None:
def _add_lag_info_v7(self, interfaces: dict[str, dict]) -> None:
"""Enrich ``interfaces`` with LAG membership from "show lag".
Example output::
@@ -711,9 +708,9 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
"lag_mode": "lacp" if "lacp" in type_s.lower() else "trunk",
}
def _parse_interfaces(self, port_output: str, intf_output: str = "") -> Dict[str, Dict]:
def _parse_interfaces(self, port_output: str, intf_output: str = "") -> dict[str, dict]:
"""Parse ``show port all`` and ``show interface all`` into NAPALM format."""
interfaces: Dict[str, Dict] = {}
interfaces: dict[str, dict] = {}
# --- parse 'show port all' ---
# Skip header lines until we hit a separator row, then parse data rows.
@@ -766,7 +763,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
}
# --- parse 'show interface all' (block format) ---
current_port: Optional[str] = None
current_port: str | None = None
for line in intf_output.splitlines():
# Block header: "Interface................................ 0/1"
m = re.match(r"^\s*Interface[.:\s]+(\d+/\d+|ch\d+)", line, re.I)
@@ -790,7 +787,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return interfaces
def get_interfaces_ip(self) -> Dict[str, Dict]:
def get_interfaces_ip(self) -> dict[str, dict]:
"""Return all configured IP addresses grouped by interface.
Parses ``show ip interface`` output. On Smart Managed switches
@@ -803,7 +800,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
Default Gateway.................................. 192.168.0.1
"""
output = self._send_command("show ip interface")
interfaces_ip: Dict[str, Dict] = {}
interfaces_ip: dict[str, dict] = {}
# Smart Managed switches expose a single management IP block;
# some firmwares add an "Interface" header for the routing VLAN.
@@ -873,7 +870,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
full: bool = False,
sanitized: bool = False,
format: str = "text",
) -> Dict[str, str]:
) -> dict[str, str]:
"""Return running and/or startup configuration.
Netgear Smart Managed does not support a candidate config;
@@ -896,7 +893,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return configs
def get_arp_table(self, vrf: str = "") -> List[Dict]:
def get_arp_table(self, vrf: str = "") -> list[dict]:
"""Return the ARP table.
Example ``show arp`` output::
@@ -949,7 +946,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return arp_table
def get_mac_address_table(self) -> List[Dict]:
def get_mac_address_table(self) -> list[dict]:
"""Return the MAC address table.
Example ``show mac-addr-table`` output::
@@ -1008,7 +1005,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return mac_table
def _get_mac_address_table_v7(self) -> List[Dict]:
def _get_mac_address_table_v7(self) -> list[dict]:
"""``get_mac_address_table()`` implementation for the "v7" Cisco-like CLI.
Example ``show mac address-table`` output::
@@ -1055,7 +1052,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return mac_table
def get_lldp_neighbors(self) -> Dict[str, List[Dict]]:
def get_lldp_neighbors(self) -> dict[str, list[dict]]:
"""Return a dict of LLDP neighbors keyed by local port.
Example ``show lldp remote-device all`` output::
@@ -1065,7 +1062,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
--------- ----- -------------------- --------------- ---------------
0/1 1 00:1a:2b:3c:4d:5e 0/1 router-1
"""
neighbors: Dict[str, List[Dict]] = {}
neighbors: dict[str, list[dict]] = {}
for row in self._get_lldp_table():
entry = {"hostname": row["system_name"], "port": row["port_id"]}
chassis_id = row.get("remote_chassis_id", "")
@@ -1074,12 +1071,12 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
neighbors.setdefault(row["local_port"], []).append(entry)
return neighbors
def _get_lldp_table(self) -> List[Dict]:
def _get_lldp_table(self) -> list[dict]:
"""Parse the LLDP neighbour table into a list of row dicts."""
if self._detect_cli_v7():
return self._get_lldp_table_v7()
output = self._send_command("show lldp remote-device all")
rows: List[Dict] = []
rows: list[dict] = []
in_table = False
for line in output.splitlines():
@@ -1111,7 +1108,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return rows
def _get_lldp_table_v7(self) -> List[Dict]:
def _get_lldp_table_v7(self) -> list[dict]:
"""Parse ``show lldp neighbor`` on v7 firmware.
The v7 CLI has neither ``show lldp remote-device all`` nor a bare
@@ -1132,7 +1129,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
except Exception: # pragma: no cover - transport failure
return []
rows: List[Dict] = []
rows: list[dict] = []
for line in output.splitlines():
if "|" not in line:
continue
@@ -1155,13 +1152,13 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
)
return rows
def get_lldp_neighbors_detail(self, interface: str = "") -> Dict[str, List[Dict]]:
def get_lldp_neighbors_detail(self, interface: str = "") -> dict[str, list[dict]]:
"""Return detailed LLDP neighbor information.
Fetches per-port detail via ``show lldp remote-device detail <port>``
and falls back to the summary table values for missing fields.
"""
details: Dict[str, List[Dict]] = {}
details: dict[str, list[dict]] = {}
for row in self._get_lldp_table():
if interface and row["local_port"] != interface:
@@ -1183,9 +1180,9 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return details
@staticmethod
def _parse_lldp_detail(output: str) -> Dict:
def _parse_lldp_detail(output: str) -> dict:
"""Parse ``show lldp remote-device detail`` block output."""
result: Dict = {
result: dict = {
"parent_interface": "",
"remote_port": "",
"remote_port_description": "",
@@ -1229,7 +1226,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return result
def get_vlans(self) -> Dict[str, Dict]:
def get_vlans(self) -> dict[str, dict]:
"""Return VLAN information.
Example ``show vlan`` output::
@@ -1243,8 +1240,8 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return self._get_vlans_v7()
output = self._send_command("show vlan")
vlans: Dict[str, Dict] = {}
current_id: Optional[str] = None
vlans: dict[str, dict] = {}
current_id: str | None = None
in_table = False
for line in output.splitlines():
@@ -1271,7 +1268,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return vlans
@staticmethod
def _parse_vlan_ports(ports_raw: str) -> List[str]:
def _parse_vlan_ports(ports_raw: str) -> list[str]:
"""Parse Netgear VLAN port string and expand ranges.
Input examples::
@@ -1279,7 +1276,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
"0/1, 0/3, 0/5" → ["0/1", "0/3", "0/5"]
"0/1-4" → ["0/1", "0/2", "0/3", "0/4"]
"""
interfaces: List[str] = []
interfaces: list[str] = []
for token in ports_raw.split(","):
token = token.strip()
if not token:
@@ -1295,7 +1292,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
interfaces.append(token)
return interfaces
def _get_vlans_v7(self) -> Dict[str, Dict]:
def _get_vlans_v7(self) -> dict[str, dict]:
"""``get_vlans()`` implementation for the "v7" Cisco-like CLI.
Example ``show vlan 1-4094`` output::
@@ -1306,7 +1303,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
10 MGMT Static --- g1-8
"""
output = self._send_paged_command("show vlan 1-4094")
vlans: Dict[str, Dict] = {}
vlans: dict[str, dict] = {}
for line in output.splitlines():
line_s = line.strip()
if not line_s or "|" not in line_s:
@@ -1341,7 +1338,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
def _get_running_config(self) -> str:
return self._send_command("show running-config")
def commit_config(self, message: str = "", revert_in: Optional[int] = None) -> None:
def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
"""Apply the staged candidate configuration to the device and persist it.
1. Saves current running config as rollback backup.
@@ -1361,7 +1358,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
self._backup_config = self._send_command("show running-config")
errors: List[str] = []
errors: list[str] = []
try:
self._enter_config_mode()
errors = self._apply_config_lines(self._candidate_config)
@@ -1401,14 +1398,14 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
self._backup_config = None
@staticmethod
def _parse_config_blocks(config_text: str) -> Dict[str, List[str]]:
def _parse_config_blocks(config_text: str) -> dict[str, list[str]]:
"""Parse a running-config into ``context → [command lines]`` mapping.
The special key ``"__global__"`` holds top-level commands.
Sub-mode blocks (``interface …``, ``vlan …``) are stored under the
opening line that introduces them.
"""
blocks: Dict[str, List[str]] = {"__global__": []}
blocks: dict[str, list[str]] = {"__global__": []}
ctx = "__global__"
for line in config_text.splitlines():
@@ -1436,7 +1433,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return blocks
@staticmethod
def _negate_command(cmd: str) -> Optional[str]:
def _negate_command(cmd: str) -> str | None:
"""Return the ``no`` form of *cmd*, or ``None`` if not known."""
for kw in (
"description",
@@ -1450,12 +1447,12 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return None
@staticmethod
def _diff_to_commands(backup: str, current: str) -> List[str]:
def _diff_to_commands(backup: str, current: str) -> list[str]:
"""Generate the minimal set of config commands to revert *current* to *backup*."""
backup_blocks = NetgearSmartDriver._parse_config_blocks(backup)
current_blocks = NetgearSmartDriver._parse_config_blocks(current)
cmds: List[str] = []
cmds: list[str] = []
all_ctxs = set(backup_blocks.keys()) | set(current_blocks.keys())
for ctx in sorted(all_ctxs):
@@ -1488,7 +1485,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
# Additional NAPALM getters
# ------------------------------------------------------------------
def get_environment(self) -> Dict:
def get_environment(self) -> dict:
"""Return device environment data (CPU, memory).
Netgear Smart Managed switches do not expose fan, temperature, or
@@ -1526,7 +1523,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
},
}
def get_interfaces_counters(self) -> Dict[str, Dict]:
def get_interfaces_counters(self) -> dict[str, dict]:
"""Return per-interface packet and byte counters.
Parses ``show interface counters`` (or ``show interface ethernet 0/1``).
@@ -1537,11 +1534,11 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
def _int(s: str) -> int:
return int(s.replace(",", "")) if s.strip().replace(",", "").isdigit() else 0
counters: Dict[str, Dict] = {}
current_port: Optional[str] = None
current: Optional[Dict] = None
counters: dict[str, dict] = {}
current_port: str | None = None
current: dict | None = None
_zero: Dict = {
_zero: dict = {
"tx_errors": 0,
"rx_errors": 0,
"tx_discards": 0,
@@ -1602,7 +1599,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return counters
def get_users(self) -> Dict[str, Dict]:
def get_users(self) -> dict[str, dict]:
"""Return local user accounts.
Example ``show users`` output::
@@ -1612,7 +1609,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
admin Read/Write 5 Enabled
"""
output = self._send_command("show users")
users: Dict[str, Dict] = {}
users: dict[str, dict] = {}
in_table = False
for line in output.splitlines():
@@ -1636,7 +1633,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return users
def get_snmp_information(self) -> Dict:
def get_snmp_information(self) -> dict:
"""Return SNMP configuration.
Parses ``show sysinfo`` for contact/location/chassis-ID and
@@ -1648,7 +1645,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
mac = self._parse_key_value(sysinfo, "Burned In MAC Address").replace("-", ":").upper()
snmp_out = self._send_command("show snmp")
communities: Dict[str, Dict] = {}
communities: dict[str, dict] = {}
in_comm = False
for line in snmp_out.splitlines():
@@ -1675,10 +1672,10 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
"chassis_id": mac,
}
def get_ntp_servers(self) -> Dict[str, Dict]:
def get_ntp_servers(self) -> dict[str, dict]:
"""Return configured NTP/SNTP servers from ``show sntp server``."""
output = self._send_command(["show sntp server", "show ntp server"])
servers: Dict[str, Dict] = {}
servers: dict[str, dict] = {}
for line in output.splitlines():
line_s = line.strip()
@@ -1699,19 +1696,19 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return servers
def get_ntp_peers(self) -> Dict[str, Dict]:
def get_ntp_peers(self) -> dict[str, dict]:
"""Return NTP peers (same as servers on Netgear Smart Managed)."""
return self.get_ntp_servers()
def get_ntp_stats(self) -> List[Dict]:
def get_ntp_stats(self) -> list[dict]:
"""Netgear Smart Managed CLI does not expose per-peer NTP statistics."""
return []
def get_optics(self) -> Dict:
def get_optics(self) -> dict:
"""SFP DDM/DOM data is not accessible via the Smart Managed CLI."""
return {}
def get_ipv6_neighbors_table(self) -> List[Dict]:
def get_ipv6_neighbors_table(self) -> list[dict]:
"""IPv6 ND table is not exposed via the Smart Managed CLI."""
return []
@@ -1720,14 +1717,14 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
destination: str = "",
protocol: str = "",
longer: bool = False,
) -> Dict[str, List[Dict]]:
) -> dict[str, list[dict]]:
"""Return routing table entries.
Smart Managed Pro switches support only a default gateway;
``show ip route`` is parsed where available.
"""
output = self._send_command("show ip route")
routes: Dict[str, List[Dict]] = {}
routes: dict[str, list[dict]] = {}
proto_map = {"c": "connected", "s": "static"}
for line in output.splitlines():
@@ -1783,7 +1780,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
count: int = 5,
vrf: str = "",
source_interface: str = "",
) -> Dict:
) -> dict:
"""Ping *destination* from the device.
Uses ``ping <dst> count <N>``. Source IP and packet size selection
@@ -1832,13 +1829,13 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
def cli(
self,
commands: List[str],
commands: list[str],
encoding: str = "text",
) -> Dict[str, Union[str, Dict[str, Any]]]:
) -> dict[str, str | dict[str, Any]]:
"""Execute a list of CLI commands and return their raw output."""
if encoding != "text":
raise NotImplementedError(f"Encoding '{encoding}' is not supported by this driver.")
result: Dict[str, Union[str, Dict[str, Any]]] = {}
result: dict[str, str | dict[str, Any]] = {}
for cmd in commands:
result[cmd] = self._send_command(cmd)
return result
@@ -1887,10 +1884,10 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
# SNMP agent (napalm-device-types SnmpAgentMixin): the caller's values
# ------------------------------------------------------------------
def _install_snmp_agent(self) -> Dict[str, Any]:
def _install_snmp_agent(self) -> dict[str, Any]:
return {"ok": True, "output": "the SNMP agent is built in"}
def _write_snmp_agent_config(self, settings: Dict[str, Any]) -> Dict[str, Any]:
def _write_snmp_agent_config(self, settings: dict[str, Any]) -> dict[str, Any]:
if self._detect_cli_v7():
# The "v7" Cisco-like CLI (e.g. GS110TPv3 7.x) has no "snmp-server"
# commands at all: SNMP is set in the web UI on this firmware.
@@ -1919,16 +1916,16 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return {"ok": True, "output": "community (read-only), contact and location set"}
@staticmethod
def _allow_snmp_sources(settings: Dict[str, Any]) -> Dict[str, Any]:
def _allow_snmp_sources(settings: dict[str, Any]) -> dict[str, Any]:
sources = ", ".join(settings.get("allowed_sources", [])) or "none given"
note = f"no host firewall on the switch; sources not enforced ({sources})"
return {"ok": True, "output": note}
def _start_snmp_agent(self) -> Dict[str, Any]:
def _start_snmp_agent(self) -> dict[str, Any]:
self._save_config()
return {"ok": True, "output": "configuration saved"}
def configure_snmp_agent(self, settings: Dict[str, Any]) -> Dict[str, Any]:
def configure_snmp_agent(self, settings: dict[str, Any]) -> dict[str, Any]:
"""The template's steps, then a check that the switch lists the community."""
result = SnmpAgentMixin.configure_snmp_agent(self, settings)
if not result["success"]:
@@ -1943,7 +1940,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
# PoE
# ------------------------------------------------------------------
def get_poe_status(self) -> Dict[str, Dict]:
def get_poe_status(self) -> dict[str, dict]:
"""Return PoE configuration per port from ``show power inline``.
Example output::
@@ -1967,7 +1964,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
elif "class" in mode:
allocation_method = "PPAM_CLASS"
result: Dict[str, Dict] = {}
result: dict[str, dict] = {}
in_table = False
for line in output.splitlines():
line_s = line.strip()
@@ -1997,7 +1994,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return result
def set_poe(self, interface: str, config: Dict) -> None:
def set_poe(self, interface: str, config: dict) -> None:
"""Update PoE configuration for a single port.
Uses the ``power inline`` interface sub-commands (ProSafe CLI).
@@ -2032,7 +2029,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
# LAG / trunk membership
# ------------------------------------------------------------------
def set_lag_members(self, lag_name: str, members: List[str]) -> None:
def set_lag_members(self, lag_name: str, members: list[str]) -> None:
"""Set the full member-port list of a LAG/trunk group (e.g. ``"ch1"``).
Diffs ``members`` against the LAG's current members (as reported by
@@ -2052,7 +2049,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
if not to_remove and not to_add:
return
lines: List[str] = []
lines: list[str] = []
for port in to_remove:
lines.append(f"interface {port}")
lines.append(f"deleteport {lag_name}")
pyproject.toml
+8 -1
View File
@@ -54,4 +54,11 @@ include = ["napalm_netgear*"]
[tool.ruff]
line-length = 100
target-version = "py38"
target-version = "py310"
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B"]
[tool.ruff.lint.per-file-ignores]
# Recorded device output keeps its lines.
"tests/**" = ["E501"]
tests/unit/test_driver.py
+2 -2
View File
@@ -1,10 +1,10 @@
"""Unit tests for NetgearSmartDriver — no real device required."""
import pytest
from unittest.mock import MagicMock, patch
from napalm_netgear.netgear_smart import NetgearSmartDriver
import pytest
from napalm_netgear.netgear_smart import NetgearSmartDriver
# ---------------------------------------------------------------------------
# Fixtures
tests/unit/test_ssh_host_key.py
+3 -1
View File
@@ -24,7 +24,9 @@ HANDLER = "napalm_netgear.netgear_smart.ConnectHandler"
def _device() -> MagicMock:
device = MagicMock()
device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = SERVER_KEY
device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = (
SERVER_KEY
)
return device