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

feat: synthesize LAG/trunk interfaces and support trunk member editing

Browse Source 
get_interfaces() now fills in ports missing from the /ports collection
(e.g. the 4th SFP uplink on a 2530-48G) using system/status/switch, and
synthesizes a logical interface for each configured LAG/trunk group with
lag_members/lag_mode. New set_lag_members() adds/removes trunk members via
the REST /cli passthrough (PUT ports/{id} silently ignores trunk_group).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Christian Manivong
2026-06-12 14:47:00 +02:00
co-authored by Claude Sonnet 4.6
parent ef544be540
commit 576752da23
4 changed files with 96 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
napalm_procurve/api_client.py
+47 -1
View File
@@ -242,8 +242,33 @@ class ProcurveApiClient:
stats_data = self.get("port-statistics")
output: Dict[str, Dict] = {}
trunk_groups: Dict[str, list] = {}
trunk_modes: Dict[str, str] = {}
for port in ports_data.get("port_element", []):
port_elements = list(ports_data.get("port_element", []))
# The /ports collection can under-report the switch's true port
# count (observed: a 2530-48G reports 51 of its 52 ports). Fill in
# any ports known to system/status/switch but missing from /ports.
seen_ids = {p.get("id") for p in port_elements}
try:
sw_status = self.get("system/status/switch")
for blade in sw_status.get("blades", []):
for p in blade.get("data_ports", []):
pid = p.get("port_name", "")
if pid and pid not in seen_ids:
port_elements.append({
"id": pid,
"name": "",
"is_port_up": p.get("operStatus") == "OPER_UP",
"is_port_enabled": p.get("adminStatus") == "ADMIN_UP",
"trunk_group": "",
})
seen_ids.add(pid)
except Exception:
pass
for port in port_elements:
pid = port.get("id", "")
output[pid] = {
"is_up": bool(port.get("is_port_up", False)),
@@ -254,12 +279,33 @@ class ProcurveApiClient:
"mtu": -1,
"mac_address": "",
}
group = port.get("trunk_group") or ""
if group:
output[pid]["trunk_group"] = group
trunk_groups.setdefault(group, []).append(pid)
if port.get("trunk_mode"):
trunk_modes[group] = port["trunk_mode"]
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))
# Synthesize a logical interface entry for each configured LAG/trunk
# group so it shows up as its own row alongside its member ports.
for group, members in trunk_groups.items():
output[group] = {
"is_up": any(output[m]["is_up"] for m in members),
"is_enabled": any(output[m]["is_enabled"] for m in members),
"description": f"LAG ({', '.join(sorted(members, key=lambda s: int(s) if s.isdigit() else 0))})",
"last_flapped": -1.0,
"speed": sum(output[m]["speed"] for m in members),
"mtu": -1,
"mac_address": "",
"lag_members": members,
"lag_mode": "lacp" if trunk_modes.get(group) == "PTT_LACP" else "trunk",
}
return output
def get_interfaces_ip(self) -> Dict[str, Dict]: