Merge pull request 'feat: port forwards are a firewall reader too, and only the WAN's' (#2) from feature/port-forwards-shared into main

get_port_forwards was declared on ResidentialGatewayDriver alone, as if a
port forward were a home-router feature. A firewall forwards ports just the
same (OPNsense calls it destination NAT), and netOrk asks both: is this host
reachable from the internet, which CVEs are exposed. The declaration moves to
NatVpnMixin, where the two roles already overlap, and PortForwardDict next to
NATTranslationDict.

The contract now says what counts. Destination NAT between internal networks
and rules that only exempt traffic are not port forwards: callers read every
entry as "reachable from outside". "ANY" forwards every protocol and an
external port of 0 every port -- a whole host forwarded is the most exposed
case and must not fall out for lack of a port number.

Declaration only, under TYPE_CHECKING: nothing changes at runtime.

README.md

@@ -70,6 +70,8 @@ Behaviour shared across roles lives in a function class, exactly once, and a rol
 is a thin bundle over them — `PackageManagementMixin`, `HealthMetricsMixin`,
 `ServiceControlMixin`, `UpdateMixin`, `NatVpnMixin`, `MacAclMixin`, `FirewallRuleMixin`,
 `DhcpServerMixin`, `PingSweepMixin`, `ConfigLifecycleMixin`, `InterfaceFilterMixin`.
+`HostRebootMixin` (`reboot_host`) is mixed into `DeviceTypeDriver` itself, since any
+device may be restartable; like the others it only declares.
 
 A function class may use the **template form** — public method concrete, the
 device-specific part a `_hook` declared under `if TYPE_CHECKING` — *when the base
@@ -225,6 +227,11 @@ class PfSenseDriver(FirewallDriver):
     def get_vpn_tunnels(self):
         # return Dict[str, VPNTunnelDict]
         ...
+
+    def get_port_forwards(self):
+        # return List[PortForwardDict] — forwards from the WAN only, never a
+        # redirect between internal networks (shared with home gateways)
+        ...
 ```
 
 ### Hypervisor
@@ -238,7 +245,7 @@ class ProxmoxDriver(HypervisorDriver):
         # return List[VMDict]
         ...
 
-    def snapshot_create(self, name, snapshot, description="", include_memory=False):
+    def create_vm_snapshot(self, name, snapshot, description="", include_memory=False):
         ...
 ```
 

napalm_device_types/__init__.py

@@ -38,6 +38,7 @@ instead of being restated on every role that happens to need it:
 * :class:`~napalm_device_types.dhcp.DhcpServerMixin`
 * :class:`~napalm_device_types.firewall_rules.FirewallRuleMixin`
 * :class:`~napalm_device_types.health_metrics.HealthMetricsMixin`
+* :class:`~napalm_device_types.host_reboot.HostRebootMixin`
 * :class:`~napalm_device_types.interface_filter.InterfaceFilterMixin`
 * :class:`~napalm_device_types.mac_acl.MacAclMixin`
 * :class:`~napalm_device_types.nat_vpn.NatVpnMixin`
@@ -60,7 +61,9 @@ from napalm_device_types.hypervisor import HypervisorDriver
 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.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
@@ -83,6 +86,7 @@ __all__ = [
     "FirewallDriver",
     "FirewallRuleMixin",
     "HealthMetricsMixin",
+    "HostRebootMixin",
     "HypervisorDriver",
     "InterfaceFilterMixin",
     "MacAclMixin",
@@ -98,6 +102,7 @@ __all__ = [
     "StorageDriver",
     "SwitchDriver",
     "UpdateMixin",
+    "add_lag_interfaces",
     "driver_supports_ping",
     "normalize_cidr",
     "normalize_mac",

napalm_device_types/base.py

@@ -12,6 +12,7 @@ from typing import NamedTuple
 
 from napalm.base import NetworkDriver
 
+from napalm_device_types.host_reboot import HostRebootMixin
 from napalm_device_types.ping_sweep import PingSweepMixin
 
 
@@ -47,7 +48,7 @@ class PortSpec(NamedTuple):
     mandatory: bool = False
 
 
-class DeviceTypeDriver(PingSweepMixin, NetworkDriver):
+class DeviceTypeDriver(PingSweepMixin, HostRebootMixin, NetworkDriver):
     """Common base for all netOrk device-type drivers.
 
     Sits between napalm.base.NetworkDriver and the type-specific abstract

napalm_device_types/host_reboot.py

@@ -0,0 +1,30 @@
+"""Restarting the device itself.
+
+Declared under ``if TYPE_CHECKING``: a contract, not a placeholder. Only a
+driver that can actually restart its device defines ``reboot_host``, so
+``hasattr(driver, "reboot_host")`` tells a caller whether to offer it.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+
+class HostRebootMixin:
+    if TYPE_CHECKING:
+
+        def reboot_host(self) -> None:
+            """
+            Restarts the device this driver is connected to.
+
+            Returns once the device has accepted the request; the session is
+            usually gone right after. Waiting for the device to come back is
+            the caller's business (see ``REBOOT_SETTLE_SECONDS``).
+
+            A driver that manages other machines restarts its *own* host, never
+            one of them: a hypervisor restarts the hypervisor, not a VM.
+
+            :raises RuntimeError: If the device refuses, e.g. an ESXi host that
+                is not in maintenance mode while VMs are running.
+            """
+            ...

napalm_device_types/hypervisor.py

@@ -43,6 +43,11 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
     ROLE: str = "hypervisor"
     TYPE_LABEL: str = "Hypervisor"
 
+    #: What cloud-init installs and starts on a VM provisioned through this
+    #: driver, so the hypervisor can read the guest's IP address back.
+    GUEST_AGENT_PACKAGES: tuple[str, ...] = ("qemu-guest-agent",)
+    GUEST_AGENT_RUNCMD: tuple[str, ...] = ("systemctl enable --now qemu-guest-agent",)
+
 
 
     # ------------------------------------------------------------------
@@ -59,7 +64,8 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
             Each entry contains:
 
             * name (string) - VM display name
-            * vmid (int) - hypervisor-internal numeric ID
+            * vmid (string) - hypervisor-internal ID (Proxmox: ``"100"``,
+              VMware: the instance UUID)
             * status (string) - ``"running"``, ``"stopped"``, ``"paused"``, ``"suspended"``
             * vcpus (int) - number of virtual CPUs assigned
             * memory (int) - configured RAM in megabytes
@@ -73,7 +79,7 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
                 [
                     {
                         "name": "web01",
-                        "vmid": 100,
+                        "vmid": "100",
                         "status": "running",
                         "vcpus": 4,
                         "memory": 8192,
@@ -84,7 +90,7 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
                     },
                     {
                         "name": "db-backup",
-                        "vmid": 101,
+                        "vmid": "101",
                         "status": "stopped",
                         "vcpus": 2,
                         "memory": 4096,
@@ -101,13 +107,14 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
             """
             Returns the full hardware configuration of a virtual machine.
 
-            :param name: VM name or numeric VMID as a string.
+            :param name: VM name or its ``vmid``.
             :raises ValueError: If no VM with the given name/ID exists.
 
             The returned dictionary contains:
 
             * name (string) - VM display name
-            * vmid (int) - hypervisor-internal numeric ID
+            * vmid (string) - hypervisor-internal ID (Proxmox: ``"100"``,
+              VMware: the instance UUID)
             * vcpus (int) - number of virtual CPUs
             * memory (int) - RAM in megabytes
             * os_type (string) - guest OS type hint (e.g. ``"l26"``, ``"win11"``, ``"other"``)
@@ -131,11 +138,21 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
             * description (string) - free-text notes / description
             * tags (list of strings) - organisational tags
 
+            Optional, present only where the hypervisor exposes them:
+
+            * os_name (string) - human-readable guest OS
+            * cpu_type (string) - emulated CPU model
+            * sockets (int), cores_per_socket (int) - vCPU topology
+            * firmware (string) - ``"bios"`` or ``"efi"``
+            * machine (string) - machine type / virtual hardware version
+            * passthrough (list) - host devices handed to the VM, each with
+              ``slot``, ``kind`` (``"pci"``/``"usb"``) and ``config``
+
             Example::
 
                 {
                     "name": "web01",
-                    "vmid": 100,
+                    "vmid": "100",
                     "vcpus": 4,
                     "memory": 8192,
                     "os_type": "l26",
@@ -174,7 +191,7 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
 
             The method blocks until the hypervisor reports the VM as running.
 
-            :param name: VM name or numeric VMID as a string.
+            :param name: VM name or its ``vmid``.
             :raises ValueError: If no VM with the given name/ID exists.
             :raises RuntimeError: If the VM cannot be started (e.g. resource limit).
 
@@ -192,7 +209,7 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
             and the method blocks until the VM is stopped.  With ``force=True``
             the VM is immediately powered off (equivalent to pulling the plug).
 
-            :param name: VM name or numeric VMID as a string.
+            :param name: VM name or its ``vmid``.
             :param force: ``True`` for immediate power-off, ``False`` for graceful shutdown.
             :raises ValueError: If no VM with the given name/ID exists.
             :raises RuntimeError: If the VM is already stopped.
@@ -211,7 +228,7 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
             With ``force=False`` (default) a graceful ACPI reboot is requested.
             With ``force=True`` the VM is reset immediately without OS shutdown.
 
-            :param name: VM name or numeric VMID as a string.
+            :param name: VM name or its ``vmid``.
             :param force: ``True`` for an immediate reset, ``False`` for graceful reboot.
             :raises ValueError: If no VM with the given name/ID exists.
             :raises RuntimeError: If the VM is not currently running.
@@ -228,7 +245,7 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
             Suspends (pauses) a running virtual machine, preserving its in-memory
             state.  The VM can be resumed with :meth:`start_vm`.
 
-            :param name: VM name or numeric VMID as a string.
+            :param name: VM name or its ``vmid``.
             :raises ValueError: If no VM with the given name/ID exists.
             :raises RuntimeError: If the VM is not currently running.
 
@@ -246,7 +263,7 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
             """
             Returns all snapshots of a virtual machine.
 
-            :param name: VM name or numeric VMID as a string.
+            :param name: VM name or its ``vmid``.
             :raises ValueError: If no VM with the given name/ID exists.
 
             Each entry contains:
@@ -286,7 +303,7 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
             """
             Creates a snapshot of a virtual machine.
 
-            :param name: VM name or numeric VMID as a string.
+            :param name: VM name or its ``vmid``.
             :param snapshot: Name for the new snapshot.
             :param description: Optional human-readable description.
             :param include_memory: Whether to include the current RAM state
@@ -306,7 +323,7 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
             """
             Deletes a snapshot of a virtual machine.
 
-            :param name: VM name or numeric VMID as a string.
+            :param name: VM name or its ``vmid``.
             :param snapshot: Name of the snapshot to delete.
             :raises ValueError: If the VM or snapshot does not exist.
             :raises RuntimeError: If other snapshots depend on this one (must delete children first).
@@ -324,7 +341,7 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
             The VM is stopped (if running), reverted, and then left in the state
             the snapshot recorded (running or stopped depending on ``has_memory``).
 
-            :param name: VM name or numeric VMID as a string.
+            :param name: VM name or its ``vmid``.
             :param snapshot: Name of the snapshot to roll back to.
             :raises ValueError: If the VM or snapshot does not exist.
             :raises RuntimeError: If the rollback fails.

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

napalm_device_types/models.py

@@ -298,6 +298,21 @@ class NATTranslationDict(TypedDict):
     age: float
 
 
+class PortForwardDict(TypedDict):
+    """A port the WAN side can reach, forwarded to a host inside.
+
+    Shared by firewalls and home gateways (``NatVpnMixin.get_port_forwards``).
+    """
+
+    name: str
+    protocol: str  # "TCP" or "UDP"
+    external_port: int
+    internal_ip: str
+    internal_port: int
+    enabled: bool
+    remote_host: NotRequired[str]  # restrict forward to a specific remote source
+
+
 class SecurityZoneDict(TypedDict):
     interfaces: List[str]
     policy: str
@@ -470,16 +485,6 @@ class WANStatusDict(TypedDict):
     link_status: NotRequired[str]  # physical line state, e.g. "Up" / "Down"
 
 
-class PortForwardDict(TypedDict):
-    name: str
-    protocol: str  # "TCP" or "UDP"
-    external_port: int
-    internal_ip: str
-    internal_port: int
-    enabled: bool
-    remote_host: NotRequired[str]  # restrict forward to a specific remote source
-
-
 class HostDict(TypedDict):
     mac: str
     ip: str
@@ -512,7 +517,7 @@ class VMNICDict(TypedDict):
 
 class VMDict(TypedDict):
     name: str
-    vmid: int
+    vmid: str
     status: str
     vcpus: int
     memory: int
@@ -522,9 +527,17 @@ class VMDict(TypedDict):
     node: str
 
 
+class VMPassthroughDict(TypedDict):
+    """A host device handed through to a VM (PCI, USB)."""
+
+    slot: str    # hypervisor's device key, e.g. "hostpci0"
+    kind: str    # "pci" or "usb"
+    config: str  # hypervisor's own description of the device
+
+
 class VMConfigDict(TypedDict):
     name: str
-    vmid: int
+    vmid: str
     vcpus: int
     memory: int
     os_type: str
@@ -533,6 +546,14 @@ class VMConfigDict(TypedDict):
     nics: List[VMNICDict]
     description: str
     tags: List[str]
+    # Hardware details not every hypervisor exposes; absent when unknown.
+    os_name: NotRequired[str]            # human-readable guest OS, e.g. "Ubuntu Linux (64-bit)"
+    cpu_type: NotRequired[str]           # e.g. "host", "kvm64"
+    sockets: NotRequired[int]
+    cores_per_socket: NotRequired[int]
+    firmware: NotRequired[str]           # "bios" or "efi"
+    machine: NotRequired[str]            # machine type / virtual hardware version
+    passthrough: NotRequired[List[VMPassthroughDict]]
 
 
 class StorageVolumeDict(TypedDict):
@@ -873,8 +894,8 @@ class NetworkTargetDict(TypedDict):
       is surfaced via ``fixed_vlan_tag`` instead, for display purposes.
     """
 
-    name: str  # Bridge or vnet name, usable directly as NICConfigDict.bridge
-    kind: str  # "bridge" or "vnet"
+    name: str  # Bridge, vnet or port group name, usable directly as NICConfigDict.bridge
+    kind: str  # "bridge", "vnet" or "portgroup" (VMware: VLAN fixed like a vnet's)
     vlan_aware: bool  # True if a NICConfigDict.vlan_tag may be set on top of this target
     fixed_vlan_tag: NotRequired[int | None]  # vnet only: the VLAN ID already baked into it
 

napalm_device_types/nat_vpn.py

@@ -1,8 +1,8 @@
 # -*- coding: utf-8 -*-
 """Address translation and VPN tunnels.
 
-A home gateway does a subset of what a firewall does, and these two readers
-are where the sets overlap exactly.
+A home gateway does a subset of what a firewall does, and these readers are
+where the sets overlap exactly.
 
 Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
 Nothing exists at runtime until a concrete driver implements it, so mixing
@@ -13,7 +13,7 @@ from __future__ import annotations
 
 from typing import Dict, List, TYPE_CHECKING
 
-from napalm_device_types.models import NATTranslationDict, VPNTunnelDict
+from napalm_device_types.models import NATTranslationDict, PortForwardDict, VPNTunnelDict
 
 
 class NatVpnMixin:
@@ -47,6 +47,46 @@ class NatVpnMixin:
             """
             ...
 
+        def get_port_forwards(self) -> List[PortForwardDict]:
+            """
+            Returns the port forwards that let traffic in from the WAN.
+
+            A port forward here means destination NAT on an interface facing
+            the internet: whoever reaches the external port is let through to
+            ``internal_ip``. A redirect between internal networks is
+            destination NAT as well, but it is **not** a port forward and must
+            be left out -- callers read every entry as "this host is reachable
+            from outside". So are rules that only exempt traffic from
+            redirection.
+
+            Each entry contains:
+
+            * name (string) - the rule's description/name
+            * protocol (string) - ``"TCP"`` or ``"UDP"``; a rule for both is
+              two entries. ``"ANY"`` forwards every protocol
+            * external_port (int) - the WAN-side port; the first of a range,
+              ``0`` for every port (a whole host forwarded)
+            * internal_ip (string) - the host the traffic is forwarded to
+            * internal_port (int) - the port on that host
+            * enabled (bool) - whether the rule is currently active
+            * remote_host (string, optional) - restricts the forward to a specific
+              remote source address; empty/absent means "any"
+
+            Example::
+
+                [
+                    {
+                        "name": "Webserver HTTPS",
+                        "protocol": "TCP",
+                        "external_port": 443,
+                        "internal_ip": "192.168.1.10",
+                        "internal_port": 443,
+                        "enabled": True,
+                    }
+                ]
+            """
+            ...
+
         def get_vpn_tunnels(self) -> Dict[str, VPNTunnelDict]:
             """
             Returns the status of VPN tunnels.

napalm_device_types/residential_gateway.py

@@ -6,8 +6,9 @@ wireless access point in a single consumer device (e.g. AVM FritzBox,
 ISP-supplied DSL/cable routers). This base class merges the relevant
 subsets of :class:`~napalm_device_types.firewall.FirewallDriver` and
 :class:`~napalm_device_types.access_point.AccessPointDriver` plus
-gateway-specific operations (WAN status, port forwarding, connected
-hosts).
+gateway-specific operations (WAN status, connected hosts). Port
+forwarding is shared with firewalls, in
+:class:`~napalm_device_types.nat_vpn.NatVpnMixin`.
 
 Usage::
 
@@ -25,7 +26,6 @@ from napalm_device_types.health_metrics import HealthMetricsMixin
 from napalm_device_types.dhcp import DhcpServerMixin
 from napalm_device_types.models import (
     HostDict,
-    PortForwardDict,
     RadioStatusDict,
     SSIDDict,
     WANStatusDict,
@@ -85,36 +85,6 @@ class ResidentialGatewayDriver(NatVpnMixin, HealthMetricsMixin, DhcpServerMixin,
             """
             ...
 
-        def get_port_forwards(self) -> List[PortForwardDict]:
-            """
-            Returns the configured port forwarding (port mapping) rules.
-
-            Each entry contains:
-
-            * name (string) - the rule's description/name
-            * protocol (string) - ``"TCP"`` or ``"UDP"``
-            * external_port (int) - the WAN-side port
-            * internal_ip (string) - the LAN host the traffic is forwarded to
-            * internal_port (int) - the LAN-side port
-            * enabled (bool) - whether the rule is currently active
-            * remote_host (string, optional) - restricts the forward to a specific
-              remote source address; empty/absent means "any"
-
-            Example::
-
-                [
-                    {
-                        "name": "Webserver HTTPS",
-                        "protocol": "TCP",
-                        "external_port": 443,
-                        "internal_ip": "192.168.1.10",
-                        "internal_port": 443,
-                        "enabled": True,
-                    }
-                ]
-            """
-            ...
-
         def get_hosts(self) -> List[HostDict]:
             """
             Returns the list of hosts known to the gateway (LAN clients).

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "1.0.0"
+version = "2.0.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_host_reboot.py

@@ -0,0 +1,29 @@
+"""reboot_host: the contract for restarting the device itself.
+
+netOrk used to restart a host by sending ``/sbin/reboot`` through a driver's
+private ``_send_command``. A driver that talks to an API instead has no such
+method, and the caller swallowed the resulting AttributeError, so the reboot
+"succeeded" without happening. A declared contract lets netOrk ask first.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+from napalm_device_types import DeviceTypeDriver, HostRebootMixin
+
+
+def test_every_device_type_driver_carries_the_declaration():
+    assert issubclass(DeviceTypeDriver, HostRebootMixin)
+
+
+def test_declared_not_implemented():
+    """hasattr is netOrk's capability probe; only a driver that implements it
+    may answer True."""
+    assert not hasattr(DeviceTypeDriver, "reboot_host")
+
+
+def test_declaration_documents_the_contract():
+    source = inspect.getsource(HostRebootMixin)
+    assert "def reboot_host(self) -> None" in source
+    assert "RuntimeError" in source

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

tests/test_port_forwards.py

@@ -0,0 +1,39 @@
+"""get_port_forwards: what the WAN side may reach inside, on any gateway.
+
+The reader used to be declared on ``ResidentialGatewayDriver`` only, as if a
+port forward were a home-router feature. A firewall forwards ports just the
+same -- OPNsense calls it destination NAT -- and the two consumers that ask
+(is this host reachable from the internet, which CVEs are exposed) need the
+answer from both. The declaration therefore lives where the two roles overlap,
+next to the NAT translations reader.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+from napalm_device_types import FirewallDriver, ResidentialGatewayDriver
+from napalm_device_types.nat_vpn import NatVpnMixin
+
+
+def test_a_firewall_and_a_gateway_share_the_declaration():
+    assert issubclass(FirewallDriver, NatVpnMixin)
+    assert issubclass(ResidentialGatewayDriver, NatVpnMixin)
+    assert "def get_port_forwards(self) -> List[PortForwardDict]" in inspect.getsource(NatVpnMixin)
+
+
+def test_it_is_declared_once():
+    assert "def get_port_forwards" not in inspect.getsource(ResidentialGatewayDriver)
+
+
+def test_absent_until_a_driver_implements_it():
+    assert not hasattr(FirewallDriver, "get_port_forwards")
+    assert not hasattr(ResidentialGatewayDriver, "get_port_forwards")
+
+
+def test_the_contract_says_what_counts():
+    """A redirect between two internal networks is destination NAT too, and
+    would make an internal host look reachable from the internet."""
+    source = inspect.getsource(NatVpnMixin)
+    assert "from the WAN" in source
+    assert "between internal networks" in source

tests/test_vm_identifiers.py

@@ -0,0 +1,70 @@
+"""A VM's ``vmid`` is a string on every hypervisor.
+
+Proxmox numbers its guests, but VMware identifies them by UUID or MoRef
+(``"vm-42"``). An ``int`` in the contract forced netOrk to call ``int()`` on
+whatever came back, which cannot represent the second kind at all. The
+provisioning dicts already carried ``vmid`` as a string; these pin the
+read-side dicts to the same type.
+"""
+
+from __future__ import annotations
+
+from typing import get_type_hints
+
+import pytest
+from napalm_device_types.models import (
+    VMConfigDict,
+    VMDict,
+    VMProvisionResultDict,
+)
+
+
+@pytest.mark.parametrize("model", [VMDict, VMConfigDict, VMProvisionResultDict])
+def test_vmid_is_a_string(model):
+    assert get_type_hints(model)["vmid"] is str
+
+
+class TestVMConfigCarriesWhatAHardwareViewShows:
+    """netOrk's VM hardware view used to read Proxmox's raw config through the
+    driver's private API. The contract has to carry those details so a second
+    hypervisor can fill the same view -- optionally, since not every platform
+    has every one of them."""
+
+    OPTIONAL = {
+        "os_name",
+        "cpu_type",
+        "sockets",
+        "cores_per_socket",
+        "firmware",
+        "machine",
+        "passthrough",
+    }
+
+    def test_hardware_details_are_optional_fields(self):
+        assert self.OPTIONAL <= VMConfigDict.__optional_keys__
+
+    def test_contract_core_stays_required(self):
+        assert "vmid" in VMConfigDict.__required_keys__
+        assert "disks" in VMConfigDict.__required_keys__
+
+    def test_passthrough_entry_shape(self):
+        from napalm_device_types.models import VMPassthroughDict
+
+        assert get_type_hints(VMPassthroughDict) == {"slot": str, "kind": str, "config": str}
+
+
+class TestGuestAgentDeclaration:
+    """netOrk's cloud-init installed qemu-guest-agent on every new VM. A VMware
+    guest reports its IP through open-vm-tools instead; the hypervisor says
+    which, and netOrk stops hard-coding one of them."""
+
+    def test_default_is_qemu_guest_agent(self):
+        from napalm_device_types import HypervisorDriver
+
+        assert HypervisorDriver.GUEST_AGENT_PACKAGES == ("qemu-guest-agent",)
+        assert HypervisorDriver.GUEST_AGENT_RUNCMD == ("systemctl enable --now qemu-guest-agent",)
+
+    def test_attributes_are_not_methods(self):
+        from napalm_device_types import HypervisorDriver
+
+        assert not callable(vars(HypervisorDriver)["GUEST_AGENT_PACKAGES"])