From fa10d6617b05ea1601c380b34e879af4f4fd0ba0 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Fri, 23 Oct 2026 08:32:43 +0200 Subject: [PATCH] style: ruff format and the lint rules of the driver standard Mechanical: ruff format, and ruff check --fix with the rule set every NAPALM driver uses (E, F, I, UP, B at line length 100, Python 3.10). What ruff cannot fix comes in the next commit. --- napalm_netgear/__init__.py | 2 +- napalm_netgear/netgear_plus.py | 105 +++++++---------- napalm_netgear/netgear_smart.py | 201 ++++++++++++++++---------------- pyproject.toml | 9 +- tests/unit/test_driver.py | 4 +- tests/unit/test_ssh_host_key.py | 4 +- 6 files changed, 156 insertions(+), 169 deletions(-) diff --git a/napalm_netgear/__init__.py b/napalm_netgear/__init__.py index 4d05736..6e58df5 100644 --- a/napalm_netgear/__init__.py +++ b/napalm_netgear/__init__.py @@ -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"] diff --git a/napalm_netgear/netgear_plus.py b/napalm_netgear/netgear_plus.py index fa5a4e0..989d078 100644 --- a/napalm_netgear/netgear_plus.py +++ b/napalm_netgear/netgear_plus.py @@ -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 @@ -41,9 +40,9 @@ _FW_CDN_BASE = "https://www.downloads.netgear.com/files/GDC" # Verified via HEAD requests against downloads.netgear.com on 2026-05-24. # Update this table when Netgear releases new firmware. _FW_KNOWN_LATEST: dict[str, str] = { - "GS305EP": "V2.0.0.10", + "GS305EP": "V2.0.0.10", "GS305EPP": "V2.0.0.10", - "GS308EP": "V2.0.0.11", + "GS308EP": "V2.0.0.11", "GS308EPP": "V2.0.0.11", } @@ -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: ...24Guest vlan_names: dict[str, str] = {} vlan_entry_re = re.compile( - r']*>.*?(\d+)\s*\s*' - r']*>([^<]*)', + r"]*>.*?(\d+)\s*\s*" + r"]*>([^<]*)", 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 1*, 24, 25, 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']*pvid-str[^>]*>([^<]+)') + pvid_span_re = re.compile(r"]*pvid-str[^>]*>([^<]+)") 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: diff --git a/napalm_netgear/netgear_smart.py b/napalm_netgear/netgear_smart.py index 369c447..5b2d748 100644 --- a/napalm_netgear/netgear_smart.py +++ b/napalm_netgear/netgear_smart.py @@ -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 `` 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 count ``. 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}") diff --git a/pyproject.toml b/pyproject.toml index f43267f..9aee7b2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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"] diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index d6211ab..5dc8da5 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -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 diff --git a/tests/unit/test_ssh_host_key.py b/tests/unit/test_ssh_host_key.py index 061c037..648ef87 100644 --- a/tests/unit/test_ssh_host_key.py +++ b/tests/unit/test_ssh_host_key.py @@ -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