initial commit
This commit is contained in:
Christian Manivong
@@ -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
|
||||
@@ -0,0 +1,5 @@
|
||||
"""NAPALM driver for HPE/Aruba ProCurve switches."""
|
||||
|
||||
from napalm_procurve.procurve import ProcurveDriver
|
||||
|
||||
__all__ = ["ProcurveDriver"]
|
||||
@@ -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/<version>/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,
|
||||
}
|
||||
}
|
||||
@@ -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 <port>`` 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 <id>"
|
||||
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 <portlist>
|
||||
utm = re.match(r"^untagged\s+(.+)$", stripped, re.I)
|
||||
if utm:
|
||||
vlans[current_id]["untagged"] = expand_procurve_ports(utm.group(1))
|
||||
continue
|
||||
|
||||
# tagged <portlist>
|
||||
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 <id>`` 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
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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"
|
||||
@@ -0,0 +1 @@
|
||||
# Tests for napalm-hpe-aruba-procurve
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user