feat: add_lag_interfaces, one logical row per trunk group

Some switches list only their member ports, each tagged with the trunk
it belongs to, and never the trunk itself. procurve over CLI is one:
`show interfaces brief` has `3-Trk3` and `4-Trk3` but no `Trk3`. Its
REST path already built the trunk row itself, in code no other driver
could reach.

Grouping members by `trunk_group` into one entry per group is the same
for every vendor, so it lives here once. The entry is up/enabled if any
member is, its speed is the members' sum, and `lag_members` is in port
order. A LAG the driver already reported is left alone.

`lag_mode` is set only when the driver passes it. netOrk shows a missing
mode as "static trunk", but a guessed "trunk" would label an LACP group
wrongly, and a label that looks sure when nothing is known is worse.

A free function, not a SwitchDriver method: role bases are declarations
only (test_role_contracts), like normalize_cidr beside DhcpServerMixin.
This commit is contained in:
2026-09-25 10:17:18 +02:00
parent 70841feaa1
commit f3fa75bbca
3 changed files with 135 additions and 0 deletions
+76
View File
@@ -0,0 +1,76 @@
"""Tests for add_lag_interfaces — one logical row per trunk group."""
from napalm_device_types import add_lag_interfaces
def _port(is_up: bool = True, is_enabled: bool = True, speed: float = 1000.0, trunk_group: str = "") -> dict:
port = {
"is_up": is_up,
"is_enabled": is_enabled,
"description": "",
"last_flapped": -1.0,
"speed": speed,
"mtu": -1,
"mac_address": "",
}
if trunk_group:
port["trunk_group"] = trunk_group
return port
def test_adds_one_row_per_trunk_group():
ifaces = {
"1": _port(),
"3": _port(trunk_group="Trk3"),
"4": _port(trunk_group="Trk3"),
"10": _port(trunk_group="Trk6"),
"7": _port(trunk_group="Trk6"),
}
result = add_lag_interfaces(ifaces)
assert result["Trk3"]["lag_members"] == ["3", "4"]
# Members in natural port order, not string order ("7" before "10").
assert result["Trk6"]["lag_members"] == ["7", "10"]
assert result["Trk6"]["description"] == "LAG (7, 10)"
assert "Trk1" not in result
def test_state_is_derived_from_members():
ifaces = {
"3": _port(is_up=False, speed=1000.0, trunk_group="Trk3"),
"4": _port(is_up=True, speed=1000.0, trunk_group="Trk3"),
"6": _port(is_up=False, is_enabled=False, trunk_group="Trk6"),
}
result = add_lag_interfaces(ifaces)
assert result["Trk3"]["is_up"] is True
assert result["Trk3"]["is_enabled"] is True
assert result["Trk3"]["speed"] == 2000.0
assert result["Trk6"]["is_up"] is False
assert result["Trk6"]["is_enabled"] is False
def test_lag_mode_only_when_known():
"""The UI reads a missing mode as "static trunk"; guessing would mislabel LACP."""
ifaces = {"3": _port(trunk_group="Trk3"), "6": _port(trunk_group="Trk6")}
result = add_lag_interfaces(ifaces, lag_modes={"Trk3": "lacp"})
assert result["Trk3"]["lag_mode"] == "lacp"
assert "lag_mode" not in result["Trk6"]
def test_keeps_a_lag_the_driver_already_reported():
ifaces = {
"3": _port(trunk_group="Trk3"),
"Trk3": {**_port(), "description": "uplink", "lag_members": ["3"]},
}
result = add_lag_interfaces(ifaces)
assert result["Trk3"]["description"] == "uplink"
def test_does_not_modify_its_input():
ifaces = {"3": _port(trunk_group="Trk3")}
add_lag_interfaces(ifaces)
assert list(ifaces) == ["3"]