Merge pull request 'feat!: a VM's vmid is a string, and its config can describe its hardware' (#1) from feature/vmid-as-string into main

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.

napalm_device_types/__init__.py

@@ -63,6 +63,7 @@ from napalm_device_types.firewall_rules import FirewallRuleMixin
 from napalm_device_types.health_metrics import HealthMetricsMixin
 from napalm_device_types.host_reboot import HostRebootMixin
 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
@@ -101,6 +102,7 @@ __all__ = [
     "StorageDriver",
     "SwitchDriver",
     "UpdateMixin",
+    "add_lag_interfaces",
     "driver_supports_ping",
     "normalize_cidr",
     "normalize_mac",

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

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"]