From 35032fd034d02ec99e4994639e6a9daa60bd8819 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Fri, 29 May 2026 09:30:53 +0200 Subject: [PATCH] initial commit --- README.md | 111 ++++ napalm_procurve/__init__.py | 5 + napalm_procurve/api_client.py | 524 ++++++++++++++++ napalm_procurve/parsers.py | 886 +++++++++++++++++++++++++++ napalm_procurve/procurve.py | 1066 +++++++++++++++++++++++++++++++++ pyproject.toml | 55 ++ tests/__init__.py | 1 + tests/unit/__init__.py | 0 tests/unit/test_driver.py | 653 ++++++++++++++++++++ 9 files changed, 3301 insertions(+) create mode 100644 README.md create mode 100644 napalm_procurve/__init__.py create mode 100644 napalm_procurve/api_client.py create mode 100644 napalm_procurve/parsers.py create mode 100644 napalm_procurve/procurve.py create mode 100644 pyproject.toml create mode 100644 tests/__init__.py create mode 100644 tests/unit/__init__.py create mode 100644 tests/unit/test_driver.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..bd410b9 --- /dev/null +++ b/README.md @@ -0,0 +1,111 @@ +# napalm-hpe-aruba-procurve + +NAPALM driver for **HPE / Aruba ProCurve** switches — from ancient 2520G-8-PoE +to modern 2530/2540 series. + +## Transport auto-detection + +The driver probes transports in order and uses the first one that succeeds: + +| Priority | Transport | Notes | +|----------|-----------|-------| +| 1 | REST API (HTTPS → HTTP, v7 → v6) | Newer 2530/2540 with `rest-interface` enabled | +| 2 | SSH (standard) | hp_procurve netmiko driver | +| 3 | SSH (legacy KEX) | Forces older kex/cipher negotiation for 2520G-8-PoE etc. | +| 4 | Telnet | Oldest devices without SSH or with broken SSH | + +Override the transport with `optional_args={"transport": "ssh_legacy"}` to skip +auto-detection. + +## Switch prerequisites + +### REST API (newer switches, e.g. 2530, 2540) + +``` +web-management ssl +rest-interface +rest-interface session-idle-timeout 120 +``` + +### SSH (all ProCurve) + +``` +crypto key generate ssh rsa +ip ssh +``` + +Telnet is enabled by default on most ProCurve switches. + +## Installation + +```bash +pip install napalm napalm-hpe-aruba-procurve +``` + +Or from source: + +```bash +git clone https://github.com/napalm-automation-community/napalm-hpe-aruba-procurve +pip install -e napalm-hpe-aruba-procurve/ +``` + +## Quick start + +```python +from napalm import get_network_driver + +Driver = get_network_driver("procurve") + +with Driver( + "10.0.0.1", + "manager", + "secret", + optional_args={ + # "transport": "ssh", # force transport (api/ssh/ssh_legacy/telnet) + # "port": 22, + # "ssl_verify": False, # disable SSL cert check for API + # "api_version": "v7", # API version hint (v3/v6/v7) + # "secret": "enablepassword", # enable password for CLI + }, +) as dev: + print(dev.get_facts()) + print(dev.get_interfaces()) +``` + +## Supported NAPALM methods + +| Method | API | SSH/Telnet | +|--------|-----|------------| +| `open()` | ✅ | ✅ | +| `close()` | ✅ | ✅ | +| `is_alive()` | ✅ | ✅ | +| `get_facts()` | ✅ | ✅ | +| `get_interfaces()` | ✅ | ✅ | +| `get_interfaces_ip()` | ✅ | ✅ | +| `get_arp_table()` | ✅ | ✅ | +| `get_mac_address_table()` | ✅ | ✅ | +| `get_lldp_neighbors()` | ✅ | ✅ | +| `get_lldp_neighbors_detail()` | ✅ | ✅ | +| `get_config()` | ✅ | ✅ | +| `get_ntp_servers()` | ✅ | ✅ | +| `get_environment()` | ❌ | ✅ | +| `get_users()` | ❌ | ✅ | +| `get_snmp_information()` | ❌ | ✅ | +| `ping()` | ✅ | ✅ | +| `cli()` | ✅ | ✅ | +| `load_merge_candidate()` | ❌ | ✅ | +| `load_replace_candidate()` | ❌ | ✅ | +| `compare_config()` | ❌ | ✅ | +| `commit_config()` | ❌ | ✅ | +| `discard_config()` | ❌ | ✅ | +| `rollback()` | ❌ | ✅ | + +## Tested devices + +- HP ProCurve 2520G-8-PoE (J9565A) — SSH legacy / Telnet +- HP ProCurve 2920-48G — SSH +- Aruba 2530-8-PoE+ — SSH + REST API + +## License + +Apache 2.0 diff --git a/napalm_procurve/__init__.py b/napalm_procurve/__init__.py new file mode 100644 index 0000000..d05e004 --- /dev/null +++ b/napalm_procurve/__init__.py @@ -0,0 +1,5 @@ +"""NAPALM driver for HPE/Aruba ProCurve switches.""" + +from napalm_procurve.procurve import ProcurveDriver + +__all__ = ["ProcurveDriver"] diff --git a/napalm_procurve/api_client.py b/napalm_procurve/api_client.py new file mode 100644 index 0000000..a982df1 --- /dev/null +++ b/napalm_procurve/api_client.py @@ -0,0 +1,524 @@ +# -*- coding: utf-8 -*- +# Licensed under the Apache License, Version 2.0 + +"""REST API client for HPE/Aruba ProCurve switches (ArubaOS REST interface). + +Compatible with ArubaOS REST API versions v3, v6, and v7. +Used automatically by ProcurveDriver when the REST API is reachable. +""" + +import logging +from typing import Any, Dict, Optional, Tuple + +import requests +import urllib3 + +from napalm.base.exceptions import ConnectionException, ConnectAuthError + +logger = logging.getLogger("napalm_procurve.api") + + +class ProcurveApiClient: + """Thin wrapper around ArubaOS REST API sessions. + + Detection approach: probe ``GET /rest//system/status`` without + authentication. A ``401`` response confirms the API is available. + """ + + # Ordered list of API versions to probe (newest first) + API_VERSIONS = ["v7", "v6", "v3"] + + def __init__( + self, + hostname: str, + username: str, + password: str, + timeout: int = 60, + ssl_verify: bool = False, + api_version: Optional[str] = None, + ) -> None: + self.hostname = hostname + self.username = username + self.password = password + self.timeout = timeout + self.ssl_verify = ssl_verify + self._api_version = api_version # None = auto-detect + self._proto: str = "https" + self._base_url: str = "" + self._session: Optional[requests.Session] = None + + # ------------------------------------------------------------------ + # Detection + # ------------------------------------------------------------------ + + @classmethod + def probe( + cls, + hostname: str, + timeout: int = 5, + ssl_verify: bool = False, + ) -> Tuple[Optional[str], Optional[str]]: + """Probe the device for an ArubaOS REST API. + + Returns ``(api_version, proto)`` on success, ``(None, None)`` if the + API is unreachable or the device is not an ArubaOS switch. + """ + if not ssl_verify: + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + + for proto in ("https", "http"): + for ver in cls.API_VERSIONS: + url = f"{proto}://{hostname}/rest/{ver}/system/status" + try: + resp = requests.get( + url, + verify=ssl_verify, + timeout=timeout, + allow_redirects=False, + ) + # 401 = API exists but not authenticated (most common) + # 200 = API exists and accessible without auth (unusual) + # 405 = wrong HTTP verb but API is there + if resp.status_code in (200, 401, 405): + logger.debug( + "API detected at %s (HTTP %s, %s %s)", + hostname, resp.status_code, proto, ver, + ) + return ver, proto + except requests.exceptions.SSLError: + # SSL error on https → try http next + continue + except requests.exceptions.ConnectionError: + # No route to host / port closed → next proto/ver + break + except requests.exceptions.Timeout: + break + return None, None + + # ------------------------------------------------------------------ + # Session management + # ------------------------------------------------------------------ + + def connect(self) -> None: + """Open an authenticated REST API session.""" + if not self._base_url: + raise ConnectionException( + "ProcurveApiClient.connect() called before api_version/proto set. " + "Use ProcurveApiClient.probe() first." + ) + if not self.ssl_verify: + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + + self._session = requests.Session() + self._session.verify = self.ssl_verify + self._session.headers.update( + {"Content-Type": "application/json", "Connection": "close"} + ) + + url = self._base_url + "login-sessions" + resp = self._session.post( + url, + json={"userName": self.username, "password": self.password}, + timeout=self.timeout, + ) + if resp.status_code != 201: + raise ConnectAuthError( + f"REST API login failed (HTTP {resp.status_code}) for {self.hostname}" + ) + logger.debug("REST API login OK for %s", self.hostname) + + def setup(self, api_version: str, proto: str) -> None: + """Set the API version and protocol (called after probe()).""" + self._api_version = api_version + self._proto = proto + self._base_url = f"{proto}://{self.hostname}/rest/{api_version}/" + + def disconnect(self) -> None: + """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 + ) + except Exception: + pass + if self._session: + self._session.close() + self._session = None + + def is_alive(self) -> bool: + """Check if the API session is still usable.""" + if not self._session: + return False + try: + resp = self._session.get( + self._base_url + "system/status", timeout=5 + ) + return resp.status_code == 200 + except Exception: + return False + + # ------------------------------------------------------------------ + # HTTP helpers + # ------------------------------------------------------------------ + + 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: + resp = self._session.get(url, timeout=self.timeout) + if resp.ok: + return resp.json() + logger.debug("GET %s returned HTTP %s", url, resp.status_code) + except Exception as exc: + logger.debug("GET %s error: %s", url, exc) + return {} + + def post(self, endpoint: str, **kwargs: Any) -> requests.Response: + """POST ``{base_url}{endpoint}``.""" + return self._session.post(self._base_url + endpoint, timeout=self.timeout, **kwargs) + + def put(self, endpoint: str, **kwargs: Any) -> requests.Response: + """PUT ``{base_url}{endpoint}``.""" + return self._session.put(self._base_url + endpoint, timeout=self.timeout, **kwargs) + + def delete(self, endpoint: str, **kwargs: Any) -> requests.Response: + """DELETE ``{base_url}{endpoint}``.""" + return self._session.delete(self._base_url + endpoint, timeout=self.timeout, **kwargs) + + # ------------------------------------------------------------------ + # NAPALM data getters (API-backed) + # ------------------------------------------------------------------ + + def get_facts(self) -> Dict: + """Return NAPALM facts from the REST API.""" + system = self.get("system/status") + if not system: + # Fallback endpoint for stacked switches + system = self.get("system/status/global_info") + + dns = self.get("dns") + domain = "" + if dns: + domains = dns.get("dns_domain_names", []) + domain = f".{domains[0]}" if domains else "" + + hostname = system.get("name", "") + fqdn = f"{hostname}{domain}" if domain else hostname + + # Interface list from blades + iface_list = [] + sw_status = self.get("system/status/switch") + for blade in sw_status.get("blades", []): + for port in blade.get("data_ports", []): + iface_list.append(port.get("port_name", "")) + + return { + "vendor": "HPE Aruba", + "model": system.get("product_model", ""), + "hostname": hostname, + "fqdn": fqdn, + "os_version": system.get("firmware_version", ""), + "serial_number": system.get("serial_number", ""), + "uptime": float(system.get("uptime_seconds", -1)), + "interface_list": iface_list, + } + + 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] = {} + + for port in ports_data.get("port_element", []): + pid = port.get("id", "") + output[pid] = { + "is_up": bool(port.get("is_port_up", False)), + "is_enabled": bool(port.get("is_port_enabled", False)), + "description": port.get("name", ""), + "last_flapped": -1.0, + "speed": 0.0, + "mtu": -1, + "mac_address": "", + } + + for stat in stats_data.get("port_statistics_element", []): + pid = stat.get("id", "") + if pid in output: + output[pid]["speed"] = float(stat.get("port_speed_mbps", 0)) + + return output + + 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] = {} + + for vlan in vlans_data.get("vlan_element", []): + vlan_id = str(vlan.get("vlan_id", "")) + vlan_name = f"VLAN{vlan_id}" + # IPv4 from /vlans/{id}/ipv4 + ipv4_data = self.get(f"vlans/{vlan_id}/ipv4") + for entry in ipv4_data.get("ipv4_element", []): + addr = entry.get("address", "") + mask = entry.get("mask", "") + if not addr or addr == "0.0.0.0": + continue + try: + import netaddr + plen = netaddr.IPNetwork(f"{addr}/{mask}").prefixlen + except Exception: + plen = 24 + result.setdefault(vlan_name, {"ipv4": {}, "ipv6": {}}) + result[vlan_name]["ipv4"][addr] = {"prefix_length": plen} + + return {k: v for k, v in result.items() if v["ipv4"] or v["ipv6"]} + + def get_arp_table(self, vrf: str = "") -> list: + """Return NAPALM ARP table from the REST API.""" + arp_data = self.get("arp") + table = [] + for entry in arp_data.get("arp_entry", []): + table.append( + { + "interface": entry.get("port_id", ""), + "mac": entry.get("mac_address", ""), + "ip": entry.get("ip_address", ""), + "age": float(entry.get("age", -1)), + } + ) + return table + + def get_mac_address_table(self) -> list: + """Return NAPALM MAC address table from the REST API.""" + mac_data = self.get("mac-table") + table = [] + for entry in mac_data.get("mac_table_entry_element", []): + table.append( + { + "mac": entry.get("mac_address", ""), + "interface": entry.get("port_id", ""), + "vlan": int(entry.get("vlan_id", 0)), + "static": entry.get("mac_addr_type", "").lower() == "static", + "active": True, + "moves": -1, + "last_move": -1.0, + } + ) + return table + + def get_lldp_neighbors(self) -> Dict[str, list]: + """Return NAPALM LLDP neighbors from the REST API.""" + 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", "") + result.setdefault(local_port, []).append( + { + "hostname": nbr.get("system_name", ""), + "port": nbr.get("port_id", ""), + } + ) + return result + + def get_lldp_neighbors_detail(self) -> Dict[str, list]: + """Return NAPALM LLDP neighbor details from the REST API.""" + 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", "") + result.setdefault(local_port, []).append( + { + "parent_interface": "", + "remote_chassis_id": nbr.get("chassis_id", ""), + "remote_port": nbr.get("port_id", ""), + "remote_port_description": nbr.get("port_description", ""), + "remote_system_name": nbr.get("system_name", ""), + "remote_system_description": nbr.get("system_description", ""), + "remote_system_capab": [], + "remote_system_enable_capab": [], + } + ) + return result + + def get_config(self) -> Dict[str, str]: + """Return running and startup configuration via REST API.""" + running = self.get("running-config") + startup = self.get("startup-config") + return { + "running": running.get("config", ""), + "startup": startup.get("config", ""), + "candidate": "", + } + + def get_ntp_servers(self) -> Dict[str, Dict]: + """Return NTP servers from REST API.""" + data = self.get("ntp/server") + 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]: + """Return VLAN information with proper tagged/untagged separation. + + Fetches ``/vlans`` for VLAN names and ``/vlans-ports`` for port membership. + + The ``port_mode`` field in ``/vlans-ports`` varies by firmware: + + ======================== ======== + Firmware ``port_mode`` Meaning + ======================== ======== + ``POM_UNTAGGED`` Untagged (access/native) + ``POM_TAGGED_STATIC`` Tagged (trunk) + ``POM_TAGGED`` Tagged (some older v3 API) + ``POM_NATIVE_UNTAGGED`` Native untagged (also PVID) + ``POM_NATIVE_TAGGED`` Native tagged (unusual) + ======================== ======== + + Returns extended NAPALM-compatible dict:: + + { + 1: {"name": "DEFAULT_VLAN", + "interfaces": ["1","2","3"], + "tagged": [], + "untagged": ["1","2","3"]}, + 10: {"name": "MGMT", + "interfaces": ["1","2","3"], + "tagged": ["1","2","3"], + "untagged": []}, + } + """ + vlans_data = self.get("vlans") + ports_data = self.get("vlans-ports") + + # Build VLAN skeleton from /vlans + result: Dict[int, Dict] = {} + for vlan in vlans_data.get("vlan_element", []): + try: + vid = int(vlan["vlan_id"]) + except (KeyError, ValueError, TypeError): + continue + result[vid] = { + "name": vlan.get("name", f"VLAN{vid}"), + "interfaces": [], + "tagged": [], + "untagged": [], + } + + # Populate tagged/untagged from /vlans-ports + for entry in ports_data.get("vlan_port_element", []): + try: + vid = int(entry["vlan_id"]) + except (KeyError, ValueError, TypeError): + continue + + port = entry.get("port_id", "") + if not port: + continue + + mode = entry.get("port_mode", "") + + # Normalise to lower-case for comparison + mode_low = mode.lower() + + # Untagged variants: POM_UNTAGGED, POM_NATIVE_UNTAGGED, "Untagged" + is_untagged = any( + 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 + # Contains "tagged" but NOT "untagged" → tagged + if "untagged" in mode_low: + is_untagged = True + elif "tagged" in mode_low: + is_untagged = False + else: + # Unknown mode: skip + logger.debug("Unknown port_mode %r for port %s VLAN %s", mode, port, vid) + continue + + if vid not in result: + result[vid] = { + "name": f"VLAN{vid}", + "interfaces": [], + "tagged": [], + "untagged": [], + } + + if is_untagged: + result[vid]["untagged"].append(port) + else: + result[vid]["tagged"].append(port) + + # Build "interfaces" as ordered union (untagged first, then tagged) + for vdata in result.values(): + seen: list = [] + for p in vdata["untagged"] + vdata["tagged"]: + if p not in seen: + seen.append(p) + vdata["interfaces"] = seen + + return result + + def delete_vlan(self, vlan_id: int) -> None: + """Remove a VLAN via the REST API (``DELETE /vlans/{id}``). + + :param vlan_id: VLAN ID (1–4094) to delete. + :raises ValueError: If *vlan_id* is out of range. + :raises ConnectionException: If the API returns a non-success status. + """ + if not 1 <= vlan_id <= 4094: + raise ValueError(f"VLAN ID {vlan_id} is out of range (1–4094)") + + resp = self.delete(f"vlans/{vlan_id}") + if not resp.ok and resp.status_code != 404: + raise ConnectionException( + f"delete_vlan({vlan_id}): REST API returned HTTP {resp.status_code}" + ) + + def ping( + self, + destination: str, + source: str = "", + ttl: int = 255, + timeout: int = 2, + size: int = 100, + count: int = 5, + vrf: str = "", + ) -> Dict: + """Execute ping via REST API (if supported) or return not-implemented.""" + payload: Dict = {"destination": destination, "count": count, "timeout": timeout} + if source: + payload["source"] = source + + resp = self.post("ping", json=payload) + if not resp.ok: + return {"error": f"Ping API returned HTTP {resp.status_code}"} + + data = resp.json() + sent = data.get("sent", count) + received = data.get("received", 0) + rtt_min = float(data.get("min_rtt", 0)) + rtt_max = float(data.get("max_rtt", 0)) + rtt_avg = float(data.get("avg_rtt", 0)) + + results = [] + for _ in range(received): + results.append({"ip_address": destination, "rtt": rtt_avg}) + + return { + "success": { + "probes_sent": sent, + "packet_loss": sent - received, + "rtt_min": rtt_min, + "rtt_max": rtt_max, + "rtt_avg": rtt_avg, + "rtt_stddev": 0.0, + "results": results, + } + } diff --git a/napalm_procurve/parsers.py b/napalm_procurve/parsers.py new file mode 100644 index 0000000..1430e37 --- /dev/null +++ b/napalm_procurve/parsers.py @@ -0,0 +1,886 @@ +# -*- coding: utf-8 -*- +# Licensed under the Apache License, Version 2.0 + +"""CLI output parsers for HPE/Aruba ProCurve switches. + +All functions are pure (no side effects) and operate on pre-collected +command output strings. The same parsers are shared by the SSH and Telnet +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() + if len(clean) != 12: + return mac + return ":".join(clean[i : i + 2] for i in range(0, 12, 2)) + + +def _kv(output: str, key: str) -> str: + """Extract value from ``key : value`` style ProCurve output (case-insensitive).""" + for line in output.splitlines(): + if re.search(re.escape(key), line, re.IGNORECASE): + parts = line.split(":", 1) + if len(parts) == 2: + return parts[1].strip() + return "" + + +# --------------------------------------------------------------------------- +# show system-information +# --------------------------------------------------------------------------- + +def parse_system_info(output: str) -> Dict: + """Parse ``show system-information`` output. + + Returns a dict with keys: hostname, os_version, serial_number, model, + base_mac, uptime_seconds. + + Handles both older ProCurve style (two-column layout) and newer Aruba style. + + Example older output:: + + System Name : myswitch + Software revision : R.11.27 Base MAC Addr : a1b2c3-d4e5f6 + Serial Number : SGXXXXXXXXXX Up Time : 5 days 3:14 + + Example newer output:: + + System Name : ArubaSW + Firmware revision: YA.16.04.0006 + Serial Number : SGXXXXXXXXXX + """ + # Use a regex to extract key:value pairs from each line. + # ProCurve CLI may pack two k/v pairs on the same line with many spaces + # 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] = {} + for raw_line in output.splitlines(): + for m in _KV_RE.finditer(raw_line): + k = m.group(1).strip() + v = m.group(2).strip() + if k and v: + kv_map[k] = v + + def get(key: str) -> str: + for k, v in kv_map.items(): + if key.lower() in k.lower(): + return v + return "" + + hostname = get("System Name") + os_version = ( + get("Software revision") + or get("Firmware revision") + or get("Software Revision") + or get("Firmware Version") + ) + serial = get("Serial Number") or get("Serial Num") + base_mac = procurve_mac_to_std(get("Base MAC Addr") or get("MAC Address") or "") + model = get("System Model") or get("product_model") or "" + + # Parse uptime — ProCurve formats: "5 days 3:14" or "0 days 0:05" + # Also: "5 day(s), 3:14:00" + uptime_seconds = _parse_procurve_uptime(get("Up Time") or get("Uptime") or "") + + return { + "hostname": hostname, + "os_version": os_version, + "serial_number": serial, + "base_mac": base_mac, + "model": model, + "uptime_seconds": uptime_seconds, + } + + +def _parse_procurve_uptime(uptime_str: str) -> float: + """Convert ProCurve uptime string to total seconds. + + Handles formats:: + + 5 days 3:14 → 5*86400 + 3*3600 + 14*60 + 0 days 0:05:03 → 303 + 5 day(s), 3:14:00 → same + 1 hr 5 min 3 sec → some devices + """ + total = 0.0 + # Format: "N days H:MM" or "N days H:MM:SS" + m = re.search(r"(\d+)\s+days?\s+(\d+):(\d+)(?::(\d+))?", uptime_str, re.I) + if m: + total = ( + int(m.group(1)) * 86400 + + int(m.group(2)) * 3600 + + int(m.group(3)) * 60 + + int(m.group(4) or 0) + ) + return float(total) + + # Format: "Xd Xh Xm Xs" or "X day(s), X hour(s), ..." + days = re.search(r"(\d+)\s+day", uptime_str, re.I) + hours = re.search(r"(\d+)\s+h(?:r|our)", uptime_str, re.I) + minutes = re.search(r"(\d+)\s+min", uptime_str, re.I) + seconds = re.search(r"(\d+)\s+sec", uptime_str, re.I) + if days or hours or minutes or seconds: + total = ( + int(days.group(1) if days else 0) * 86400 + + int(hours.group(1) if hours else 0) * 3600 + + int(minutes.group(1) if minutes else 0) * 60 + + int(seconds.group(1) if seconds else 0) + ) + return float(total) + + +# --------------------------------------------------------------------------- +# show version +# --------------------------------------------------------------------------- + +def parse_version(output: str) -> str: + """Extract firmware version from ``show version`` or ``show system-information``. + + ProCurve ``show version`` example:: + + HP J9565A 2520G-8-PoE Switch + Software revision : R.11.27 + + Aruba image stamp format:: + + Image stamp: /ws/.../WC.16.11.0004 + WC.16.11.0004 + """ + # "Software revision : R.11.27" style + m = re.search(r"(?:software|firmware)\s+revision\s*:\s*(\S+)", output, re.I) + if m: + return m.group(1) + # Bare revision on its own line (image stamp format) + for line in output.splitlines(): + line = line.strip() + if re.match(r"^[A-Z]{2}\.\d+\.\d+", line): + return line + return "" + + +def parse_model_from_version(output: str) -> str: + """Extract model from ``show version`` first line. + + Example: ``HP J9565A 2520G-8-PoE Switch`` → ``2520G-8-PoE`` + """ + first = output.strip().splitlines()[0] if output.strip() else "" + m = re.search(r"HP\s+\S+\s+(\S+)\s+Switch", first, re.I) + if m: + return m.group(1) + # Aruba format: "Aruba 2530-8-PoE+ Switch" + m = re.search(r"(?:Aruba|HP)\s+(\d\S+)\s+(?:Switch|)", first, re.I) + if m: + return m.group(1) + return "" + + +# --------------------------------------------------------------------------- +# show interfaces brief / show port status +# --------------------------------------------------------------------------- + +# ProCurve "show interfaces brief" tabular header line reference: +# Port Type | Enabled Link MDI Mode FlwCtrl BCast +# ------ ------------ + ------- ------ ----- ----------- -------- ----- +# 1 100/1000T | Yes Down Auto Unknown off 0 + +_INTF_BRIEF_RE = re.compile( + r"^\s*(\S+)\s+" # Port + r"(\S+)\s+\|" # Type | + r"\s+(Yes|No)\s+" # Enabled + r"(Up|Down)\s+" # Link + r"\S+\s+" # MDI (ignored) + r"(\S+)", # Mode (speed/duplex) + re.IGNORECASE, +) + + +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] = {} + for line in output.splitlines(): + m = _INTF_BRIEF_RE.match(line) + if not m: + continue + port, itype, enabled, link, mode = ( + m.group(1), m.group(2), m.group(3), m.group(4), m.group(5) + ) + speed = 0.0 + duplex = "" + # Mode examples: "1000FDx", "100HDx", "Unknown", "Auto" + sm = re.match(r"(\d+)(FDx|HDx)?", mode, re.I) + if sm: + speed = float(sm.group(1)) + duplex = "full" if (sm.group(2) or "").lower() == "fdx" else "half" + + interfaces[port] = { + "is_up": link.lower() == "up", + "is_enabled": enabled.lower() == "yes", + "speed": speed, + "duplex": duplex, + "description": "", + "last_flapped": -1.0, + "mtu": -1, + "mac_address": "", + } + return interfaces + + +# --------------------------------------------------------------------------- +# show interfaces (per-port detail for MAC / description) +# --------------------------------------------------------------------------- + +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") + return {"mac_address": mac, "description": name} + + +# --------------------------------------------------------------------------- +# show ip / show ip interface +# --------------------------------------------------------------------------- + +def parse_ip_interfaces(output: str) -> Dict[str, Dict]: + """Parse ``show ip`` or ``show ip interface brief`` output. + + Returns NAPALM-compatible dict:: + + {"VLAN1": {"ipv4": {"192.168.0.1": {"prefix_length": 24}}}} + + ProCurve ``show ip`` multi-VLAN format:: + + IP Configuration for VLAN 1: + IP Config : Manual + IP Address : 192.168.0.1 + Subnet Mask : 255.255.255.0 + + IP Configuration for VLAN 99: + ... + """ + result: Dict[str, Dict] = {} + current_vlan = "" + current_ip = "" + + for line in output.splitlines(): + # Detect VLAN section header: "IP Configuration for VLAN X" or "... VLAN 1:" + vm = re.search(r"IP\s+Config(?:uration)?\s+for\s+(?:VLAN\s+)?(\S+)", line, re.I) + if vm: + raw_id = vm.group(1).rstrip(":") + # Strip a leading "VLAN" prefix if the regex captured it (e.g. "VLAN1"→"1") + num = re.sub(r"^VLAN", "", raw_id, flags=re.I) + current_vlan = f"VLAN{num}" + result.setdefault(current_vlan, {"ipv4": {}, "ipv6": {}}) + continue + + # Single VLAN output (no header): detect from "IP Address" line + if re.search(r"IP\s+Address\s*:", line, re.I): + ip = line.split(":", 1)[-1].strip() + if ip and ip not in ("0.0.0.0", ""): + current_ip = ip + if not current_vlan: + current_vlan = "VLAN1" + result.setdefault(current_vlan, {"ipv4": {}, "ipv6": {}}) + continue + + if re.search(r"Subnet\s+Mask\s*:", line, re.I) and current_ip: + mask = line.split(":", 1)[-1].strip() + try: + import netaddr + prefix_len = netaddr.IPNetwork(f"{current_ip}/{mask}").prefixlen + except Exception: + prefix_len = 24 + if current_vlan: + result[current_vlan]["ipv4"][current_ip] = {"prefix_length": prefix_len} + current_ip = "" + continue + + # Clean up empty entries + return {k: v for k, v in result.items() if v["ipv4"] or v["ipv6"]} + + +# --------------------------------------------------------------------------- +# show arp +# --------------------------------------------------------------------------- + +_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) +) + + +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 "" + ) + table.append( + { + "interface": port, + "mac": procurve_mac_to_std(mac), + "ip": ip, + "age": -1.0, + } + ) + return table + + +# --------------------------------------------------------------------------- +# show mac-address +# --------------------------------------------------------------------------- + +_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 +) + + +def parse_mac_table(output: str) -> List[Dict]: + """Parse ``show mac-address`` into NAPALM MAC table format.""" + table = [] + for line in output.splitlines(): + m = _MAC_RE.search(line) + if not m: + continue + mac, port, vlan, entry_type = m.group(1), m.group(2), m.group(3), m.group(4) + table.append( + { + "mac": procurve_mac_to_std(mac), + "interface": port, + "vlan": int(vlan), + "static": entry_type.lower() in ("static", "permanent"), + "active": True, + "moves": -1, + "last_move": -1.0, + } + ) + return table + + +# --------------------------------------------------------------------------- +# show lldp info remote-device +# --------------------------------------------------------------------------- + +def parse_lldp_neighbors(output: str) -> Dict[str, List[Dict]]: + """Parse ``show lldp info remote-device`` into NAPALM neighbors dict. + + Output format:: + + LocalPort | ChassisId PortId SysName ... + --------- + ------------------- ------------ ------------ ... + 1 | a1 b2 c3 d4 e5 f6 Gi 0/1 otherswitch ... + """ + neighbors: Dict[str, List[Dict]] = {} + in_table = False + + for line in output.splitlines(): + # Skip header / separator lines + if re.match(r"\s*-+\s*\+", line) or re.match(r"\s*LocalPort", line, re.I): + in_table = True + continue + if not in_table: + continue + if not line.strip(): + continue + + # Format: " 1 | aa bb cc ... Gi0/1 hostname ..." + m = re.match(r"\s*(\S+)\s*\|\s*(.+)", line) + if not m: + continue + local_port = m.group(1) + rest = m.group(2).strip() + + # ChassisId: may be space-separated octets (aa bb cc dd ee ff) or MAC + # Then PortId, then SysName + # Split by 2+ spaces to separate columns + cols = re.split(r"\s{2,}", rest) + if len(cols) < 3: + continue + chassis_raw = cols[0].strip() + port_id = cols[1].strip() + sys_name = cols[2].strip() + + # Normalise chassis ID: space-separated octets → colon MAC + chassis = re.sub(r"\s+", "", chassis_raw) + 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} + ) + + return neighbors + + +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]] = {} + current_port = "" + entry: Dict = {} + + def _save(): + if current_port and entry: + neighbors.setdefault(current_port, []).append(entry.copy()) + + for line in output.splitlines(): + # New port section: "LLDP Remote Device Information Detail for Port N" + m = re.search(r"for\s+[Pp]ort\s+(\S+)", line) + if m: + _save() + current_port = m.group(1) + entry = { + "parent_interface": "", + "remote_chassis_id": "", + "remote_port": "", + "remote_port_description": "", + "remote_system_name": "", + "remote_system_description": "", + "remote_system_capab": [], + "remote_system_enable_capab": [], + } + continue + + if not entry: + continue + + 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 + elif "port id" in low and "descr" not in low: + entry["remote_port"] = line.split(":", 1)[-1].strip() + elif "port description" in low: + entry["remote_port_description"] = line.split(":", 1)[-1].strip() + elif "system name" in low: + entry["remote_system_name"] = line.split(":", 1)[-1].strip() + 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() + ) + 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() + ) + + _save() + return neighbors + + +def _parse_capabilities(cap_str: str) -> List[str]: + """Convert ProCurve capability abbreviations to NAPALM strings.""" + _MAP = { + "B": "bridge", + "R": "router", + "T": "telephone", + "W": "wlan-access-point", + "H": "host", + "D": "docsis-cable-device", + "O": "other", + "S": "station", + "C": "customer-vlan-bridge", + } + result = [] + for token in re.split(r"[,\s]+", cap_str): + token = token.strip() + if not token: + continue + result.append(_MAP.get(token, token.lower())) + return result + + +# --------------------------------------------------------------------------- +# show ntp associations / show timesync +# --------------------------------------------------------------------------- + +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] = {} + 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) + if m: + ip = m.group(1) + # Skip gateway / default-gateway entries + if "gateway" in line.lower() or "default" in line.lower(): + continue + servers[ip] = {} + return servers + + +# --------------------------------------------------------------------------- +# show environment +# --------------------------------------------------------------------------- + +def parse_environment(output: str) -> Dict: + """Parse ``show environment`` output into NAPALM environment dict.""" + env: Dict = { + "fans": {}, + "temperature": {}, + "power": {}, + "cpu": [{"%usage": 0.0}], + "memory": {"available_ram": -1, "used_ram": -1}, + } + + # CPU utilisation from show system-information + m = re.search(r"CPU\s+Util(?:ization)?\s*[(%:]+\s*(\d+)", output, re.I) + if m: + env["cpu"][0]["%usage"] = float(m.group(1)) + + # Memory + total_m = re.search(r"Memory\s*[-–]\s*Total\s*[:\s]+(\d+)", output, re.I) + free_m = re.search(r"Free\s*[:\s]+(\d+)", output, re.I) + if total_m: + env["memory"]["available_ram"] = int(total_m.group(1)) + if free_m: + env["memory"]["used_ram"] = ( + env["memory"]["available_ram"] - int(free_m.group(1)) + if env["memory"]["available_ram"] > 0 + else -1 + ) + + # Temperature sensors: "Sensor 1: 38C" or "Temperature: 38C" + for m in re.finditer( + r"(?:Sensor\s+(\d+)|Temperature(?:\s+Sensor)?)\s*[:\s]+(\d+)\s*[Cc°]", + output, + ): + sensor_id = f"sensor_{m.group(1) or '1'}" + temp = float(m.group(2)) + env["temperature"][sensor_id] = { + "temperature": temp, + "is_alert": False, + "is_critical": False, + } + + # Fan status + 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")} + + return env + + +# --------------------------------------------------------------------------- +# show local-users +# --------------------------------------------------------------------------- + +def parse_users(output: str) -> Dict[str, Dict]: + """Parse ``show local-users`` or ``show users`` output. + + Returns NAPALM users dict. + """ + users: Dict[str, Dict] = {} + for line in output.splitlines(): + # " manager Operator Enabled ..." or " manager Manager ..." + m = re.match( + r"\s+(\w+)\s+(\w+(?:\s+\w+)?)\s+(Enabled|Disabled)", + line, + re.I, + ) + if not m: + continue + uname, level, _state = m.group(1), m.group(2).lower(), m.group(3) + # Map privilege level + if "manager" in level or "admin" in level: + priv = 15 + elif "operator" in level: + priv = 5 + else: + priv = 1 + users[uname] = { + "level": priv, + "password": "", + "sshkeys": [], + } + return users + + +# --------------------------------------------------------------------------- +# show snmp-server / show snmp +# --------------------------------------------------------------------------- + +def parse_snmp_info(output: str) -> Dict: + """Parse ``show snmp-server`` output into NAPALM snmp_information 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 = "" + + for line in output.splitlines(): + # " community_name Manager Restricted" (read only) + # " community_name Operator Unrestricted" (read write) + m = re.match(r"\s+(\S+)\s+(Manager|Operator)\s+(Restricted|Unrestricted)", line, re.I) + if m: + comm, _role, access = m.group(1), m.group(2), m.group(3) + communities[comm] = { + "acl": "", + "mode": "ro" if access.lower() == "restricted" else "rw", + } + + return { + "chassis_id": chassis_id, + "community": communities, + "contact": contact, + "location": location, + } + + +# --------------------------------------------------------------------------- +# VLAN helpers +# --------------------------------------------------------------------------- + +def expand_procurve_ports(port_str: str) -> List[str]: + """Expand a ProCurve port-list string into a flat list of port names. + + Handles: + - Numeric ranges: ``1-4,6,8`` → ``["1","2","3","4","6","8"]`` + - Alphanumeric ranges: ``A1-A8`` → ``["A1","A2",..."A8"]`` + - Trunk ports: ``Trk1,Trk2`` → ``["Trk1","Trk2"]`` + - Mixed: ``1-3,Trk1,A1`` → ``["1","2","3","Trk1","A1"]`` + """ + ports: List[str] = [] + for token in port_str.split(","): + token = token.strip() + if not token: + continue + + # Try alphanumeric range: e.g. "A1-A8" (prefix + digits) + alpha_range = re.match(r"^([A-Za-z]+)(\d+)-([A-Za-z]+)(\d+)$", token) + if alpha_range: + prefix = alpha_range.group(1) + start = int(alpha_range.group(2)) + end = int(alpha_range.group(4)) + ports.extend(f"{prefix}{i}" for i in range(start, end + 1)) + continue + + # Try pure numeric range: e.g. "1-4" + num_range = re.match(r"^(\d+)-(\d+)$", token) + if num_range: + start = int(num_range.group(1)) + end = int(num_range.group(2)) + ports.extend(str(i) for i in range(start, end + 1)) + continue + + # Single port (numeric, Trk, alphanumeric stacking port etc.) + ports.append(token) + + return ports + + +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. + + Typical running-config VLAN block:: + + vlan 1 + name "DEFAULT_VLAN" + untagged 1-4,6-8 + ip address 192.168.0.1 255.255.255.0 + exit + + vlan 10 + name "MGMT" + tagged 1,2,5 + exit + + Returns extended NAPALM-compatible dict:: + + { + 1: {"name": "DEFAULT_VLAN", + "interfaces": ["1","2","3","4","6","7","8","5"], + "tagged": [], + "untagged": ["1","2","3","4","6","7","8"]}, + 10: {"name": "MGMT", + "interfaces": ["1","2","5"], + "tagged": ["1","2","5"], + "untagged": []}, + } + """ + vlans: Dict[int, Dict] = {} + current_id: Optional[int] = None + + for line in config_text.splitlines(): + stripped = line.strip() + + # Start of a vlan block: "vlan " + m = re.match(r"^vlan\s+(\d+)\s*$", stripped, re.I) + if m: + current_id = int(m.group(1)) + vlans[current_id] = {"name": f"VLAN{current_id}", "tagged": [], "untagged": []} + continue + + if current_id is None: + continue + + # End of vlan block + if stripped.lower() == "exit": + current_id = None + continue + + # name "VLAN-Name" (with or without quotes) + nm = re.match(r'^name\s+"?([^"]+)"?\s*$', stripped, re.I) + if nm: + vlans[current_id]["name"] = nm.group(1).strip() + continue + + # untagged + utm = re.match(r"^untagged\s+(.+)$", stripped, re.I) + if utm: + vlans[current_id]["untagged"] = expand_procurve_ports(utm.group(1)) + continue + + # tagged + tgm = re.match(r"^tagged\s+(.+)$", stripped, re.I) + if tgm: + vlans[current_id]["tagged"] = expand_procurve_ports(tgm.group(1)) + continue + + # Populate "interfaces" as union of tagged + untagged (NAPALM standard key) + for vdata in vlans.values(): + seen: List[str] = [] + for p in vdata["untagged"] + vdata["tagged"]: + if p not in seen: + seen.append(p) + vdata["interfaces"] = seen + + return vlans + + +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. + + Example output:: + + VLAN ID Name | Status Voice Jumbo + ------- ---------------- + ---------- ----- ----- + 1 DEFAULT_VLAN | Port-based No No + 10 MANAGEMENT | Port-based No No + + Returns ``{vlan_id: name}`` dict. + """ + vlans: Dict[int, str] = {} + in_table = False + + for line in output.splitlines(): + if re.match(r"\s*-{4,}", line): + in_table = True + continue + if not in_table: + continue + m = re.match(r"\s*(\d+)\s+(\S+)", line) + if m: + vlans[int(m.group(1))] = m.group(2) + + return vlans + + +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. + + Example output:: + + Port Information Mode Unknown VLAN Status + ---------------- -------- ------------ ------- + 1 Untagged Learn Up + 2 Tagged Learn Up + Trk1 Tagged Learn Up + """ + tagged: List[str] = [] + untagged: List[str] = [] + in_table = False + + for line in output.splitlines(): + if re.match(r"\s*-{4,}", line): + in_table = True + continue + if not in_table: + continue + # Columns: Port | Mode | Unknown VLAN | Status + m = re.match(r"\s*(\S+)\s+(Tagged|Untagged)\s+", line, re.I) + if m: + port, mode = m.group(1), m.group(2).lower() + if mode == "tagged": + tagged.append(port) + else: + untagged.append(port) + + return {"tagged": tagged, "untagged": untagged} + + +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. + A port that only appears in ``tagged`` lists has no PVID (trunk-only, PVID = 1 by default). + + Returns ``{"port": pvid_int, ...}`` mapping. + + Example:: + + # Port 1 is untagged in VLAN 1 → PVID 1 + # Port 5 only tagged in VLAN 10 → PVID 1 (default) + derive_port_pvids({ + 1: {"untagged": ["1","2","3"], "tagged": []}, + 10: {"untagged": [], "tagged": ["5"]}, + }) + # → {"1": 1, "2": 1, "3": 1, "5": 1} + """ + pvids: Dict[str, int] = {} + + # First pass: assign PVID from untagged membership + for vid, vdata in sorted(vlans.items()): + for port in vdata.get("untagged", []): + pvids[port] = vid + + # Second pass: ports that only appear in tagged lists get PVID 1 (default) + for vdata in vlans.values(): + for port in vdata.get("tagged", []): + if port not in pvids: + pvids[port] = 1 + + return pvids diff --git a/napalm_procurve/procurve.py b/napalm_procurve/procurve.py new file mode 100644 index 0000000..8db1977 --- /dev/null +++ b/napalm_procurve/procurve.py @@ -0,0 +1,1066 @@ +# -*- coding: utf-8 -*- +# Licensed under the Apache License, Version 2.0 + +"""NAPALM driver for HPE/Aruba ProCurve switches. + +Supports legacy ProCurve (2520G-8-PoE etc.) through to modern Aruba 2530/2540 +by auto-detecting the best available transport: + + 1. REST API (HTTPS then HTTP, API v7 → v6 → v3) — newer Aruba 2530/2540 + 2. SSH — standard netmiko hp_procurve + 3. SSH legacy KEX — forces older algorithms for 2520G etc. + 4. Telnet — oldest devices + +A specific transport can be forced with ``optional_args={"transport": "ssh"}``. + +Netmiko device type: ``hp_procurve`` / ``hp_procurve_telnet`` +""" + +import logging +import re +import socket +from typing import Any, Dict, List, Optional, Union + +import netaddr +from netmiko import ConnectHandler +from netmiko.exceptions import ( + NetmikoAuthenticationException, + NetmikoTimeoutException, +) +from napalm_device_types import SwitchDriver +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_procurve.api_client import ProcurveApiClient +from napalm_procurve import parsers + +logger = logging.getLogger("napalm_procurve") + +# --------------------------------------------------------------------------- +# KEX / cipher profiles for legacy SSH +# --------------------------------------------------------------------------- + +# Standard: netmiko's HPProcurveBase already disables rsa-sha2-* pubkeys which +# is needed for any ProCurve running SSH (they use plain ssh-rsa host keys). +_SSH_DISABLED_STANDARD = { + "pubkeys": ["rsa-sha2-256", "rsa-sha2-512"], +} + +# Legacy: additionally remove modern kex algorithms so the negotiation is +# forced down to diffie-hellman-group14-sha1 / diffie-hellman-group1-sha1, +# which very old switches (2520G etc.) may require. +_SSH_DISABLED_LEGACY = { + "pubkeys": ["rsa-sha2-256", "rsa-sha2-512"], + "kex": [ + "curve25519-sha256@libssh.org", + "ecdh-sha2-nistp256", + "ecdh-sha2-nistp384", + "ecdh-sha2-nistp521", + "diffie-hellman-group16-sha512", + "diffie-hellman-group-exchange-sha256", + "diffie-hellman-group14-sha256", + ], +} + + +class ProcurveDriver(SwitchDriver): + """NAPALM driver for HPE/Aruba ProCurve switches. + + Auto-detects transport on ``open()``. See module docstring for details. + """ + + VENDOR = "HPE" + NETMIKO_DEVICE_TYPE = "hp_procurve" + NETMIKO_DEVICE_TYPE_TELNET = "hp_procurve_telnet" + + def __init__( + self, + hostname: str, + username: str, + password: str, + timeout: int = 60, + optional_args: Optional[Dict] = None, + ) -> None: + self.hostname = hostname + self.username = username + self.password = password + self.timeout = timeout + + if optional_args is None: + optional_args = {} + + # --- 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.ssl_verify: bool = optional_args.get("ssl_verify", False) + self.api_version_hint: Optional[str] = optional_args.get("api_version") + # Enable password for CLI (defaults to login password) + self._secret: str = optional_args.get("secret", password) + + if optional_args.get("debugging"): + logger.setLevel(logging.DEBUG) + + self.netmiko_optional_args = netmiko_args(optional_args) + + # --- runtime state --- + # Active transport: "api" | "ssh" | "ssh_legacy" | "telnet" + self._transport: Optional[str] = None + # CLI backend (SSH or Telnet) + self._device: Optional[ConnectHandler] = None + # REST API backend + self._api: Optional[ProcurveApiClient] = None + + # 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 + + # ------------------------------------------------------------------ + # Connection management + # ------------------------------------------------------------------ + + def open(self) -> None: + """Open connection, auto-detecting the best available transport. + + Detection order (unless overridden by ``optional_args["transport"]``): + + 1. REST API + 2. SSH (standard) + 3. SSH (legacy KEX — old ProCurve like 2520G-8-PoE) + 4. Telnet + """ + forced = self.force_transport + tried: List[str] = [] + + # --- 1. REST API --- + if not forced or forced == "api": + if self._try_api(): + return + tried.append("api") + + # --- 2. SSH standard --- + if not forced or forced == "ssh": + if self._try_ssh(legacy=False): + return + tried.append("ssh") + + # --- 3. SSH legacy KEX --- + if not forced or forced == "ssh_legacy": + if self._try_ssh(legacy=True): + return + tried.append("ssh_legacy") + + # --- 4. Telnet --- + if not forced or forced == "telnet": + if self._try_telnet(): + return + tried.append("telnet") + + raise ConnectionException( + f"Cannot connect to {self.hostname}. " + f"Tried transports: {', '.join(tried)}. " + "Check connectivity, credentials and whether SSH/Telnet/API is enabled." + ) + + def close(self) -> None: + """Close the active connection.""" + if self._api: + self._api.disconnect() + self._api = None + if self._device: + try: + self._device.disconnect() + except Exception: + pass + self._device = None + self._transport = None + + 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): + return {"is_alive": False} + return {"is_alive": False} + + # ------------------------------------------------------------------ + # Transport probes + # ------------------------------------------------------------------ + + def _try_api(self) -> bool: + """Probe and connect via REST API. Returns True on success.""" + logger.debug("Trying REST API for %s", self.hostname) + ver, proto = ProcurveApiClient.probe( + self.hostname, timeout=5, ssl_verify=self.ssl_verify + ) + # Allow hint override (e.g. user knows the API version) + if self.api_version_hint and proto: + ver = self.api_version_hint + + if not ver: + logger.debug("REST API not available on %s", self.hostname) + return False + + client = ProcurveApiClient( + hostname=self.hostname, + username=self.username, + password=self.password, + timeout=self.timeout, + ssl_verify=self.ssl_verify, + api_version=ver, + ) + client.setup(ver, proto) + try: + client.connect() + except Exception as exc: + logger.debug("REST API connect failed: %s", exc) + return False + + self._api = client + self._transport = "api" + logger.info("Connected to %s via REST API (%s %s)", self.hostname, proto, ver) + return True + + def _try_ssh(self, legacy: bool = False) -> bool: + """Probe and connect via SSH. Returns True on success.""" + label = "SSH-legacy" if legacy else "SSH" + logger.debug("Trying %s for %s", label, self.hostname) + disabled = _SSH_DISABLED_LEGACY if legacy else _SSH_DISABLED_STANDARD + try: + conn = ConnectHandler( + device_type=self.NETMIKO_DEVICE_TYPE, + host=self.hostname, + username=self.username, + password=self.password, + secret=self._secret, + port=self.port, + timeout=self.timeout, + disabled_algorithms=disabled, + **self.netmiko_optional_args, + ) + self._device = conn + self._transport = "ssh_legacy" if legacy else "ssh" + logger.info("Connected to %s via %s", self.hostname, label) + return True + except NetmikoAuthenticationException as exc: + raise ConnectionException( + f"Authentication failed for {self.hostname}: {exc}" + ) from exc + except NetmikoTimeoutException: + logger.debug("%s timeout for %s", label, self.hostname) + return False + except Exception as exc: + logger.debug("%s failed for %s: %s", label, self.hostname, exc) + return False + + def _try_telnet(self) -> bool: + """Probe and connect via Telnet. Returns True on success.""" + logger.debug("Trying Telnet for %s", self.hostname) + try: + conn = ConnectHandler( + device_type=self.NETMIKO_DEVICE_TYPE_TELNET, + host=self.hostname, + username=self.username, + password=self.password, + secret=self._secret, + port=self.telnet_port, + timeout=self.timeout, + ) + self._device = conn + self._transport = "telnet" + logger.info("Connected to %s via Telnet", self.hostname) + return True + except NetmikoAuthenticationException as exc: + raise ConnectionException( + f"Authentication failed for {self.hostname}: {exc}" + ) from exc + except Exception as exc: + logger.debug("Telnet failed for %s: %s", self.hostname, exc) + return False + + # ------------------------------------------------------------------ + # CLI helpers + # ------------------------------------------------------------------ + + def _send_command(self, command: Union[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 + non-error result is returned. + """ + if self._device is None: + raise ConnectionClosedException("Device connection is not open.") + + prompt_pattern = rf"{re.escape(self._device.base_prompt)}[>#]" + + def _do(cmd: str) -> str: + return self._device.send_command( + cmd, + expect_string=prompt_pattern, + read_timeout=self.timeout, + ).strip() + + try: + if isinstance(command, list): + last = "" + for cmd in command: + last = _do(cmd) + if "% Invalid" not in last and "Error" not in last: + return last + return last + return _do(command) + except (socket.error, EOFError) as exc: + raise ConnectionClosedException(str(exc)) from exc + + def _exec_prompt(self) -> str: + return rf"{re.escape(self._device.base_prompt)}[>#]" + + def _conf_prompt(self) -> str: + return rf"{re.escape(self._device.base_prompt)}\(config[^)]*\)[>#]" + + def _any_prompt(self) -> str: + return rf"{re.escape(self._device.base_prompt)}(?:\([^)]*\))?[>#]" + + def _enter_config_mode(self) -> None: + self._device.send_command( + "configure", + expect_string=self._conf_prompt(), + read_timeout=self.timeout, + ) + + def _exit_config_mode(self) -> None: + self._device.send_command( + "end", + expect_string=self._exec_prompt(), + read_timeout=self.timeout, + ) + + def _save_config(self) -> None: + """Save running configuration to startup (``write memory``).""" + self._device.send_command( + "write memory", + expect_string=self._exec_prompt(), + read_timeout=self.timeout, + ) + + 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] = [] + for line in config_text.splitlines(): + stripped = line.strip() + if not stripped or stripped.startswith(("!", "#", ";")): + continue + out = self._device.send_command( + stripped, + expect_string=ep, + read_timeout=self.timeout, + ).strip() + if out and re.search(r"(error|invalid|unknown|unrecog)", out, re.I): + errors.append(f" {stripped!r}: {out}") + return errors + + # ------------------------------------------------------------------ + # NAPALM: get_facts + # ------------------------------------------------------------------ + + def get_facts(self) -> Dict: + """Return general device information.""" + if self._transport == "api": + return self._api.get_facts() + + # CLI path + sysinfo = self._send_command( + ["show system-information", "show system information"] + ) + parsed = parsers.parse_system_info(sysinfo) + + # get model from show version if not in system-info + if not parsed["model"]: + ver_out = self._send_command("show version") + parsed["model"] = parsers.parse_model_from_version(ver_out) + if not parsed["os_version"]: + parsed["os_version"] = parsers.parse_version(ver_out) + + iface_list = self._get_interface_list() + + return { + "vendor": self.VENDOR, + "model": parsed["model"], + "hostname": parsed["hostname"], + "fqdn": parsed["hostname"], + "os_version": parsed["os_version"], + "serial_number": parsed["serial_number"], + "uptime": parsed["uptime_seconds"], + "interface_list": iface_list, + } + + 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) + ], + ) + + # ------------------------------------------------------------------ + # NAPALM: get_interfaces + # ------------------------------------------------------------------ + + def get_interfaces(self) -> Dict[str, Dict]: + """Return a dictionary of interface details.""" + if self._transport == "api": + return self._api.get_interfaces() + + out = self._send_command("show interfaces brief") + ifaces = parsers.parse_interfaces_brief(out) + + # Enrich with MAC address from per-port detail (best-effort) + for port in list(ifaces.keys()): + try: + detail_out = self._send_command(f"show interfaces {port}") + detail = parsers.parse_interface_detail(detail_out, port) + ifaces[port]["mac_address"] = detail.get("mac_address", "") + if detail.get("description"): + ifaces[port]["description"] = detail["description"] + except Exception: + pass + + return ifaces + + # ------------------------------------------------------------------ + # NAPALM: get_interfaces_ip + # ------------------------------------------------------------------ + + def get_interfaces_ip(self) -> Dict[str, Dict]: + """Return IP addresses assigned to interfaces.""" + if self._transport == "api": + return self._api.get_interfaces_ip() + + out = self._send_command("show ip") + return parsers.parse_ip_interfaces(out) + + # ------------------------------------------------------------------ + # NAPALM: get_arp_table + # ------------------------------------------------------------------ + + def get_arp_table(self, vrf: str = "") -> List[Dict]: + """Return the ARP table.""" + if self._transport == "api": + return self._api.get_arp_table(vrf) + + out = self._send_command("show arp") + return parsers.parse_arp_table(out) + + # ------------------------------------------------------------------ + # NAPALM: get_mac_address_table + # ------------------------------------------------------------------ + + def get_mac_address_table(self) -> List[Dict]: + """Return the MAC address table.""" + if self._transport == "api": + return self._api.get_mac_address_table() + + out = self._send_command(["show mac-address", "show mac address"]) + return parsers.parse_mac_table(out) + + # ------------------------------------------------------------------ + # NAPALM: get_lldp_neighbors + # ------------------------------------------------------------------ + + 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"] + ) + return parsers.parse_lldp_neighbors(out) + + # ------------------------------------------------------------------ + # NAPALM: get_lldp_neighbors_detail + # ------------------------------------------------------------------ + + 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() + + if interface: + cmd = f"show lldp info remote-device {interface}" + else: + cmd = "show lldp info remote-device detail" + out = self._send_command(cmd) + return parsers.parse_lldp_neighbors_detail(out) + + # ------------------------------------------------------------------ + # NAPALM: get_config + # ------------------------------------------------------------------ + + def get_config( + self, retrieve: str = "all", full: bool = False, sanitized: bool = False + ) -> Dict[str, str]: + """Return device configuration.""" + if self._transport == "api": + return self._api.get_config() + + running = "" + startup = "" + candidate = self._candidate_config or "" + + 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"] + ) + + return {"running": running, "startup": startup, "candidate": candidate} + + # ------------------------------------------------------------------ + # NAPALM: get_environment + # ------------------------------------------------------------------ + + 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"] + ) + env_out = self._send_command(["show environment", "show system"]) + combined = sysinfo + "\n" + env_out + return parsers.parse_environment(combined) + + # ------------------------------------------------------------------ + # NAPALM: get_ntp_servers + # ------------------------------------------------------------------ + + 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"] + ) + return parsers.parse_ntp_servers(out) + + # ------------------------------------------------------------------ + # NAPALM: get_ntp_stats + # ------------------------------------------------------------------ + + def get_ntp_stats(self) -> List[Dict]: + """Return NTP peer statistics.""" + out = self._send_command( + ["show ntp associations", "show ntp status"] + ) + stats = [] + for line in out.splitlines(): + m = re.match( + r"\s*([*+]?)\s*(\d{1,3}(?:\.\d{1,3}){3})" + r"\s+(\S+)\s+(\d+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)", + line, + ) + if not m: + continue + synced = m.group(1) == "*" + stats.append( + { + "remote": m.group(2), + "referenceid": m.group(3), + "synchronized": synced, + "stratum": int(m.group(4)), + "type": "-", + "when": m.group(5), + "hostpoll": int(m.group(6)) if m.group(6).isdigit() else -1, + "reachability": int(m.group(7), 8) if m.group(7).isdigit() else -1, + "delay": float(m.group(8)) if _is_float(m.group(8)) else 0.0, + "offset": float(m.group(9)) if _is_float(m.group(9)) else 0.0, + "jitter": 0.0, + } + ) + return stats + + # ------------------------------------------------------------------ + # NAPALM: get_users + # ------------------------------------------------------------------ + + def get_users(self) -> Dict[str, Dict]: + """Return local users.""" + out = self._send_command(["show local-users", "show users"]) + return parsers.parse_users(out) + + # ------------------------------------------------------------------ + # NAPALM: get_snmp_information + # ------------------------------------------------------------------ + + def get_snmp_information(self) -> Dict: + """Return SNMP configuration.""" + out = self._send_command(["show snmp-server", "show snmp"]) + return parsers.parse_snmp_info(out) + + # ------------------------------------------------------------------ + # NAPALM: get_vlans (extended: tagged + untagged per VLAN) + # ------------------------------------------------------------------ + + 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 + two extra keys per VLAN: + + - ``tagged`` — ports where this VLAN is carried tagged (802.1Q trunk) + - ``untagged`` — ports where this VLAN is the native/access VLAN (PVID) + + The standard ``interfaces`` key remains the union of both lists, + keeping backward compatibility with NAPALM tooling. + + Example return:: + + { + 1: { + "name": "DEFAULT_VLAN", + "interfaces": ["1", "2", "3", "4", "5"], + "tagged": ["5"], + "untagged": ["1", "2", "3", "4"], + }, + 10: { + "name": "MGMT", + "interfaces": ["1", "2", "5"], + "tagged": ["1", "2", "5"], + "untagged": [], + }, + } + + Use :meth:`get_port_pvids` to get the per-port PVID mapping derived + from the ``untagged`` lists. + """ + if self._transport == "api": + return self._api.get_vlans() + + # --- CLI path --- + # Primary: parse VLAN blocks from running-config (single command, + # reliable across all ProCurve firmware, contains untagged/tagged lines) + running = self._send_command("show running-config") + vlans = parsers.parse_vlans_from_running_config(running) + + if vlans: + return vlans + + # 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]: + """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] = {} + 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] = [] + for p in untagged + tagged: + if p not in interfaces: + interfaces.append(p) + result[vid] = { + "name": name, + "interfaces": interfaces, + "tagged": tagged, + "untagged": untagged, + } + + return result + + # ------------------------------------------------------------------ + # Custom: get_port_pvids (PVID per port, derived from get_vlans) + # ------------------------------------------------------------------ + + 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 + ID is the port's PVID. Ports that only appear in tagged lists have + PVID 1 (the default). + + This method calls :meth:`get_vlans` internally and derives PVIDs from + the ``untagged`` lists. + + Example return:: + + { + "1": 1, # port 1 is untagged in VLAN 1 → PVID 1 + "2": 1, + "3": 10, # port 3 is untagged in VLAN 10 → PVID 10 + "Trk1": 1, # Trk1 only tagged → PVID 1 (default) + } + """ + vlans = self.get_vlans() + return parsers.derive_port_pvids(vlans) + + # ------------------------------------------------------------------ + # NAPALM: ping + # ------------------------------------------------------------------ + + # ------------------------------------------------------------------ + # SwitchDriver: write methods + # ------------------------------------------------------------------ + + def set_vlan(self, vlan_id: int, config: VlanConfigDict) -> None: + """Create or update a VLAN on the switch.""" + if self._transport == "api": + self._api_set_vlan(vlan_id, config) + else: + self._cli_set_vlan(vlan_id, config) + + def _api_set_vlan(self, vlan_id: int, config: VlanConfigDict) -> None: + name = config.get("name", f"VLAN{vlan_id}") + payload = {"vlan_id": vlan_id, "name": name} + resp = self._api.post("vlans", json=payload) + if not resp.ok: + if resp.status_code in (400, 409): + # VLAN already exists — update name via PUT + resp2 = self._api.put(f"vlans/{vlan_id}", json={"name": name}) + if not resp2.ok: + raise ConnectionException( + f"set_vlan({vlan_id}): API PUT returned HTTP {resp2.status_code}" + ) + else: + raise ConnectionException( + f"set_vlan({vlan_id}): API POST returned HTTP {resp.status_code}" + ) + + def _cli_set_vlan(self, vlan_id: int, config: VlanConfigDict) -> None: + self._enter_config_mode() + try: + lines = [f"vlan {vlan_id}"] + if "name" in config: + # Quote the name to handle spaces + lines.append(f' name "{config["name"]}"') + lines.append("exit") + errors = self._apply_config_lines("\n".join(lines)) + if errors: + raise CommandErrorException( + f"set_vlan({vlan_id}) errors: {errors}" + ) + finally: + self._exit_config_mode() + self._save_config() + + def set_interface(self, interface: str, config: InterfaceConfigDict) -> None: + """Configure a switch interface (mode, VLAN membership).""" + if self._transport == "api": + self._api_set_interface(interface, config) + else: + self._cli_set_interface(interface, config) + + def _api_set_interface(self, interface: str, config: InterfaceConfigDict) -> None: + mode = config.get("mode") + if mode == "trunk": + for vid in config.get("trunk_vlans", []): + payload = { + "vlan_id": vid, + "port_id": interface, + "port_mode": "POM_TAGGED_STATIC", + } + resp = self._api.post("vlans-ports", json=payload) + if not resp.ok and resp.status_code != 409: + raise ConnectionException( + f"set_interface({interface}): vlans-ports POST HTTP {resp.status_code}" + ) + elif mode == "access": + if "access_vlan" in config: + payload = { + "vlan_id": config["access_vlan"], + "port_id": interface, + "port_mode": "POM_UNTAGGED", + } + resp = self._api.post("vlans-ports", json=payload) + if not resp.ok and resp.status_code != 409: + raise ConnectionException( + f"set_interface({interface}): vlans-ports POST HTTP {resp.status_code}" + ) + + def _cli_set_interface(self, interface: str, config: InterfaceConfigDict) -> None: + mode = config.get("mode") + self._enter_config_mode() + try: + lines: List[str] = [] + if mode == "trunk": + for vid in config.get("trunk_vlans", []): + lines.append(f"vlan {vid}") + lines.append(f" tagged {interface}") + lines.append("exit") + elif mode == "access": + if "access_vlan" in config: + lines.append(f"vlan {config['access_vlan']}") + lines.append(f" untagged {interface}") + lines.append("exit") + if lines: + errors = self._apply_config_lines("\n".join(lines)) + if errors: + raise CommandErrorException( + f"set_interface({interface}) errors: {errors}" + ) + finally: + self._exit_config_mode() + self._save_config() + + def delete_vlan(self, vlan_id: int) -> None: + """Remove a VLAN from the switch.""" + if self._transport == "api": + self._api.delete_vlan(vlan_id) + else: + self._cli_delete_vlan(vlan_id) + + def _cli_delete_vlan(self, vlan_id: int) -> None: + self._enter_config_mode() + try: + errors = self._apply_config_lines(f"no vlan {vlan_id}") + if errors: + raise CommandErrorException( + f"delete_vlan({vlan_id}) errors: {errors}" + ) + finally: + self._exit_config_mode() + self._save_config() + + def ping( + self, + destination: str, + source: str = "", + ttl: int = 255, + timeout: int = 2, + size: int = 100, + count: int = 5, + vrf: str = "", + ) -> 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, + ) + + # ProCurve CLI ping: "ping repetitions " + cmd = f"ping {destination} repetitions {count} timeout {timeout}" + if source: + cmd += f" source {source}" + out = self._send_command(cmd) + return _parse_ping_output(out, count) + + # ------------------------------------------------------------------ + # NAPALM: cli + # ------------------------------------------------------------------ + + 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] = {} + for cmd in commands: + resp = self._api.post("cli", json={"cmd": cmd}) + if resp.ok: + results[cmd] = resp.json().get("result_base64_encoded", "") + else: + results[cmd] = f"Error: HTTP {resp.status_code}" + return results + + return {cmd: self._send_command(cmd) for cmd in commands} + + # ------------------------------------------------------------------ + # NAPALM: Configuration management (CLI only) + # ------------------------------------------------------------------ + + def load_merge_candidate( + self, filename: Optional[str] = None, config: Optional[str] = None + ) -> None: + """Load a candidate configuration for merging.""" + if self._transport == "api": + raise MergeConfigException( + "Merge config is not supported via REST API transport. Use SSH." + ) + if filename and config: + raise MergeConfigException("Provide either filename or config, not both.") + if filename: + with open(filename) as f: + config = f.read() + if config is None: + raise MergeConfigException("No configuration provided.") + self._candidate_config = config + self._candidate_mode = "merge" + + def load_replace_candidate( + self, filename: Optional[str] = None, config: Optional[str] = None + ) -> None: + """Load a candidate configuration for replacing. + + Note: ProCurve does not natively support atomic config replace. + The candidate is applied line-by-line (same as merge) when committed. + """ + if self._transport == "api": + raise ReplaceConfigException( + "Replace config is not supported via REST API transport. Use SSH." + ) + if filename and config: + raise ReplaceConfigException("Provide either filename or config, not both.") + if filename: + with open(filename) as f: + config = f.read() + if config is None: + raise ReplaceConfigException("No configuration provided.") + self._candidate_config = config + self._candidate_mode = "replace" + + def compare_config(self) -> str: + """Return a diff between the running config and the candidate config.""" + if self._candidate_config is None: + return "" + import difflib + + running = self._send_command("show running-config") + diff = difflib.unified_diff( + running.splitlines(), + self._candidate_config.splitlines(), + fromfile="running-config", + tofile="candidate-config", + lineterm="", + ) + return "\n".join(diff) + + def commit_config(self, message: str = "", revert_in: Optional[int] = None) -> None: + """Apply the candidate configuration to the device.""" + if self._candidate_config is None: + raise MergeConfigException("No candidate config loaded.") + + # Save running config as backup before applying + self._backup_config = self._send_command("show running-config") + + try: + self._enter_config_mode() + errors = self._apply_config_lines(self._candidate_config) + self._exit_config_mode() + except Exception as exc: + self._exit_config_mode() + raise MergeConfigException(f"Error applying config: {exc}") from exc + + if errors: + raise MergeConfigException( + "The following lines were rejected:\n" + "\n".join(errors) + ) + + self._save_config() + self._candidate_config = None + self._candidate_mode = None + + def discard_config(self) -> None: + """Discard the loaded candidate configuration.""" + self._candidate_config = None + self._candidate_mode = None + + def rollback(self) -> None: + """Re-apply the backup configuration saved before the last commit.""" + if self._backup_config is None: + raise MergeConfigException("No backup config available for rollback.") + self._candidate_config = self._backup_config + self._candidate_mode = "merge" + self.commit_config() + self._backup_config = None + + def has_pending_commit(self) -> bool: + """Return True if a candidate config is loaded.""" + return self._candidate_config is not None + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _is_float(s: str) -> bool: + try: + float(s) + return True + except (ValueError, TypeError): + return False + + +def _parse_ping_output(output: str, count: int) -> Dict: + """Parse ProCurve CLI ping output into NAPALM ping result dict. + + ProCurve ping output:: + + PING 10.0.0.1: 100 data bytes + !!!! + ----10.0.0.1 PING Statistics---- + 5 packets transmitted, 5 packets received, 0% packet loss + round-trip (ms) min/avg/max = 0/0/1 + """ + # Packet statistics line + m = re.search( + r"(\d+)\s+packets\s+transmitted,\s+(\d+)\s+packets\s+received", + output, re.I, + ) + if not m: + # Check for total failure + if "unreachable" in output.lower() or "no route" in output.lower(): + return { + "success": { + "probes_sent": count, + "packet_loss": count, + "rtt_min": 0.0, + "rtt_max": 0.0, + "rtt_avg": 0.0, + "rtt_stddev": 0.0, + "results": [], + } + } + return {"error": "Could not parse ping output"} + + sent = int(m.group(1)) + received = int(m.group(2)) + loss = sent - received + + # RTT line: "round-trip (ms) min/avg/max = 0/0/1" + rtt_min = rtt_avg = rtt_max = 0.0 + rtt_m = re.search(r"min/avg/max\s*=\s*([\d.]+)/([\d.]+)/([\d.]+)", output, re.I) + if rtt_m: + rtt_min = float(rtt_m.group(1)) + rtt_avg = float(rtt_m.group(2)) + rtt_max = float(rtt_m.group(3)) + + results = [{"ip_address": "", "rtt": rtt_avg}] * received + + return { + "success": { + "probes_sent": sent, + "packet_loss": loss, + "rtt_min": rtt_min, + "rtt_max": rtt_max, + "rtt_avg": rtt_avg, + "rtt_stddev": 0.0, + "results": results, + } + } diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..8bce098 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,55 @@ +[build-system] +requires = ["setuptools>=64", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "napalm-hpe-aruba-procurve" +version = "0.2.0" +description = "NAPALM driver for HPE/Aruba ProCurve switches with auto-detecting transport (REST API, SSH, legacy SSH, Telnet)" +readme = "README.md" +license = { text = "Apache-2.0" } +requires-python = ">=3.8" +authors = [ + { name = "Christian Manivong" }, +] +classifiers = [ + "Topic :: Utilities", + "License :: OSI Approved :: Apache Software License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Operating System :: POSIX :: Linux", + "Operating System :: MacOS", +] +dependencies = [ + "napalm>=4.0.0", + "netmiko>=4.0.0", + "netaddr", + "requests>=2.25.0", + "urllib3", +] + +[project.optional-dependencies] +dev = [ + "pytest", + "pytest-cov", + "black", + "ruff", +] + +[project.entry-points."napalm.drivers"] +procurve = "napalm_procurve:ProcurveDriver" + +[project.urls] +Repository = "https://github.com/napalm-automation-community/napalm-hpe-aruba-procurve" + +[tool.setuptools.packages.find] +where = ["."] +include = ["napalm_procurve*"] + +[tool.ruff] +line-length = 100 +target-version = "py38" diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..dbf6a25 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +# Tests for napalm-hpe-aruba-procurve diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py new file mode 100644 index 0000000..46956cf --- /dev/null +++ b/tests/unit/test_driver.py @@ -0,0 +1,653 @@ +"""Unit tests for ProcurveDriver and parsers — no real device required.""" + +import pytest +from unittest.mock import MagicMock, patch + +from napalm_procurve.procurve import ProcurveDriver, _parse_ping_output +from napalm_procurve import parsers + + +# =========================================================================== +# Sample CLI output fixtures (captured from real ProCurve devices) +# =========================================================================== + +SHOW_SYSTEM_INFO_2520G = """\ +Status and Counters - General System Information + + System Name : myswitch + System Contact : + System Location : + + MAC Age Time (sec) : 300 + Time Zone : 0 + Daylight Time Rule : None + + Software revision : R.11.27 Base MAC Addr : a1b2c3-d4e5f6 + ROM Version : R.10.06 Serial Number : SG12345678 + Active Image : Primary Up Time : 5 days 3:14 + + CPU Util (%) : 2 Memory - Total : 65011712 + IP Mgmt - Pkts Rx : 12345 Free : 37056512 +""" + +SHOW_SYSTEM_INFO_2530 = """\ +Status and Counters - General System Information + + System Name : aruba2530 + System Contact : + System Location : + + Firmware revision: YA.16.04.0006 + + Serial Number : SG87654321 Up Time : 2 days 1:30 + Base MAC Addr : aabbcc-ddeeff +""" + +SHOW_VERSION_2520G = """\ + HP J9565A 2520G-8-PoE Switch + Software revision : R.11.27 +""" + +SHOW_INTERFACES_BRIEF = """\ +Status and Counters - Port Status + + Flow Bcast + Port Type | Enabled Link MDI Mode Ctrl Limit + ------- ------------ + ------- ------ ----- ----------- ---- ----- + 1 100/1000T | Yes Up Auto 1000FDx off 0 + 2 100/1000T | Yes Down Auto 10HDx off 0 + 3 100/1000T | No Down Auto Unknown off 0 +""" + +SHOW_INTERFACES_PORT = """\ +Status and Counters - Port Counters for port 1 + + Name : + Link status : Up + MAC address : a1b2c3-d4e5f6 + Totals (since last boot): + Bytes Rx : 123456 + Bytes Tx : 654321 +""" + +SHOW_ARP = """\ + IP ARP table + + IP Address MAC Address Type Port + --------------- ----------------- ------- ---- + 192.168.0.1 a1b2c3-d4e5f6 dynamic 1 + 192.168.0.254 112233-445566 dynamic 2 +""" + +SHOW_MAC = """\ +Status and Counters - Port Address Table - All Ports + + MAC Address Port VLAN Type + ---------------- ------ ---- ------- + a1b2c3-d4e5f6 1 1 dynamic + 112233-445566 2 1 static +""" + +SHOW_LLDP_REMOTE = """\ +Status and Counters - Local Device Info + + LLDP Remote Devices Information + + LocalPort | ChassisId PortId SysName Capabilities TTL + --------- + --------------------- ------------- -------------- ------------ ------- + 1 | a1 b2 c3 d4 e5 f6 Gi 0/1 other-switch B 120 +""" + +SHOW_LLDP_DETAIL = """\ + LLDP Remote Device Information Detail for Port 1 + + ChassisType : mac-address + ChassisId : a1 b2 c3 d4 e5 f6 + PortType : locally-assigned + PortId : Gi0/1 + Port Description : GigabitEthernet0/1 + System Name : other-switch + System Description : Cisco IOS Software + System Capabilities : B R + Enabled Capabilities : B +""" + +SHOW_IP = """\ + Internet (IP) Service + + Default Gateway : 192.168.0.254 + + IP Configuration for VLAN 1: + + IP Config : Manual + IP Address : 192.168.0.1 + Subnet Mask : 255.255.255.0 +""" + +SHOW_NTP = """\ + NTP associations + + Address Ref-clock St When Poll Reach Delay Offset Jitter + --------------- --------- -- ----- ---- ----- ------ ------ ------ + *10.0.0.10 10.0.0.1 2 64 1024 377 0.620 0.000 0.000 + 10.0.0.20 .INIT. 16 - 1024 0 0.000 0.000 0.000 +""" + +SHOW_RUNNING_CONFIG = """\ +; J9565A Configuration Editor; Created on release #R.11.27 +; Ver #01:14.3f.f8 +hostname "myswitch" +module 1 type J9565A +vlan 1 + name "DEFAULT_VLAN" + untagged 1-8 + ip address 192.168.0.1 255.255.255.0 + exit +ip default-gateway 192.168.0.254 +""" + +PING_OUTPUT_SUCCESS = """\ +PING 10.0.0.1: 100 data bytes +!!!!! +----10.0.0.1 PING Statistics---- +5 packets transmitted, 5 packets received, 0% packet loss +round-trip (ms) min/avg/max = 0/1/2 +""" + +PING_OUTPUT_FAILURE = """\ +PING 10.99.99.99: 100 data bytes +..... +----10.99.99.99 PING Statistics---- +5 packets transmitted, 0 packets received, 100% packet loss +""" + +# =========================================================================== +# Fixtures +# =========================================================================== + + +@pytest.fixture +def driver(): + """Return a ProcurveDriver with mocked SSH connection (no real device).""" + with patch("napalm_procurve.procurve.ConnectHandler"): + drv = ProcurveDriver( + hostname="192.168.0.1", + username="manager", + password="secret", + ) + drv._transport = "ssh" + drv._device = MagicMock() + drv._device.base_prompt = "HP-2520G" + yield drv + + +# =========================================================================== +# Parser unit tests (pure functions — no mock needed) +# =========================================================================== + + +class TestMacConversion: + def test_standard_conversion(self): + assert parsers.procurve_mac_to_std("a1b2c3-d4e5f6") == "a1:b2:c3:d4:e5:f6" + + def test_already_colon(self): + assert parsers.procurve_mac_to_std("a1:b2:c3:d4:e5:f6") == "a1:b2:c3:d4:e5:f6" + + def test_uppercase_normalised(self): + assert parsers.procurve_mac_to_std("A1B2C3-D4E5F6") == "a1:b2:c3:d4:e5:f6" + + +class TestParseSystemInfo: + def test_2520g(self): + info = parsers.parse_system_info(SHOW_SYSTEM_INFO_2520G) + assert info["hostname"] == "myswitch" + assert info["os_version"] == "R.11.27" + assert info["serial_number"] == "SG12345678" + assert info["base_mac"] == "a1:b2:c3:d4:e5:f6" + assert info["uptime_seconds"] == 5 * 86400 + 3 * 3600 + 14 * 60 + + def test_2530(self): + info = parsers.parse_system_info(SHOW_SYSTEM_INFO_2530) + assert info["hostname"] == "aruba2530" + assert info["os_version"] == "YA.16.04.0006" + assert info["serial_number"] == "SG87654321" + + +class TestParseInterfacesBrief: + def test_parses_ports(self): + ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF) + assert "1" in ifaces + assert "2" in ifaces + assert "3" in ifaces + + def test_port_1_up(self): + ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF) + assert ifaces["1"]["is_up"] is True + assert ifaces["1"]["is_enabled"] is True + assert ifaces["1"]["speed"] == 1000.0 + + def test_port_2_down(self): + ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF) + assert ifaces["2"]["is_up"] is False + assert ifaces["2"]["is_enabled"] is True + + def test_port_3_disabled(self): + ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF) + assert ifaces["3"]["is_enabled"] is False + + +class TestParseArpTable: + def test_two_entries(self): + table = parsers.parse_arp_table(SHOW_ARP) + assert len(table) == 2 + + def test_first_entry(self): + table = parsers.parse_arp_table(SHOW_ARP) + entry = next(e for e in table if e["ip"] == "192.168.0.1") + assert entry["mac"] == "a1:b2:c3:d4:e5:f6" + assert entry["interface"] == "1" + + +class TestParseMacTable: + def test_two_entries(self): + table = parsers.parse_mac_table(SHOW_MAC) + assert len(table) == 2 + + def test_static_flag(self): + table = parsers.parse_mac_table(SHOW_MAC) + static_entry = next(e for e in table if e["mac"] == "11:22:33:44:55:66") + assert static_entry["static"] is True + + def test_dynamic_flag(self): + table = parsers.parse_mac_table(SHOW_MAC) + dyn_entry = next(e for e in table if e["mac"] == "a1:b2:c3:d4:e5:f6") + assert dyn_entry["static"] is False + + +class TestParseLldpNeighbors: + def test_one_neighbor(self): + nbrs = parsers.parse_lldp_neighbors(SHOW_LLDP_REMOTE) + assert "1" in nbrs + assert nbrs["1"][0]["hostname"] == "other-switch" + + def test_port_id(self): + nbrs = parsers.parse_lldp_neighbors(SHOW_LLDP_REMOTE) + assert "Gi" in nbrs["1"][0]["port"] or "0/1" in nbrs["1"][0]["port"] + + +class TestParseLldpDetail: + def test_system_name(self): + detail = parsers.parse_lldp_neighbors_detail(SHOW_LLDP_DETAIL) + assert "1" in detail + assert detail["1"][0]["remote_system_name"] == "other-switch" + + def test_port_description(self): + detail = parsers.parse_lldp_neighbors_detail(SHOW_LLDP_DETAIL) + assert "GigabitEthernet0/1" in detail["1"][0]["remote_port_description"] + + +class TestParseIpInterfaces: + def test_vlan1(self): + result = parsers.parse_ip_interfaces(SHOW_IP) + assert "VLAN1" in result + assert "192.168.0.1" in result["VLAN1"]["ipv4"] + assert result["VLAN1"]["ipv4"]["192.168.0.1"]["prefix_length"] == 24 + + +class TestParseNtpServers: + def test_two_servers(self): + servers = parsers.parse_ntp_servers(SHOW_NTP) + assert "10.0.0.10" in servers + assert "10.0.0.20" in servers + + +class TestParsePingOutput: + def test_success(self): + result = _parse_ping_output(PING_OUTPUT_SUCCESS, 5) + assert result["success"]["probes_sent"] == 5 + assert result["success"]["packet_loss"] == 0 + assert result["success"]["rtt_avg"] == 1.0 + + def test_failure(self): + result = _parse_ping_output(PING_OUTPUT_FAILURE, 5) + assert result["success"]["probes_sent"] == 5 + assert result["success"]["packet_loss"] == 5 + + +# =========================================================================== +# Driver integration tests (mocked netmiko) +# =========================================================================== + + +class TestDriverGetFacts: + def test_facts_from_cli(self, driver): + def side_effect(cmd, **kw): + cmd_str = cmd if isinstance(cmd, str) else cmd[0] + if "system-information" in cmd_str or "system information" in cmd_str: + return SHOW_SYSTEM_INFO_2520G + if "interfaces brief" in cmd_str: + return SHOW_INTERFACES_BRIEF + if "version" in cmd_str: + return SHOW_VERSION_2520G + return "" + + driver._send_command = MagicMock(side_effect=side_effect) + facts = driver.get_facts() + + assert facts["vendor"] == "HPE" + assert facts["hostname"] == "myswitch" + assert facts["os_version"] == "R.11.27" + assert facts["serial_number"] == "SG12345678" + assert isinstance(facts["interface_list"], list) + assert "1" in facts["interface_list"] + + +class TestDriverGetInterfaces: + def test_interfaces_from_cli(self, driver): + driver._send_command = MagicMock(return_value=SHOW_INTERFACES_BRIEF) + ifaces = driver.get_interfaces() + assert "1" in ifaces + assert ifaces["1"]["is_up"] is True + + +class TestDriverGetArpTable: + def test_arp_from_cli(self, driver): + driver._send_command = MagicMock(return_value=SHOW_ARP) + table = driver.get_arp_table() + assert len(table) == 2 + assert any(e["ip"] == "192.168.0.1" for e in table) + + +class TestDriverGetMacTable: + def test_mac_from_cli(self, driver): + driver._send_command = MagicMock(return_value=SHOW_MAC) + table = driver.get_mac_address_table() + assert len(table) == 2 + + +class TestDriverGetConfig: + def test_running_config(self, driver): + driver._send_command = MagicMock(return_value=SHOW_RUNNING_CONFIG) + cfg = driver.get_config(retrieve="running") + assert "hostname" in cfg["running"] + assert cfg["startup"] == "" + + +class TestDriverConfigManagement: + def test_load_and_discard(self, driver): + driver.load_merge_candidate(config="hostname newname") + assert driver._candidate_config == "hostname newname" + driver.discard_config() + assert driver._candidate_config is None + + def test_compare_config(self, driver): + driver._send_command = MagicMock(return_value=SHOW_RUNNING_CONFIG) + driver.load_merge_candidate(config='hostname "differentname"') + diff = driver.compare_config() + assert "differentname" in diff or "myswitch" in diff + + def test_has_pending_commit_false(self, driver): + assert driver.has_pending_commit() is False + + def test_has_pending_commit_true(self, driver): + driver.load_merge_candidate(config="no lldp run") + assert driver.has_pending_commit() is True + + +class TestDriverTransportDetection: + def test_force_ssh_transport(self): + """Forcing transport=ssh skips API probe.""" + with patch("napalm_procurve.procurve.ConnectHandler") as mock_conn: + mock_conn.return_value = MagicMock() + drv = ProcurveDriver( + "192.168.0.1", "manager", "secret", + optional_args={"transport": "ssh"}, + ) + drv.open() + assert drv._transport == "ssh" + + def test_force_telnet_transport(self): + """Forcing transport=telnet skips API and SSH probes.""" + with patch("napalm_procurve.procurve.ConnectHandler") as mock_conn: + mock_conn.return_value = MagicMock() + drv = ProcurveDriver( + "192.168.0.1", "manager", "secret", + optional_args={"transport": "telnet"}, + ) + drv.open() + assert drv._transport == "telnet" + + def test_all_transports_fail_raises(self): + """ConnectionException raised when all transports fail.""" + with patch("napalm_procurve.procurve.ProcurveApiClient.probe", return_value=(None, None)): + with patch("napalm_procurve.procurve.ConnectHandler", side_effect=Exception("no conn")): + drv = ProcurveDriver("192.168.0.1", "manager", "secret") + with pytest.raises(Exception): + drv.open() + + +# =========================================================================== +# VLAN parser tests +# =========================================================================== + +RUNNING_CONFIG_VLANS = """\ +; J9565A Configuration Editor; Created on release #R.11.27 +; Ver #01:14.3f.f8 +hostname "myswitch" +module 1 type J9565A + +vlan 1 + name "DEFAULT_VLAN" + untagged 1-4,6-8 + ip address 192.168.0.1 255.255.255.0 + exit + +vlan 10 + name "MGMT" + tagged 1,2,5 + exit + +vlan 20 + name "IoT" + untagged 5 + tagged 1-3 + exit +""" + +SHOW_VLANS_BRIEF = """\ +Status and Counters - VLAN Information + + VLAN ID Name | Status Voice Jumbo + ------- ---------------- + ---------- ----- ----- + 1 DEFAULT_VLAN | Port-based No No + 10 MANAGEMENT | Port-based No No + 20 IoT | Port-based No No +""" + +SHOW_VLAN_DETAIL_10 = """\ +Status and Counters - VLAN Information - VLAN 10 + + VLAN ID : 10 + Name : MANAGEMENT + + Port Information Mode Unknown VLAN Status + ---------------- -------- ------------ ------- + 1 Tagged Learn Up + 2 Tagged Learn Up + 5 Tagged Learn Up +""" + +SHOW_VLAN_DETAIL_20 = """\ +Status and Counters - VLAN Information - VLAN 20 + + VLAN ID : 20 + Name : IoT + + Port Information Mode Unknown VLAN Status + ---------------- -------- ------------ ------- + 1 Tagged Learn Up + 2 Tagged Learn Up + 3 Tagged Learn Up + 5 Untagged Learn Up +""" + + +class TestExpandProcurvePorts: + def test_numeric_range(self): + assert parsers.expand_procurve_ports("1-4") == ["1", "2", "3", "4"] + + def test_mixed_list(self): + assert parsers.expand_procurve_ports("1-3,6,8") == ["1", "2", "3", "6", "8"] + + def test_trunk_port(self): + assert parsers.expand_procurve_ports("Trk1,Trk2") == ["Trk1", "Trk2"] + + def test_alpha_range(self): + assert parsers.expand_procurve_ports("A1-A3") == ["A1", "A2", "A3"] + + def test_single_port(self): + assert parsers.expand_procurve_ports("5") == ["5"] + + def test_mixed_all(self): + result = parsers.expand_procurve_ports("1-2,Trk1,A1-A2") + assert result == ["1", "2", "Trk1", "A1", "A2"] + + +class TestParseVlansFromRunningConfig: + def test_vlan_count(self): + vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) + assert len(vlans) == 3 + assert 1 in vlans + assert 10 in vlans + assert 20 in vlans + + def test_vlan_names(self): + vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) + assert vlans[1]["name"] == "DEFAULT_VLAN" + assert vlans[10]["name"] == "MGMT" + assert vlans[20]["name"] == "IoT" + + def test_vlan1_untagged_ports(self): + vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) + # "1-4,6-8" → 1,2,3,4,6,7,8 + assert vlans[1]["untagged"] == ["1", "2", "3", "4", "6", "7", "8"] + assert vlans[1]["tagged"] == [] + + def test_vlan10_tagged_ports(self): + vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) + assert vlans[10]["tagged"] == ["1", "2", "5"] + assert vlans[10]["untagged"] == [] + + def test_vlan20_mixed(self): + vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) + assert vlans[20]["untagged"] == ["5"] + assert vlans[20]["tagged"] == ["1", "2", "3"] + + def test_interfaces_union_untagged_first(self): + vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) + # For VLAN 20: untagged ["5"] + tagged ["1","2","3"] → interfaces ordered untagged first + assert vlans[20]["interfaces"][0] == "5" + assert set(vlans[20]["interfaces"]) == {"1", "2", "3", "5"} + + def test_interfaces_napalm_key_present(self): + vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) + for vdata in vlans.values(): + assert "interfaces" in vdata + + +class TestParseVlansBrief: + def test_detects_all_vlans(self): + result = parsers.parse_vlans_brief(SHOW_VLANS_BRIEF) + assert set(result.keys()) == {1, 10, 20} + + def test_names(self): + result = parsers.parse_vlans_brief(SHOW_VLANS_BRIEF) + assert result[1] == "DEFAULT_VLAN" + assert result[20] == "IoT" + + +class TestParseVlanPortsDetail: + def test_all_tagged(self): + ports = parsers.parse_vlan_ports_detail(SHOW_VLAN_DETAIL_10) + assert ports["tagged"] == ["1", "2", "5"] + assert ports["untagged"] == [] + + def test_mixed_tagged_untagged(self): + ports = parsers.parse_vlan_ports_detail(SHOW_VLAN_DETAIL_20) + assert ports["tagged"] == ["1", "2", "3"] + assert ports["untagged"] == ["5"] + + +class TestDerivePvids: + def test_basic_pvid_derivation(self): + vlans = { + 1: {"untagged": ["1", "2", "3"], "tagged": []}, + 10: {"untagged": ["4"], "tagged": ["5"]}, + } + pvids = parsers.derive_port_pvids(vlans) + assert pvids["1"] == 1 + assert pvids["2"] == 1 + assert pvids["3"] == 1 + assert pvids["4"] == 10 + + def test_tagged_only_port_gets_pvid1(self): + vlans = { + 1: {"untagged": [], "tagged": []}, + 10: {"untagged": [], "tagged": ["5"]}, + } + pvids = parsers.derive_port_pvids(vlans) + assert pvids["5"] == 1 + + def test_pvid_from_running_config(self): + vlans = parsers.parse_vlans_from_running_config(RUNNING_CONFIG_VLANS) + pvids = parsers.derive_port_pvids(vlans) + # Port 1 is untagged in VLAN 1 + assert pvids["1"] == 1 + # Port 5 is untagged in VLAN 20 + assert pvids["5"] == 20 + # Port 2 is untagged in VLAN 1 but tagged in VLAN 20 — PVID is 1 + assert pvids["2"] == 1 + + +class TestDriverGetVlans: + def test_get_vlans_from_cli(self, driver): + driver._send_command = MagicMock(return_value=RUNNING_CONFIG_VLANS) + vlans = driver.get_vlans() + + assert 1 in vlans + assert 10 in vlans + assert vlans[1]["name"] == "DEFAULT_VLAN" + assert vlans[10]["tagged"] == ["1", "2", "5"] + assert vlans[20]["untagged"] == ["5"] + # NAPALM compat + assert "interfaces" in vlans[1] + + def test_get_vlans_fallback_to_show_vlans(self, driver): + """When running-config has no VLANs, fall back to show vlans.""" + call_count = [0] + + 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 + 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]): + return SHOW_VLAN_DETAIL_10 + if "show vlans 20" in (cmd if isinstance(cmd, str) else cmd[0]): + return SHOW_VLAN_DETAIL_20 + return "" + + driver._send_command = MagicMock(side_effect=side_effect) + vlans = driver.get_vlans() + assert 10 in vlans + assert vlans[10]["tagged"] == ["1", "2", "5"] + + def test_get_port_pvids(self, driver): + driver._send_command = MagicMock(return_value=RUNNING_CONFIG_VLANS) + pvids = driver.get_port_pvids() + # Port 5 is untagged in VLAN 20 + assert pvids["5"] == 20 + # Ports 1-4 are untagged in VLAN 1 + assert pvids["1"] == 1 + assert pvids["4"] == 1