diff --git a/napalm_procurve/api_client.py b/napalm_procurve/api_client.py index 03ae7cc..988badf 100644 --- a/napalm_procurve/api_client.py +++ b/napalm_procurve/api_client.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 """REST API client for HPE/Aruba ProCurve switches (ArubaOS REST interface). @@ -8,13 +7,12 @@ Used automatically by ProcurveDriver when the REST API is reachable. """ import logging -from typing import Any, Dict, Optional, Tuple +from typing import Any import requests import urllib3 - from napalm.base import helpers as napalm_helpers -from napalm.base.exceptions import ConnectionException, ConnectAuthError +from napalm.base.exceptions import ConnectAuthError, ConnectionException from napalm_device_types import add_lag_interfaces logger = logging.getLogger("napalm_procurve.api") @@ -37,7 +35,7 @@ class ProcurveApiClient: password: str, timeout: int = 60, ssl_verify: bool = False, - api_version: Optional[str] = None, + api_version: str | None = None, ) -> None: self.hostname = hostname self.username = username @@ -47,7 +45,7 @@ class ProcurveApiClient: self._api_version = api_version # None = auto-detect self._proto: str = "https" self._base_url: str = "" - self._session: Optional[requests.Session] = None + self._session: requests.Session | None = None # ------------------------------------------------------------------ # Detection @@ -59,7 +57,7 @@ class ProcurveApiClient: hostname: str, timeout: int = 5, ssl_verify: bool = False, - ) -> Tuple[Optional[str], Optional[str]]: + ) -> tuple[str | None, str | None]: """Probe the device for an ArubaOS REST API. Returns ``(api_version, proto)`` on success, ``(None, None)`` if the @@ -86,7 +84,10 @@ class ProcurveApiClient: if resp.status_code in (200, 301, 302, 303, 401, 403, 405): logger.debug( "API detected at %s (HTTP %s, %s %s)", - hostname, resp.status_code, proto, ver, + hostname, + resp.status_code, + proto, + ver, ) return ver, proto except requests.exceptions.SSLError: @@ -115,9 +116,7 @@ class ProcurveApiClient: self._session = requests.Session() self._session.verify = self.ssl_verify - self._session.headers.update( - {"Content-Type": "application/json", "Connection": "close"} - ) + self._session.headers.update({"Content-Type": "application/json", "Connection": "close"}) url = self._base_url + "login-sessions" resp = self._session.post( @@ -148,9 +147,7 @@ class ProcurveApiClient: """Log out from the REST API session.""" if self._session and self._base_url: try: - self._session.delete( - self._base_url + "login-sessions", timeout=self.timeout - ) + self._session.delete(self._base_url + "login-sessions", timeout=self.timeout) except Exception: pass if self._session: @@ -162,9 +159,7 @@ class ProcurveApiClient: if not self._session: return False try: - resp = self._session.get( - self._base_url + "system/status", timeout=5 - ) + resp = self._session.get(self._base_url + "system/status", timeout=5) return resp.status_code == 200 except Exception: return False @@ -173,7 +168,7 @@ class ProcurveApiClient: # HTTP helpers # ------------------------------------------------------------------ - def get(self, endpoint: str) -> Dict[str, Any]: + def get(self, endpoint: str) -> dict[str, Any]: """GET ``{base_url}{endpoint}`` and return parsed JSON (or empty dict).""" url = self._base_url + endpoint try: @@ -183,7 +178,9 @@ class ProcurveApiClient: if resp.status_code == 404: logger.debug("GET %s returned HTTP 404", url) else: - logger.warning("GET %s returned HTTP %s: %s", url, resp.status_code, resp.text[:200]) + logger.warning( + "GET %s returned HTTP %s: %s", url, resp.status_code, resp.text[:200] + ) except Exception as exc: logger.warning("GET %s error: %s", url, exc) return {} @@ -204,7 +201,7 @@ class ProcurveApiClient: # NAPALM data getters (API-backed) # ------------------------------------------------------------------ - def get_facts(self) -> Dict: + def get_facts(self) -> dict: """Return NAPALM facts from the REST API.""" system = self.get("system/status") if not system: @@ -230,6 +227,7 @@ class ProcurveApiClient: model_raw = system.get("product_model", "") # Extract J-code part numbers (e.g. J9777A) from the model string import re as _re + _PN_RE = _re.compile(r"\b(J\d{4}[A-Z]{1,2})\b") pn_match = _PN_RE.search(model_raw) part_number = pn_match.group(1) if pn_match else "" @@ -253,14 +251,14 @@ class ProcurveApiClient: "interface_list": iface_list, } - def get_interfaces(self) -> Dict[str, Dict]: + def get_interfaces(self) -> dict[str, dict]: """Return NAPALM interfaces from the REST API.""" ports_data = self.get("ports") stats_data = self.get("port-statistics") - output: Dict[str, Dict] = {} - trunk_groups: Dict[str, list] = {} - trunk_modes: Dict[str, str] = {} + output: dict[str, dict] = {} + trunk_groups: dict[str, list] = {} + trunk_modes: dict[str, str] = {} port_elements = list(ports_data.get("port_element", [])) @@ -274,13 +272,15 @@ class ProcurveApiClient: for p in blade.get("data_ports", []): pid = p.get("port_name", "") if pid and pid not in seen_ids: - port_elements.append({ - "id": pid, - "name": "", - "is_port_up": p.get("operStatus") == "OPER_UP", - "is_port_enabled": p.get("adminStatus") == "ADMIN_UP", - "trunk_group": "", - }) + port_elements.append( + { + "id": pid, + "name": "", + "is_port_up": p.get("operStatus") == "OPER_UP", + "is_port_enabled": p.get("adminStatus") == "ADMIN_UP", + "trunk_group": "", + } + ) seen_ids.add(pid) except Exception: pass @@ -309,15 +309,18 @@ class ProcurveApiClient: output[pid]["speed"] = float(stat.get("port_speed_mbps", 0)) # One row per LAG/trunk group alongside its member ports. - return add_lag_interfaces(output, { - group: "lacp" if trunk_modes.get(group) == "PTT_LACP" else "trunk" - for group in trunk_groups - }) + return add_lag_interfaces( + output, + { + group: "lacp" if trunk_modes.get(group) == "PTT_LACP" else "trunk" + for group in trunk_groups + }, + ) - def get_interfaces_ip(self) -> Dict[str, Dict]: + def get_interfaces_ip(self) -> dict[str, dict]: """Return NAPALM interfaces IP from the REST API.""" vlans_data = self.get("vlans") - result: Dict[str, Dict] = {} + result: dict[str, dict] = {} for vlan in vlans_data.get("vlan_element", []): vlan_id = str(vlan.get("vlan_id", "")) @@ -331,6 +334,7 @@ class ProcurveApiClient: continue try: import netaddr + plen = netaddr.IPNetwork(f"{addr}/{mask}").prefixlen except Exception: plen = 24 @@ -343,12 +347,12 @@ class ProcurveApiClient: # PoE # ------------------------------------------------------------------ - def get_poe_ports(self) -> Dict[str, Dict]: + def get_poe_ports(self) -> dict[str, dict]: """Return PoE configuration for all ports, keyed by port id.""" data = self.get("poe/ports") return {p["port_id"]: p for p in data.get("port_poe", []) if p.get("port_id")} - def set_port_poe(self, port_id: str, payload: Dict[str, Any]) -> requests.Response: + def set_port_poe(self, port_id: str, payload: dict[str, Any]) -> requests.Response: """PUT PoE configuration changes for a single port.""" return self.put(f"ports/{port_id}/poe", json=payload) @@ -390,9 +394,9 @@ class ProcurveApiClient: ) return table - def get_lldp_neighbors(self) -> Dict[str, list]: + def get_lldp_neighbors(self) -> dict[str, list]: """Return NAPALM LLDP neighbors from the REST API.""" - result: Dict[str, list] = {} + result: dict[str, list] = {} data = self.get("lldp/remote-device") for nbr in data.get("lldp_remote_device_element", []): local_port = nbr.get("local_port", "") @@ -404,9 +408,9 @@ class ProcurveApiClient: ) return result - def get_lldp_neighbors_detail(self) -> Dict[str, list]: + def get_lldp_neighbors_detail(self) -> dict[str, list]: """Return NAPALM LLDP neighbor details from the REST API.""" - result: Dict[str, list] = {} + result: dict[str, list] = {} data = self.get("lldp/remote-device") for nbr in data.get("lldp_remote_device_element", []): local_port = nbr.get("local_port", "") @@ -424,7 +428,7 @@ class ProcurveApiClient: ) return result - def get_config(self) -> Dict[str, str]: + def get_config(self) -> dict[str, str]: """Return running and startup configuration via REST API. AOS-Switch returns the config as plain text, not JSON. @@ -447,17 +451,17 @@ class ProcurveApiClient: logger.warning("GET %s error: %s", url, exc) return "" - def get_ntp_servers(self) -> Dict[str, Dict]: + def get_ntp_servers(self) -> dict[str, dict]: """Return NTP servers from REST API.""" data = self.get("ntp/server") - servers: Dict[str, Dict] = {} + servers: dict[str, dict] = {} for entry in data.get("ntp_server_element", []): addr = entry.get("server_address", "") if addr: servers[addr] = {} return servers - def get_vlans(self) -> Dict[int, Dict]: + def get_vlans(self) -> dict[int, dict]: """Return VLAN information with proper tagged/untagged separation. Fetches ``/vlans`` for VLAN names and ``/vlans-ports`` for port membership. @@ -491,7 +495,7 @@ class ProcurveApiClient: ports_data = self.get("vlans-ports") # Build VLAN skeleton from /vlans - result: Dict[int, Dict] = {} + result: dict[int, dict] = {} for vlan in vlans_data.get("vlan_element", []): try: vid = int(vlan["vlan_id"]) @@ -522,8 +526,7 @@ class ProcurveApiClient: # Untagged variants: POM_UNTAGGED, POM_NATIVE_UNTAGGED, "Untagged" is_untagged = any( - kw in mode_low - for kw in ("untagged", "native_untagged", "pom_untagged") + kw in mode_low for kw in ("untagged", "native_untagged", "pom_untagged") ) and "tagged" not in mode_low.replace("untagged", "") # More precise: contains "untagged" anywhere → untagged @@ -576,17 +579,17 @@ class ProcurveApiClient: f"delete_vlan({vlan_id}): REST API returned HTTP {resp.status_code}" ) - def get_snmp_communities(self) -> Dict[str, Dict]: + def get_snmp_communities(self) -> dict[str, dict]: """Return configured SNMP communities from REST API.""" data = self.get("snmp-server/community") - result: Dict[str, Dict] = {} + result: dict[str, dict] = {} for entry in data.get("snmp_community_element", []): name = entry.get("community_name", "") if name: result[name] = {"access_type": entry.get("access_type", "")} return result - def configure_snmp_community(self, community: str = "public") -> Tuple[bool, str]: + def configure_snmp_community(self, community: str = "public") -> tuple[bool, str]: """Ensure an SNMP read-only community exists via REST resource endpoint. Returns (success, message). @@ -595,10 +598,13 @@ class ProcurveApiClient: if community in existing: return True, f"SNMP community '{community}' already exists." - resp = self.post("snmp-server/community", json={ - "community_name": community, - "access_type": "MIB2", - }) + resp = self.post( + "snmp-server/community", + json={ + "community_name": community, + "access_type": "MIB2", + }, + ) if resp.ok: return True, f"SNMP community '{community}' configured via REST API." if resp.status_code == 409: @@ -615,9 +621,9 @@ class ProcurveApiClient: size: int = 100, count: int = 5, vrf: str = "", - ) -> Dict: + ) -> dict: """Execute ping via REST API (if supported) or return not-implemented.""" - payload: Dict = {"destination": destination, "count": count, "timeout": timeout} + payload: dict = {"destination": destination, "count": count, "timeout": timeout} if source: payload["source"] = source diff --git a/napalm_procurve/parsers.py b/napalm_procurve/parsers.py index c520294..119f9fc 100644 --- a/napalm_procurve/parsers.py +++ b/napalm_procurve/parsers.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 """CLI output parsers for HPE/Aruba ProCurve switches. @@ -9,13 +8,12 @@ backends. """ import re -from typing import Dict, List, Optional - # --------------------------------------------------------------------------- # MAC address helpers # --------------------------------------------------------------------------- + def procurve_mac_to_std(mac: str) -> str: """Convert ProCurve MAC format ``a1b2c3-d4e5f6`` to ``a1:b2:c3:d4:e5:f6``.""" clean = mac.replace("-", "").replace(":", "").replace(".", "").lower() @@ -38,7 +36,8 @@ def _kv(output: str, key: str) -> str: # show system-information # --------------------------------------------------------------------------- -def parse_system_info(output: str) -> Dict: + +def parse_system_info(output: str) -> dict: """Parse ``show system-information`` output. Returns a dict with keys: hostname, os_version, serial_number, model, @@ -63,10 +62,8 @@ def parse_system_info(output: str) -> Dict: # between them, e.g.: # " Software revision : R.11.27 Base MAC Addr : a1b2c3-d4e5f6" # We capture all k:v pairs on every line. - _KV_RE = re.compile( - r"([A-Za-z][A-Za-z0-9 ()/#-]{1,40}?)\s*:\s*(\S.*?)(?=\s{3,}[A-Za-z]|$)" - ) - kv_map: Dict[str, str] = {} + _KV_RE = re.compile(r"([A-Za-z][A-Za-z0-9 ()/#-]{1,40}?)\s*:\s*(\S.*?)(?=\s{3,}[A-Za-z]|$)") + kv_map: dict[str, str] = {} for raw_line in output.splitlines(): for m in _KV_RE.finditer(raw_line): k = m.group(1).strip() @@ -163,6 +160,7 @@ def _parse_procurve_uptime(uptime_str: str) -> float: # show version # --------------------------------------------------------------------------- + def parse_version(output: str) -> str: """Extract firmware version from ``show version`` or ``show system-information``. @@ -234,24 +232,24 @@ def parse_model_from_version(output: str) -> tuple[str, str]: # 1 100/1000T | Yes Down Auto Unknown off 0 _INTF_BRIEF_RE = re.compile( - r"^\s*([^\s-]+)" # Port - r"(?:-(Trk\d+))?\s+" # Trunk group of a member port ("3-Trk3") - r"(\S+)\s+\|" # Type | + r"^\s*([^\s-]+)" # Port + r"(?:-(Trk\d+))?\s+" # Trunk group of a member port ("3-Trk3") + r"(\S+)\s+\|" # Type | r"\s+(?:(?:Yes|No)\s+)?" # Intrusion Alert, on firmware that has the column - r"(Yes|No)\s+" # Enabled - r"(Up|Down)\s+" # Link - r"(\S+)\s+(\S+)", # Mode and MDI, in either order depending on firmware + r"(Yes|No)\s+" # Enabled + r"(Up|Down)\s+" # Link + r"(\S+)\s+(\S+)", # Mode and MDI, in either order depending on firmware re.IGNORECASE, ) -def parse_interfaces_brief(output: str) -> Dict[str, Dict]: +def parse_interfaces_brief(output: str) -> dict[str, dict]: """Parse ``show interfaces brief`` tabular output. Returns dict keyed by port name with sub-keys: is_up, is_enabled, speed, duplex, description. """ - interfaces: Dict[str, Dict] = {} + interfaces: dict[str, dict] = {} for line in output.splitlines(): m = _INTF_BRIEF_RE.match(line) if not m: @@ -260,9 +258,7 @@ def parse_interfaces_brief(output: str) -> Dict[str, Dict]: speed = 0.0 duplex = "" # Mode examples: "1000FDx", "100HDx", "Unknown"; MDI is "Auto", "MDI", "MDIX", "NA" - sm = next( - filter(None, (re.match(r"(\d+)(FDx|HDx)?", t, re.I) for t in mode_or_mdi)), None - ) + sm = next(filter(None, (re.match(r"(\d+)(FDx|HDx)?", t, re.I) for t in mode_or_mdi)), None) if sm: speed = float(sm.group(1)) duplex = "full" if (sm.group(2) or "").lower() == "fdx" else "half" @@ -286,7 +282,8 @@ def parse_interfaces_brief(output: str) -> Dict[str, Dict]: # show interfaces (per-port detail for MAC / description) # --------------------------------------------------------------------------- -def parse_interface_detail(output: str, port: str) -> Dict: + +def parse_interface_detail(output: str, port: str) -> dict: """Parse ``show interfaces `` for MAC address and description.""" mac = procurve_mac_to_std(_kv(output, "MAC address")) name = _kv(output, "Name") or _kv(output, "Description") @@ -297,7 +294,8 @@ def parse_interface_detail(output: str, port: str) -> Dict: # show ip / show ip interface # --------------------------------------------------------------------------- -def parse_ip_interfaces(output: str) -> Dict[str, Dict]: + +def parse_ip_interfaces(output: str) -> dict[str, dict]: """Parse ``show ip`` or ``show ip interface brief`` output. Returns NAPALM-compatible dict:: @@ -314,7 +312,7 @@ def parse_ip_interfaces(output: str) -> Dict[str, Dict]: IP Configuration for VLAN 99: ... """ - result: Dict[str, Dict] = {} + result: dict[str, dict] = {} current_vlan = "" current_ip = "" @@ -343,6 +341,7 @@ def parse_ip_interfaces(output: str) -> Dict[str, Dict]: mask = line.split(":", 1)[-1].strip() try: import netaddr + prefix_len = netaddr.IPNetwork(f"{current_ip}/{mask}").prefixlen except Exception: prefix_len = 24 @@ -362,23 +361,21 @@ def parse_ip_interfaces(output: str) -> Dict[str, Dict]: _ARP_RE = re.compile( r"(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})" # IP r"\s+" - r"([0-9a-fA-F]{6}-[0-9a-fA-F]{6}|" # MAC ProCurve hex-dash format - r"[0-9a-fA-F:]{17})" # or standard colon format - r"\s+(\S+)" # Type - r"(?:\s+(\S+))?", # Port (optional) + r"([0-9a-fA-F]{6}-[0-9a-fA-F]{6}|" # MAC ProCurve hex-dash format + r"[0-9a-fA-F:]{17})" # or standard colon format + r"\s+(\S+)" # Type + r"(?:\s+(\S+))?", # Port (optional) ) -def parse_arp_table(output: str) -> List[Dict]: +def parse_arp_table(output: str) -> list[dict]: """Parse ``show arp`` output into NAPALM ARP table format.""" table = [] for line in output.splitlines(): m = _ARP_RE.search(line) if not m: continue - ip, mac, entry_type, port = ( - m.group(1), m.group(2), m.group(3), m.group(4) or "" - ) + ip, mac, entry_type, port = (m.group(1), m.group(2), m.group(3), m.group(4) or "") table.append( { "interface": port, @@ -396,13 +393,13 @@ def parse_arp_table(output: str) -> List[Dict]: _MAC_RE = re.compile( r"([0-9a-fA-F]{6}-[0-9a-fA-F]{6}|[0-9a-fA-F:]{17})" # MAC - r"\s+(\S+)" # Port - r"\s+(\d+)" # VLAN - r"\s+(\w+)", # Type + r"\s+(\S+)" # Port + r"\s+(\d+)" # VLAN + r"\s+(\w+)", # Type ) -def parse_mac_table(output: str) -> List[Dict]: +def parse_mac_table(output: str) -> list[dict]: """Parse ``show mac-address`` into NAPALM MAC table format.""" table = [] for line in output.splitlines(): @@ -428,7 +425,8 @@ def parse_mac_table(output: str) -> List[Dict]: # show lldp info remote-device # --------------------------------------------------------------------------- -def parse_lldp_neighbors(output: str) -> Dict[str, List[Dict]]: + +def parse_lldp_neighbors(output: str) -> dict[str, list[dict]]: """Parse ``show lldp info remote-device`` into NAPALM neighbors dict. Output format:: @@ -437,7 +435,7 @@ def parse_lldp_neighbors(output: str) -> Dict[str, List[Dict]]: --------- + ------------------- ------------ ------------ ... 1 | a1 b2 c3 d4 e5 f6 Gi 0/1 otherswitch ... """ - neighbors: Dict[str, List[Dict]] = {} + neighbors: dict[str, list[dict]] = {} in_table = False for line in output.splitlines(): @@ -472,21 +470,19 @@ def parse_lldp_neighbors(output: str) -> Dict[str, List[Dict]]: if re.match(r"^[0-9a-fA-F]{12}$", chassis): chassis = ":".join(chassis[i : i + 2] for i in range(0, 12, 2)) - neighbors.setdefault(local_port, []).append( - {"hostname": sys_name, "port": port_id} - ) + neighbors.setdefault(local_port, []).append({"hostname": sys_name, "port": port_id}) return neighbors -def parse_lldp_neighbors_detail(output: str) -> Dict[str, List[Dict]]: +def parse_lldp_neighbors_detail(output: str) -> dict[str, list[dict]]: """Parse ``show lldp info remote-device detail`` output. Returns NAPALM lldp_neighbors_detail format. """ - neighbors: Dict[str, List[Dict]] = {} + neighbors: dict[str, list[dict]] = {} current_port = "" - entry: Dict = {} + entry: dict = {} def _save(): if current_port and entry: @@ -516,9 +512,11 @@ def parse_lldp_neighbors_detail(output: str) -> Dict[str, List[Dict]]: low = line.lower() if "chassis id" in low: val = line.split(":", 1)[-1].strip() - entry["remote_chassis_id"] = procurve_mac_to_std( - re.sub(r"\s+", "", val) - ) if re.search(r"[0-9a-fA-F ]{14,}", val) else val + entry["remote_chassis_id"] = ( + procurve_mac_to_std(re.sub(r"\s+", "", val)) + if re.search(r"[0-9a-fA-F ]{14,}", val) + else val + ) elif "port id" in low and "descr" not in low: entry["remote_port"] = line.split(":", 1)[-1].strip() elif "port description" in low: @@ -528,9 +526,7 @@ def parse_lldp_neighbors_detail(output: str) -> Dict[str, List[Dict]]: elif "system description" in low: entry["remote_system_description"] = line.split(":", 1)[-1].strip() elif "system capabilities" in low and "enabled" not in low: - entry["remote_system_capab"] = _parse_capabilities( - line.split(":", 1)[-1].strip() - ) + entry["remote_system_capab"] = _parse_capabilities(line.split(":", 1)[-1].strip()) elif "enabled capabilities" in low or ("system capabilities" in low and "enabled" in low): entry["remote_system_enable_capab"] = _parse_capabilities( line.split(":", 1)[-1].strip() @@ -540,7 +536,7 @@ def parse_lldp_neighbors_detail(output: str) -> Dict[str, List[Dict]]: return neighbors -def _parse_capabilities(cap_str: str) -> List[str]: +def _parse_capabilities(cap_str: str) -> list[str]: """Convert ProCurve capability abbreviations to NAPALM strings.""" _MAP = { "B": "bridge", @@ -566,12 +562,13 @@ def _parse_capabilities(cap_str: str) -> List[str]: # show ntp associations / show timesync # --------------------------------------------------------------------------- -def parse_ntp_servers(output: str) -> Dict[str, Dict]: + +def parse_ntp_servers(output: str) -> dict[str, dict]: """Parse ``show ntp associations`` or ``show ntp status`` output. Returns dict of {server_ip: {}} as per NAPALM get_ntp_servers. """ - servers: Dict[str, Dict] = {} + servers: dict[str, dict] = {} for line in output.splitlines(): # " 10.0.0.1 sys.peer 2 ..." or "NTP server: 10.0.0.1" m = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})", line) @@ -588,9 +585,10 @@ def parse_ntp_servers(output: str) -> Dict[str, Dict]: # show environment # --------------------------------------------------------------------------- -def parse_environment(output: str) -> Dict: + +def parse_environment(output: str) -> dict: """Parse ``show environment`` output into NAPALM environment dict.""" - env: Dict = { + env: dict = { "fans": {}, "temperature": {}, "power": {}, @@ -629,9 +627,7 @@ def parse_environment(output: str) -> Dict: } # Fan status - for m in re.finditer( - r"Fan\s*(\d+)\s*[:\s]+(\S+)", output, re.I - ): + for m in re.finditer(r"Fan\s*(\d+)\s*[:\s]+(\S+)", output, re.I): fan_id = f"fan_{m.group(1)}" status = m.group(2).lower() env["fans"][fan_id] = {"status": status in ("ok", "good", "normal", "operating")} @@ -643,12 +639,13 @@ def parse_environment(output: str) -> Dict: # show local-users # --------------------------------------------------------------------------- -def parse_users(output: str) -> Dict[str, Dict]: + +def parse_users(output: str) -> dict[str, dict]: """Parse ``show local-users`` or ``show users`` output. Returns NAPALM users dict. """ - users: Dict[str, Dict] = {} + users: dict[str, dict] = {} for line in output.splitlines(): # " manager Operator Enabled ..." or " manager Manager ..." m = re.match( @@ -678,9 +675,10 @@ def parse_users(output: str) -> Dict[str, Dict]: # show snmp-server / show snmp # --------------------------------------------------------------------------- -def parse_snmp_info(output: str) -> Dict: + +def parse_snmp_info(output: str) -> dict: """Parse ``show snmp-server`` output into NAPALM snmp_information dict.""" - communities: Dict[str, Dict] = {} + communities: dict[str, dict] = {} contact = _kv(output, "System Contact") or _kv(output, "Contact") location = _kv(output, "System Location") or _kv(output, "Location") chassis_id = "" @@ -708,7 +706,8 @@ def parse_snmp_info(output: str) -> Dict: # VLAN helpers # --------------------------------------------------------------------------- -def expand_procurve_ports(port_str: str) -> List[str]: + +def expand_procurve_ports(port_str: str) -> list[str]: """Expand a ProCurve port-list string into a flat list of port names. Handles: @@ -717,7 +716,7 @@ def expand_procurve_ports(port_str: str) -> List[str]: - Trunk ports: ``Trk1,Trk2`` → ``["Trk1","Trk2"]`` - Mixed: ``1-3,Trk1,A1`` → ``["1","2","3","Trk1","A1"]`` """ - ports: List[str] = [] + ports: list[str] = [] for token in port_str.split(","): token = token.strip() if not token: @@ -746,7 +745,7 @@ def expand_procurve_ports(port_str: str) -> List[str]: return ports -def parse_vlans_from_running_config(config_text: str) -> Dict[int, Dict]: +def parse_vlans_from_running_config(config_text: str) -> dict[int, dict]: """Parse VLAN blocks from ProCurve ``show running-config`` output. Extracts ``name``, ``tagged``, and ``untagged`` port lists per VLAN. @@ -777,8 +776,8 @@ def parse_vlans_from_running_config(config_text: str) -> Dict[int, Dict]: "untagged": []}, } """ - vlans: Dict[int, Dict] = {} - current_id: Optional[int] = None + vlans: dict[int, dict] = {} + current_id: int | None = None for line in config_text.splitlines(): stripped = line.strip() @@ -818,7 +817,7 @@ def parse_vlans_from_running_config(config_text: str) -> Dict[int, Dict]: # Populate "interfaces" as union of tagged + untagged (NAPALM standard key) for vdata in vlans.values(): - seen: List[str] = [] + seen: list[str] = [] for p in vdata["untagged"] + vdata["tagged"]: if p not in seen: seen.append(p) @@ -827,7 +826,7 @@ def parse_vlans_from_running_config(config_text: str) -> Dict[int, Dict]: return vlans -def parse_vlans_brief(output: str) -> Dict[int, str]: +def parse_vlans_brief(output: str) -> dict[int, str]: """Parse ``show vlans`` brief table for VLAN IDs and names only. Used as a discovery step when running-config is unavailable. @@ -841,7 +840,7 @@ def parse_vlans_brief(output: str) -> Dict[int, str]: Returns ``{vlan_id: name}`` dict. """ - vlans: Dict[int, str] = {} + vlans: dict[int, str] = {} in_table = False for line in output.splitlines(): @@ -857,7 +856,7 @@ def parse_vlans_brief(output: str) -> Dict[int, str]: return vlans -def parse_vlan_ports_detail(output: str) -> Dict[str, List[str]]: +def parse_vlan_ports_detail(output: str) -> dict[str, list[str]]: """Parse ``show vlans `` per-VLAN port detail output. Returns ``{"tagged": [...], "untagged": [...]}`` for ONE vlan. @@ -870,8 +869,8 @@ def parse_vlan_ports_detail(output: str) -> Dict[str, List[str]]: 2 Tagged Learn Up Trk1 Tagged Learn Up """ - tagged: List[str] = [] - untagged: List[str] = [] + tagged: list[str] = [] + untagged: list[str] = [] in_table = False for line in output.splitlines(): @@ -892,7 +891,7 @@ def parse_vlan_ports_detail(output: str) -> Dict[str, List[str]]: return {"tagged": tagged, "untagged": untagged} -def derive_port_pvids(vlans: Dict[int, Dict]) -> Dict[str, int]: +def derive_port_pvids(vlans: dict[int, dict]) -> dict[str, int]: """Derive PVID (native/access VLAN) for each port from the ``get_vlans`` result. On ProCurve a port is untagged in exactly ONE VLAN — that is its PVID. @@ -910,7 +909,7 @@ def derive_port_pvids(vlans: Dict[int, Dict]) -> Dict[str, int]: }) # → {"1": 1, "2": 1, "3": 1, "5": 1} """ - pvids: Dict[str, int] = {} + pvids: dict[str, int] = {} # First pass: assign PVID from untagged membership for vid, vdata in sorted(vlans.items()): diff --git a/napalm_procurve/procurve.py b/napalm_procurve/procurve.py index ee1d49d..9ed2442 100644 --- a/napalm_procurve/procurve.py +++ b/napalm_procurve/procurve.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 """NAPALM driver for HPE/Aruba ProCurve switches. @@ -18,15 +17,16 @@ Netmiko device type: ``hp_procurve`` / ``hp_procurve_telnet`` import logging import re -import socket -from typing import Any, ClassVar, Dict, List, Optional, Union +from typing import Any, ClassVar -import netaddr -from netmiko import ConnectHandler -from netmiko.exceptions import ( - NetmikoAuthenticationException, - NetmikoTimeoutException, +from napalm.base.exceptions import ( + CommandErrorException, + ConnectionClosedException, + ConnectionException, + MergeConfigException, + ReplaceConfigException, ) +from napalm.base.netmiko_helpers import netmiko_args from napalm_device_types import ( ConfigLifecycleMixin, FingerprintRule, @@ -35,24 +35,20 @@ from napalm_device_types import ( run_fix_steps, ) from napalm_device_types.models import InterfaceConfigDict, VlanConfigDict -from napalm.base import helpers as napalm_helpers -from napalm.base.exceptions import ( - CommandErrorException, - ConnectionException, - ConnectionClosedException, - MergeConfigException, - ReplaceConfigException, -) -from napalm.base.netmiko_helpers import netmiko_args from napalm_device_types.ssh_host_keys import ( KNOWN_KEYS_ARG, OBSERVER_ARG, HostKeyCheck, SshHostKeyMismatch, ) +from netmiko import ConnectHandler +from netmiko.exceptions import ( + NetmikoAuthenticationException, + NetmikoTimeoutException, +) -from napalm_procurve.api_client import ProcurveApiClient from napalm_procurve import parsers +from napalm_procurve.api_client import ProcurveApiClient logger = logging.getLogger("napalm_procurve") @@ -118,7 +114,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): #: Where the switch publishes its health values over SNMP #: (napalm-device-types ``SnmpHealthProfile``); the application polls. #: Memory is NETSWITCH-MIB ``hpLocalMemEntry``: .5 total, .6 free, .7 allocated. - SNMP_HEALTH_PROFILE: ClassVar[Dict[str, Any]] = { + SNMP_HEALTH_PROFILE: ClassVar[dict[str, Any]] = { "uptime": {"oids": ["1.3.6.1.4.1.11.2.14.11.1.2.4.1.5.1"]}, "cpu": {"oids": ["1.3.6.1.4.1.11.2.14.11.5.1.9.6.1.0", "1.3.6.1.4.1.11.2.3.7.8.6.1.0"]}, "memory": { @@ -134,7 +130,7 @@ class ProcurveDriver(ConfigLifecycleMixin, 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 @@ -147,11 +143,11 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # --- optional_args --- self.port: int = optional_args.get("port", 22) self.telnet_port: int = optional_args.get("telnet_port", 23) - self.force_transport: Optional[str] = optional_args.get("transport") + self.force_transport: str | None = optional_args.get("transport") self.ssl_verify: bool = optional_args.get( "ssl_verify", optional_args.get("verify_ssl", optional_args.get("verify", False)) ) - self.api_version_hint: Optional[str] = optional_args.get("api_version") + self.api_version_hint: str | None = optional_args.get("api_version") # Enable password for CLI (defaults to login password) self._secret: str = optional_args.get("secret", password) @@ -160,24 +156,24 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): self.netmiko_optional_args = netmiko_args(optional_args) # The host keys to expect and where the one seen is reported (netork#1167). - self._host_key_args: Dict[str, Any] = { + self._host_key_args: dict[str, Any] = { k: optional_args[k] for k in (KNOWN_KEYS_ARG, OBSERVER_ARG) if k in optional_args } # --- runtime state --- # Active transport: "api" | "ssh" | "ssh_legacy" | "telnet" - self._transport: Optional[str] = None + self._transport: str | None = None # CLI backend (SSH or Telnet) - self._device: Optional[ConnectHandler] = None + self._device: ConnectHandler | None = None # REST API backend - self._api: Optional[ProcurveApiClient] = None + self._api: ProcurveApiClient | None = None # Why each transport failed during the current open(), as ": " - self._attempts: List[str] = [] + self._attempts: list[str] = [] # Config management state (CLI only) - self._candidate_config: Optional[str] = None - self._candidate_mode: Optional[str] = None # "merge" | "replace" - self._backup_config: Optional[str] = None + self._candidate_config: str | None = None + self._candidate_mode: str | None = None # "merge" | "replace" + self._backup_config: str | None = None # ------------------------------------------------------------------ # Connection management @@ -246,14 +242,14 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): self._device = None self._transport = None - def is_alive(self) -> Dict[str, bool]: + def is_alive(self) -> dict[str, bool]: """Return transport-level liveness without writing to the channel.""" if self._api: return {"is_alive": self._api.is_alive()} if self._device: try: return {"is_alive": self._device.remote_conn.transport.is_active()} - except (AttributeError, socket.error, EOFError): + except (OSError, AttributeError, EOFError): return {"is_alive": False} return {"is_alive": False} @@ -267,9 +263,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): logger.info("Probing REST API for %s (timeout=%ds)", self.hostname, probe_timeout) # Probe without SSL verification — no credentials are sent during probing, # so this is safe and avoids failing on self-signed certificates. - ver, proto = ProcurveApiClient.probe( - self.hostname, timeout=probe_timeout, ssl_verify=False - ) + ver, proto = ProcurveApiClient.probe(self.hostname, timeout=probe_timeout, ssl_verify=False) # Allow hint override (e.g. user knows the API version) if self.api_version_hint and proto: ver = self.api_version_hint @@ -281,8 +275,8 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # Try connecting with the requested SSL setting first; if it fails due to a # self-signed certificate (ssl_verify=True), transparently retry unverified. - error: Optional[Exception] = None - for ssl_verify in ([self.ssl_verify] if not self.ssl_verify else [True, False]): + error: Exception | None = None + for ssl_verify in [self.ssl_verify] if not self.ssl_verify else [True, False]: client = ProcurveApiClient( hostname=self.hostname, username=self.username, @@ -300,8 +294,13 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): continue self._api = client self._transport = "api" - logger.info("Connected to %s via REST API (%s %s, ssl_verify=%s)", - self.hostname, proto, ver, ssl_verify) + logger.info( + "Connected to %s via REST API (%s %s, ssl_verify=%s)", + self.hostname, + proto, + ver, + ssl_verify, + ) return True logger.warning("REST API connect failed for %s — falling back to CLI", self.hostname) @@ -314,8 +313,11 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): Explicit fields (port, timeout, disabled_algorithms) take precedence and are excluded from the spread to avoid 'multiple values for keyword argument' errors. """ - extra = {k: v for k, v in self.netmiko_optional_args.items() - if k not in ("port", "timeout", "disabled_algorithms")} + extra = { + k: v + for k, v in self.netmiko_optional_args.items() + if k not in ("port", "timeout", "disabled_algorithms") + } return dict( device_type=self.NETMIKO_DEVICE_TYPE, host=self.hostname, @@ -328,7 +330,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): **extra, ) - def _ssh_connect(self, legacy: bool = False, port: Optional[int] = None) -> ConnectHandler: + def _ssh_connect(self, legacy: bool = False, port: int | None = None) -> ConnectHandler: """A netmiko SSH session, its host key verified and reported (netork#1167). Every SSH session of this driver opens here. A refused key raises @@ -397,7 +399,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # CLI helpers # ------------------------------------------------------------------ - def _send_command(self, command: Union[str, List[str]]) -> str: + def _send_command(self, command: str | list[str]) -> str: """Send a CLI command and return stripped output. When a list is supplied, commands are tried in order and the first @@ -421,11 +423,15 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): for cmd in command: last = _do(cmd) # ProCurve says "Invalid input: …"; other firmware "% Invalid …" - if "Invalid input" not in last and "% Invalid" not in last and "Error" not in last: + if ( + "Invalid input" not in last + and "% Invalid" not in last + and "Error" not in last + ): return last return last return _do(command) - except (socket.error, EOFError) as exc: + except (OSError, EOFError) as exc: raise ConnectionClosedException(str(exc)) from exc def _exec_prompt(self) -> str: @@ -462,13 +468,13 @@ class ProcurveDriver(ConfigLifecycleMixin, 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]: """Push config lines to the device while in config mode. Returns a list of error messages for rejected lines. """ ep = self._any_prompt() - errors: List[str] = [] + errors: list[str] = [] for line in config_text.splitlines(): stripped = line.strip() if not stripped or stripped.startswith(("!", "#", ";")): @@ -486,7 +492,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # NAPALM: get_facts # ------------------------------------------------------------------ - def get_facts(self) -> Dict: + def get_facts(self) -> dict: """Return general device information.""" if self._transport == "api": return self._api.get_facts() @@ -520,22 +526,20 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): "interface_list": iface_list, } - def _get_interface_list(self) -> List[str]: + def _get_interface_list(self) -> list[str]: """Return sorted list of interface names from ``show interfaces brief``.""" out = self._send_command("show interfaces brief") ifaces = parsers.parse_interfaces_brief(out) return sorted( ifaces.keys(), - key=lambda s: [ - int(x) if x.isdigit() else x for x in re.split(r"(\d+)", s) - ], + key=lambda s: [int(x) if x.isdigit() else x for x in re.split(r"(\d+)", s)], ) # ------------------------------------------------------------------ # NAPALM: get_interfaces # ------------------------------------------------------------------ - def get_interfaces(self) -> Dict[str, Dict]: + def get_interfaces(self) -> dict[str, dict]: """Return a dictionary of interface details.""" if self._transport == "api": return self._api.get_interfaces() @@ -561,7 +565,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # NAPALM: get_interfaces_ip # ------------------------------------------------------------------ - def get_interfaces_ip(self) -> Dict[str, Dict]: + def get_interfaces_ip(self) -> dict[str, dict]: """Return IP addresses assigned to interfaces.""" if self._transport == "api": return self._api.get_interfaces_ip() @@ -573,7 +577,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # NAPALM: get_arp_table # ------------------------------------------------------------------ - def get_arp_table(self, vrf: str = "") -> List[Dict]: + def get_arp_table(self, vrf: str = "") -> list[dict]: """Return the ARP table.""" if self._transport == "api": return self._api.get_arp_table(vrf) @@ -585,7 +589,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # NAPALM: get_mac_address_table # ------------------------------------------------------------------ - def get_mac_address_table(self) -> List[Dict]: + def get_mac_address_table(self) -> list[dict]: """Return the MAC address table.""" if self._transport == "api": return self._api.get_mac_address_table() @@ -597,21 +601,19 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # NAPALM: get_lldp_neighbors # ------------------------------------------------------------------ - def get_lldp_neighbors(self) -> Dict[str, List[Dict]]: + def get_lldp_neighbors(self) -> dict[str, list[dict]]: """Return LLDP neighbor information.""" if self._transport == "api": return self._api.get_lldp_neighbors() - out = self._send_command( - ["show lldp info remote-device", "show lldp neighbors"] - ) + out = self._send_command(["show lldp info remote-device", "show lldp neighbors"]) return parsers.parse_lldp_neighbors(out) # ------------------------------------------------------------------ # NAPALM: get_lldp_neighbors_detail # ------------------------------------------------------------------ - 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.""" if self._transport == "api": return self._api.get_lldp_neighbors_detail() @@ -629,7 +631,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): def get_config( self, retrieve: str = "all", full: bool = False, sanitized: bool = False - ) -> Dict[str, str]: + ) -> dict[str, str]: """Return device configuration.""" if self._transport == "api": result = self._api.get_config() @@ -652,9 +654,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): if retrieve in ("all", "running"): running = self._send_command("show running-config") if retrieve in ("all", "startup"): - startup = self._send_command( - ["show startup-config", "show config"] - ) + startup = self._send_command(["show startup-config", "show config"]) return {"running": running, "startup": startup, "candidate": candidate} @@ -680,14 +680,12 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # NAPALM: get_environment # ------------------------------------------------------------------ - def get_environment(self) -> Dict: + def get_environment(self) -> dict: """Return environment information (temperature, fans, CPU, memory).""" if self._transport == "api": raise NotImplementedError("Environment data not available via REST API") - sysinfo = self._send_command( - ["show system-information", "show system information"] - ) + sysinfo = self._send_command(["show system-information", "show system information"]) env_out = self._send_command(["show environment", "show system"]) combined = sysinfo + "\n" + env_out return parsers.parse_environment(combined) @@ -696,25 +694,21 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # NAPALM: get_ntp_servers # ------------------------------------------------------------------ - def get_ntp_servers(self) -> Dict[str, Dict]: + def get_ntp_servers(self) -> dict[str, dict]: """Return configured NTP servers.""" if self._transport == "api": return self._api.get_ntp_servers() - out = self._send_command( - ["show ntp associations", "show ntp status", "show timesync"] - ) + out = self._send_command(["show ntp associations", "show ntp status", "show timesync"]) return parsers.parse_ntp_servers(out) # ------------------------------------------------------------------ # NAPALM: get_ntp_stats # ------------------------------------------------------------------ - def get_ntp_stats(self) -> List[Dict]: + def get_ntp_stats(self) -> list[dict]: """Return NTP peer statistics.""" - out = self._send_command( - ["show ntp associations", "show ntp status"] - ) + out = self._send_command(["show ntp associations", "show ntp status"]) stats = [] for line in out.splitlines(): m = re.match( @@ -746,7 +740,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # NAPALM: get_users # ------------------------------------------------------------------ - def get_users(self) -> Dict[str, Dict]: + def get_users(self) -> dict[str, dict]: """Return local users.""" out = self._send_command(["show local-users", "show users"]) return parsers.parse_users(out) @@ -755,7 +749,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # NAPALM: get_snmp_information # ------------------------------------------------------------------ - def get_snmp_information(self) -> Dict: + def get_snmp_information(self) -> dict: """Return SNMP configuration.""" out = self._send_command(["show snmp-server", "show snmp"]) return parsers.parse_snmp_info(out) @@ -764,7 +758,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # NAPALM: get_vlans (extended: tagged + untagged per VLAN) # ------------------------------------------------------------------ - def get_vlans(self) -> Dict[int, Dict]: + def get_vlans(self) -> dict[int, dict]: """Return VLAN information with tagged/untagged port separation. This method extends the standard NAPALM ``get_vlans()`` return with @@ -811,18 +805,18 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # Fallback: show vlans (name discovery) + show vlans per VLAN return self._get_vlans_via_show_vlans() - def _get_vlans_via_show_vlans(self) -> Dict[int, Dict]: + def _get_vlans_via_show_vlans(self) -> dict[int, dict]: """Fallback VLAN discovery via ``show vlans`` + per-VLAN detail queries.""" brief_out = self._send_command("show vlans") vlan_names = parsers.parse_vlans_brief(brief_out) - result: Dict[int, Dict] = {} + result: dict[int, dict] = {} for vid, name in vlan_names.items(): detail_out = self._send_command(f"show vlans {vid}") ports = parsers.parse_vlan_ports_detail(detail_out) tagged = ports["tagged"] untagged = ports["untagged"] - interfaces: List[str] = [] + interfaces: list[str] = [] for p in untagged + tagged: if p not in interfaces: interfaces.append(p) @@ -839,7 +833,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # Custom: get_port_pvids (PVID per port, derived from get_vlans) # ------------------------------------------------------------------ - def get_port_pvids(self) -> Dict[str, int]: + def get_port_pvids(self) -> dict[str, int]: """Return the PVID (native / access VLAN) for every switch port. On ProCurve, a port is untagged in exactly **one** VLAN — that VLAN's @@ -906,9 +900,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): lines.append("exit") errors = self._apply_config_lines("\n".join(lines)) if errors: - raise CommandErrorException( - f"set_vlan({vlan_id}) errors: {errors}" - ) + raise CommandErrorException(f"set_vlan({vlan_id}) errors: {errors}") finally: self._exit_config_mode() self._save_config() @@ -917,7 +909,7 @@ class ProcurveDriver(ConfigLifecycleMixin, 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. Diffs ``members`` against the trunk's current members (as reported @@ -939,7 +931,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): mode = current.get("lag_mode") or "trunk" trk_name = lag_name.lower() - commands: List[str] = [] + commands: list[str] = [] if to_remove: commands.append(f"no trunk {','.join(to_remove)}") if to_add: @@ -1024,7 +1016,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): mode = config.get("mode") self._enter_config_mode() try: - lines: List[str] = [] + lines: list[str] = [] if "enabled" in config or "description" in config: lines.append(f"interface {interface}") if "enabled" in config: @@ -1049,9 +1041,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): if lines: errors = self._apply_config_lines("\n".join(lines)) if errors: - raise CommandErrorException( - f"set_interface({interface}) errors: {errors}" - ) + raise CommandErrorException(f"set_interface({interface}) errors: {errors}") finally: self._exit_config_mode() self._save_config() @@ -1068,9 +1058,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): try: errors = self._apply_config_lines(f"no vlan {vlan_id}") if errors: - raise CommandErrorException( - f"delete_vlan({vlan_id}) errors: {errors}" - ) + raise CommandErrorException(f"delete_vlan({vlan_id}) errors: {errors}") finally: self._exit_config_mode() self._save_config() @@ -1079,7 +1067,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # PoE # ------------------------------------------------------------------ - def get_poe_status(self) -> Dict[str, Dict]: + def get_poe_status(self) -> dict[str, dict]: """Return PoE configuration for all ports, keyed by interface name. Only available via the REST API transport. @@ -1088,7 +1076,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): return self._api.get_poe_ports() raise NotImplementedError("PoE status is only available via the REST API") - def set_poe(self, interface: str, config: Dict) -> None: + def set_poe(self, interface: str, config: dict) -> None: """Update PoE configuration for a single interface.""" if self._transport != "api": raise NotImplementedError("PoE configuration is only available via the REST API") @@ -1125,12 +1113,17 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): size: int = 100, count: int = 5, vrf: str = "", - ) -> Dict: + ) -> dict: """Execute a ping from the device.""" if self._transport == "api": return self._api.ping( - destination, source=source, ttl=ttl, - timeout=timeout, size=size, count=count, vrf=vrf, + destination, + source=source, + ttl=ttl, + timeout=timeout, + size=size, + count=count, + vrf=vrf, ) # ProCurve CLI ping: "ping repetitions " @@ -1144,11 +1137,11 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # NAPALM: cli # ------------------------------------------------------------------ - def cli(self, commands: List[str], encoding: str = "text") -> Dict[str, str]: + def cli(self, commands: list[str], encoding: str = "text") -> dict[str, str]: """Execute raw CLI commands and return output dict.""" if self._transport == "api": # ArubaOS REST API supports CLI via POST /rest//cli - results: Dict[str, str] = {} + results: dict[str, str] = {} for cmd in commands: resp = self._api.post("cli", json={"cmd": cmd}) if resp.ok: @@ -1166,9 +1159,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): def _get_running_config(self) -> str: return self._send_command("show running-config") - def load_merge_candidate( - self, filename: Optional[str] = None, config: Optional[str] = None - ) -> None: + def load_merge_candidate(self, filename: str | None = None, config: str | None = None) -> None: if self._transport == "api": raise MergeConfigException( "Merge config is not supported via REST API transport. Use SSH." @@ -1176,7 +1167,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): super().load_merge_candidate(filename, config) def load_replace_candidate( - self, filename: Optional[str] = None, config: Optional[str] = None + self, filename: str | None = None, config: str | None = None ) -> None: if self._transport == "api": raise ReplaceConfigException( @@ -1184,7 +1175,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): ) super().load_replace_candidate(filename, 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 candidate configuration to the device.""" if self._candidate_config is None: raise MergeConfigException("No candidate config loaded.") @@ -1201,9 +1192,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): raise MergeConfigException(f"Error applying config: {exc}") from exc if errors: - raise MergeConfigException( - "The following lines were rejected:\n" + "\n".join(errors) - ) + raise MergeConfigException("The following lines were rejected:\n" + "\n".join(errors)) self._save_config() self._candidate_config = None @@ -1224,11 +1213,15 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # Use a single command with short read_timeout to avoid blocking the poll. try: - out = self._device.send_command( - "show snmp-server", - expect_string=rf"{re.escape(self._device.base_prompt)}[>#]", - read_timeout=10, - ).strip() if self._device else "" + out = ( + self._device.send_command( + "show snmp-server", + expect_string=rf"{re.escape(self._device.base_prompt)}[>#]", + read_timeout=10, + ).strip() + if self._device + else "" + ) except Exception: return None if not out: @@ -1256,7 +1249,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # SNMP agent (napalm-device-types SnmpAgentMixin): the caller's values # ------------------------------------------------------------------ - 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 agent is built in; configure community, contact and location, then save. The REST API has no writable SNMP endpoint, so over REST a temporary SSH @@ -1296,7 +1289,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): pass self._api, self._device, self._transport = old_api, old_device, old_transport - def _snmp_configure_cli(self, settings: Dict[str, Any]) -> Dict[str, Any]: + def _snmp_configure_cli(self, settings: dict[str, Any]) -> dict[str, Any]: community = settings["community"] if not _CLI_WORD.match(community): raise ValueError("the community must be one word without quotes or #") @@ -1317,12 +1310,12 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): return {"ok": True, "output": "community (read-only), contact and location set"} @staticmethod - def _snmp_sources_note(settings: Dict[str, Any]) -> Dict[str, Any]: + def _snmp_sources_note(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 _snmp_save_and_verify(self, settings: Dict[str, Any]) -> Dict[str, Any]: + def _snmp_save_and_verify(self, settings: dict[str, Any]) -> dict[str, Any]: self._save_config() shown = self._send_command("show snmp-server") if re.search(rf"^\s*{re.escape(settings['community'])}\s", shown, re.M): @@ -1334,6 +1327,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): # Helpers # --------------------------------------------------------------------------- + def _is_float(s: str) -> bool: try: float(s) @@ -1342,7 +1336,7 @@ def _is_float(s: str) -> bool: return False -def _parse_ping_output(output: str, count: int) -> Dict: +def _parse_ping_output(output: str, count: int) -> dict: """Parse ProCurve CLI ping output into NAPALM ping result dict. ProCurve ping output:: @@ -1356,7 +1350,8 @@ def _parse_ping_output(output: str, count: int) -> Dict: # Packet statistics line m = re.search( r"(\d+)\s+packets\s+transmitted,\s+(\d+)\s+packets\s+received", - output, re.I, + output, + re.I, ) if not m: # Check for total failure @@ -1399,4 +1394,3 @@ def _parse_ping_output(output: str, count: int) -> Dict: "results": results, } } - diff --git a/pyproject.toml b/pyproject.toml index 0f8e0e5..fba2f8b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,4 +54,11 @@ include = ["napalm_procurve*"] [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 9def7a8..f8f4818 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -1,13 +1,12 @@ """Unit tests for ProcurveDriver and parsers — no real device required.""" -import pytest from unittest.mock import MagicMock, patch +import pytest from napalm.base.exceptions import ConnectionException -from napalm_procurve.procurve import _SSH_DISABLED_STANDARD, ProcurveDriver, _parse_ping_output from napalm_procurve import parsers - +from napalm_procurve.procurve import _SSH_DISABLED_STANDARD, ProcurveDriver, _parse_ping_output # =========================================================================== # Sample CLI output fixtures (captured from real ProCurve devices) @@ -510,7 +509,9 @@ class TestDriverTransportDetection: with patch("napalm_procurve.procurve.ConnectHandler") as mock_conn: mock_conn.return_value = MagicMock() drv = ProcurveDriver( - "192.168.0.1", "manager", "secret", + "192.168.0.1", + "manager", + "secret", optional_args={"transport": "ssh"}, ) drv.open() @@ -521,7 +522,9 @@ class TestDriverTransportDetection: with patch("napalm_procurve.procurve.ConnectHandler") as mock_conn: mock_conn.return_value = MagicMock() drv = ProcurveDriver( - "192.168.0.1", "manager", "secret", + "192.168.0.1", + "manager", + "secret", optional_args={"transport": "telnet"}, ) drv.open() @@ -565,7 +568,9 @@ class TestDriverTransportDetection: def test_all_transports_fail_names_each_reason(self): """The final error gives a reason per transport, not just its name.""" with patch("napalm_procurve.procurve.ProcurveApiClient.probe", return_value=(None, None)): - with patch("napalm_procurve.procurve.ConnectHandler", side_effect=OSError("no route to host")): + with patch( + "napalm_procurve.procurve.ConnectHandler", side_effect=OSError("no route to host") + ): drv = ProcurveDriver("192.168.0.1", "manager", "secret") with pytest.raises(ConnectionException) as exc_info: drv.open() @@ -745,8 +750,8 @@ class TestParseVlanPortsDetail: class TestDerivePvids: def test_basic_pvid_derivation(self): vlans = { - 1: {"untagged": ["1", "2", "3"], "tagged": []}, - 10: {"untagged": ["4"], "tagged": ["5"]}, + 1: {"untagged": ["1", "2", "3"], "tagged": []}, + 10: {"untagged": ["4"], "tagged": ["5"]}, } pvids = parsers.derive_port_pvids(vlans) assert pvids["1"] == 1 @@ -756,7 +761,7 @@ class TestDerivePvids: def test_tagged_only_port_gets_pvid1(self): vlans = { - 1: {"untagged": [], "tagged": []}, + 1: {"untagged": [], "tagged": []}, 10: {"untagged": [], "tagged": ["5"]}, } pvids = parsers.derive_port_pvids(vlans) @@ -793,7 +798,7 @@ class TestDriverGetVlans: def side_effect(cmd, **kw): call_count[0] += 1 if "running-config" in (cmd if isinstance(cmd, str) else cmd[0]): - return "hostname myswitch\n" # no vlan blocks + return "hostname myswitch\n" # no vlan blocks if "show vlans" == (cmd if isinstance(cmd, str) else cmd[0]): return SHOW_VLANS_BRIEF if "show vlans 10" in (cmd if isinstance(cmd, str) else cmd[0]): @@ -900,22 +905,36 @@ class TestApiGetInterfacesLag: client = ProcurveApiClient(hostname="192.168.0.1", username="manager", password="secret") responses = { "ports": {"port_element": ports}, - "port-statistics": {"port_statistics_element": [ - {"id": p["id"], "port_speed_mbps": 1000} for p in ports - ]}, + "port-statistics": { + "port_statistics_element": [{"id": p["id"], "port_speed_mbps": 1000} for p in ports] + }, "system/status/switch": {}, } client.get = MagicMock(side_effect=lambda endpoint, **kw: responses[endpoint]) return client def test_trunk_rows_with_mode(self): - client = self._client([ - {"id": "1", "is_port_up": True, "is_port_enabled": True, "trunk_group": ""}, - {"id": "3", "is_port_up": False, "is_port_enabled": True, "trunk_group": "trk3", "trunk_mode": "PTT_LACP"}, - {"id": "4", "is_port_up": True, "is_port_enabled": True, "trunk_group": "trk3", "trunk_mode": "PTT_LACP"}, - {"id": "10", "is_port_up": False, "is_port_enabled": True, "trunk_group": "trk6"}, - {"id": "7", "is_port_up": False, "is_port_enabled": True, "trunk_group": "trk6"}, - ]) + client = self._client( + [ + {"id": "1", "is_port_up": True, "is_port_enabled": True, "trunk_group": ""}, + { + "id": "3", + "is_port_up": False, + "is_port_enabled": True, + "trunk_group": "trk3", + "trunk_mode": "PTT_LACP", + }, + { + "id": "4", + "is_port_up": True, + "is_port_enabled": True, + "trunk_group": "trk3", + "trunk_mode": "PTT_LACP", + }, + {"id": "10", "is_port_up": False, "is_port_enabled": True, "trunk_group": "trk6"}, + {"id": "7", "is_port_up": False, "is_port_enabled": True, "trunk_group": "trk6"}, + ] + ) ifaces = client.get_interfaces() assert ifaces["trk3"]["lag_members"] == ["3", "4"] @@ -931,7 +950,9 @@ class TestApiGetInterfacesLag: class TestCliGetInterfacesLag: def test_trunk_rows_from_member_ports(self, driver): driver._send_command = MagicMock( - side_effect=lambda cmd: SHOW_INTERFACES_BRIEF_INTRUSION if cmd == "show interfaces brief" else "" + side_effect=lambda cmd: ( + SHOW_INTERFACES_BRIEF_INTRUSION if cmd == "show interfaces brief" else "" + ) ) ifaces = driver.get_interfaces() diff --git a/tests/unit/test_ssh_host_key.py b/tests/unit/test_ssh_host_key.py index ec90ab4..1b6e1eb 100644 --- a/tests/unit/test_ssh_host_key.py +++ b/tests/unit/test_ssh_host_key.py @@ -30,7 +30,9 @@ HANDLER = "napalm_procurve.procurve.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 + ) device.send_command.return_value = "hostname x" return device