NAPALM/napalm-hpe-aruba-procurve
1
0 Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

style: ruff format and the lint rules of the driver standard

Browse Source 
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.
This commit is contained in:
Christian Manivong committed 2026-10-23 08:32:50 +02:00
1 parent 739283e76b
commit c2a788fa7a
6 files changed
+307 -278
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

napalm_procurve/api_client.py
+65 -59
View File
@@ -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
napalm_procurve/parsers.py
+72 -73
View File
@@ -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 <port>`` 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 <id>`` 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()):
napalm_procurve/procurve.py
+117 -123
View File
@@ -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 "<transport>: <reason>"
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 <id> 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 <ip> repetitions <n>"
@@ -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/<ver>/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,
}
}
pyproject.toml
+8 -1
View File
@@ -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"]
tests/unit/test_driver.py
+42 -21
View File
@@ -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()
tests/unit/test_ssh_host_key.py
+3 -1
View File
@@ -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