diff --git a/napalm_device_types/__init__.py b/napalm_device_types/__init__.py index 37d2060..489c22e 100644 --- a/napalm_device_types/__init__.py +++ b/napalm_device_types/__init__.py @@ -61,6 +61,7 @@ from napalm_device_types.os import OSDriver from napalm_device_types.firewall_rules import FirewallRuleMixin from napalm_device_types.health_metrics import HealthMetricsMixin from napalm_device_types.interface_filter import InterfaceFilterMixin +from napalm_device_types.lag import add_lag_interfaces from napalm_device_types.mac_acl import MacAclMixin from napalm_device_types.media import MediaDriver from napalm_device_types.nat_vpn import NatVpnMixin @@ -98,6 +99,7 @@ __all__ = [ "StorageDriver", "SwitchDriver", "UpdateMixin", + "add_lag_interfaces", "driver_supports_ping", "normalize_cidr", "normalize_mac", diff --git a/napalm_device_types/lag.py b/napalm_device_types/lag.py new file mode 100644 index 0000000..d3a850c --- /dev/null +++ b/napalm_device_types/lag.py @@ -0,0 +1,57 @@ +# -*- coding: utf-8 -*- +"""Logical LAG entries for ``get_interfaces()``, built from their member ports. + +Some switches list only physical ports, each tagged with the trunk it belongs +to, and never the trunk itself. Turning those tags into one row per trunk is +the same for every vendor, so it lives here once; a driver only has to set +``trunk_group`` on member ports and, where the device says so, pass the mode. +""" + +from __future__ import annotations + +import re +from typing import Any, Dict, List, Optional + + +def _port_order(name: str) -> List[Any]: + return [int(p) if p.isdigit() else p for p in re.split(r"(\d+)", name)] + + +def add_lag_interfaces( + interfaces: Dict[str, Dict[str, Any]], + lag_modes: Optional[Dict[str, str]] = None, +) -> Dict[str, Dict[str, Any]]: + """Return *interfaces* plus one logical entry per ``trunk_group``. + + The LAG entry is up/enabled if any member is, its speed is the members' + sum, and ``lag_members`` lists them in port order. A LAG the driver + already reported is left as it is. *interfaces* itself is not modified. + + :param lag_modes: ``{lag_name: "lacp" | "trunk"}``. A LAG without a known + mode gets no ``lag_mode`` key rather than a guessed one. + """ + result = dict(interfaces) + groups: Dict[str, List[str]] = {} + for name, iface in interfaces.items(): + group = iface.get("trunk_group") + if group: + groups.setdefault(group, []).append(name) + + for group, members in groups.items(): + if group in result: + continue + members = sorted(members, key=_port_order) + lag: Dict[str, Any] = { + "is_up": any(interfaces[m].get("is_up") for m in members), + "is_enabled": any(interfaces[m].get("is_enabled") for m in members), + "description": f"LAG ({', '.join(members)})", + "last_flapped": -1.0, + "speed": sum(float(interfaces[m].get("speed") or 0) for m in members), + "mtu": -1, + "mac_address": "", + "lag_members": members, + } + if lag_modes and group in lag_modes: + lag["lag_mode"] = lag_modes[group] + result[group] = lag + return result diff --git a/tests/test_lag_interfaces.py b/tests/test_lag_interfaces.py new file mode 100644 index 0000000..1707c2d --- /dev/null +++ b/tests/test_lag_interfaces.py @@ -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"]