feat: list systemd services in one round trip and control them, once for every driver

Listing a host's services and starting or stopping one is the same on every
host that runs systemd, so the command, its parse, the check of a unit name
and the reading of an action's exit status live here once, and a driver
supplies only the transport (napalm-linux#7, napalm-proxmox#6):

- SYSTEMD_SERVICES_COMMAND: one read-only POSIX sh line. list-unit-files, then
  one systemctl show over every loaded service unit (Id, Names, LoadState,
  ActiveState, SubState, UnitFileState, MainPID), and is-enabled only for
  generated units, whose boot state lives in a SysV script's rc links. Framed;
  [no-systemd] when /run/systemd/system is missing. Replaces an is-enabled and
  a show per unit: 0.8 s instead of 6 s on a 180-unit Ubuntu host.
- parse_systemd_services(): loaded units except not-found, plus installed unit
  files that are not loaded; no templates, no aliases (also not the ones older
  systemd lists as "enabled"). enabled = UnitFileState enabled or
  enabled-runtime, read from systemctl show and never from list-unit-files'
  second column, which has had a preset column after it since systemd 245.
  A report whose end is missing raises ValueError, so a list cut short never
  reads as services that went away; a host without systemd raises
  SystemdUnavailable, a NotImplementedError, so a driver can fall back.
- unit_name() / service_action_command() / parse_action_result(): template
  instances, dots, colons and \xHH escapes accepted; a bare template, a leading
  "-" and anything a shell reads refused. The action runs as
  "timeout 45 systemctl --no-ask-password <action> -- <unit>.service" with its
  exit status printed after it; only that status decides, 124 is not called
  done, and terminal colour codes are dropped. The marker also keeps the output
  from ever being empty, which a transport that retries on an empty answer
  would take as a reason to run the action twice.
- SystemdServicesMixin, in the template form: get_services() and
  manage_service() are concrete, _run_service_command(command, *, privileged,
  timeout) is the driver's hook. Mixed in by the drivers whose host runs
  systemd, not by OSDriver.

Version 2.2.0.

README.md

@@ -22,6 +22,152 @@ NAPALM's `NetworkDriver` defines a common interface for all network devices. In
 
 `napalm-device-types` sits in between: it adds one well-typed layer of abstract methods per device category, so every driver for the same category exposes the same interface.
 
+## Roles: what a device *is*
+
+A device is often several things at once. A QNAP NAS runs VMs on a Linux userland; an
+OpenMediaVault box is a NAS built on Debian. So a driver inherits **one role base per
+role its device fills**, and **the order it lists them in is the ranking**:
+
+```python
+class QnapQtsDriver(StorageDriver, HypervisorDriver, LinuxDriver):
+    ...   # primary_role_of(...) == "storage"
+```
+
+`roles_of(cls)`, `role_keys_of(cls)` and `primary_role_of(cls)` read that back. Nothing
+restates the ranking: there is no precedence table and no attribute to override.
+
+### Role bases declare; they never implement
+
+**A role base must not contain a single runtime method — not even a
+`NotImplementedError` placeholder.** Its methods are declared under `if TYPE_CHECKING`:
+
+```python
+class StorageDriver(DeviceTypeDriver):
+    """Contract, not code."""
+    ROLE: str = "storage"
+    TYPE_LABEL: str = "Storage"
+
+    if TYPE_CHECKING:                      # nothing exists at runtime
+        def get_disks(self) -> List[PhysicalDiskDict]: ...
+```
+
+This is not a style preference. A placeholder on a base class is not neutral under
+multiple inheritance: it wins the MRO against a sibling base's *working* implementation
+and silently replaces it. Adding one stub to a base is therefore a breaking change for
+every driver that mixes that base with another. It happened three times here before the
+rule existed, and each time the fix was hand-written forwarding methods in the driver.
+
+Two things follow, and both are improvements:
+
+- **`hasattr` is truthful again.** A method exists on a driver class exactly when that
+  driver implemented it, which is how netOrk asks "can this driver list disks".
+- **A method that was never implemented raises `AttributeError`, not
+  `NotImplementedError`.** Ask before calling.
+
+## Function classes: what a device *can do*
+
+Behaviour shared across roles lives in a function class, exactly once, and a role base
+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.
+
+`KernelFactsMixin` (`get_kernel_facts`) is the exception that is mixed in by a driver
+rather than by a role base: what a Linux kernel has built and loaded is read the same way
+everywhere, so the command and its parse are concrete here and a driver supplies only
+`_run_kernel_facts_command`. `OSDriver` does not carry it — a Windows host is an OS driver
+too, and `hasattr(driver, "get_kernel_facts")` has to stay truthful.
+
+`SystemdServicesMixin` (`get_services`, `manage_service`) is mixed in the same way, by
+the drivers whose host runs systemd. Listing the services, checking a unit name and
+reading an action's exit status are the same on every such host, so they are concrete
+here, and a driver supplies only `_run_service_command(command, *, privileged, timeout)`
+— how a command reaches its host and how it gains root there. The listing is one round
+trip (`list-unit-files` plus one `systemctl show` over every loaded unit) instead of an
+`is-enabled` and a `show` per unit. A host without systemd raises `SystemdUnavailable`,
+a `NotImplementedError`, so a driver can fall back to another init system.
+
+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
+genuinely does work* on the result: normalising, sorting, validating, or orchestrating
+several hooks. `ConfigLifecycleMixin.compare_config` over `_get_running_config` is the
+model. Where the base would only pass the call through, declare the method directly;
+two names for one pass-through is ceremony, not design.
+
+NAPALM's own getters (`get_facts`, `get_interfaces`, `ping`, `get_config`) are never
+wrapped in a template — they belong to NAPALM, and code outside this repo relies on
+their contract.
+
+## Design principle: generic vs. device-specific logic
+
+When adding behavior to a device-type base class, split it along one line: **would
+this exact logic work unchanged for a different vendor's driver of the same
+device-type, if that driver only implemented the same abstract methods?**
+
+- If yes, it's generic — implement it once as a **concrete** method on the
+  device-type base class (here, in this repo).
+- If no — it talks to the device itself (a specific REST endpoint, a CLI command,
+  a vendor-specific payload format) — it belongs in the concrete driver as the
+  implementation of an **abstract** method the base class declares.
+
+Concretely: matching/comparison/reconciliation algorithms, orchestration flows, and
+generic data shapes belong here. Only the actual device communication belongs in
+`vendor/napalm-<name>`.
+
+**Worked example — firewall rule diff/apply** (`FirewallDriver`):
+
+```python
+class FirewallDriver(DeviceTypeDriver):
+    # Abstract — every driver implements its own device communication.
+    def get_firewall_rules(self) -> List[FirewallRuleDict]: raise NotImplementedError
+    def apply_firewall_rule(self, rule: FirewallRuleDict, *, uuid: Optional[str] = None) -> Dict[str, Any]: raise NotImplementedError
+    def commit_firewall_rules(self) -> Dict[str, Any]: raise NotImplementedError
+
+    # Concrete — the matching/comparison/orchestration algorithm is identical
+    # for every firewall vendor, so it lives here once.
+    def diff_firewall_rules(self, desired: List[FirewallRuleDict]) -> FirewallRuleDiffDict:
+        ...  # matches self.get_firewall_rules() against `desired` by description
+
+    def apply_firewall_ruleset(self, desired: List[FirewallRuleDict]):
+        ...  # computes the diff, calls apply_firewall_rule() per change, commits
+```
+
+The same split applies to `DhcpServerMixin`: `get_dhcp_reservations`/
+`apply_dhcp_reservation`/`commit_dhcp_reservations` are abstract (Kea REST on
+OPNsense, dnsmasq/odhcpd UCI on OpenWrt), while `diff_dhcp_reservations` and
+`apply_dhcp_reservationset` are concrete — matching by normalised MAC and the
+apply-then-commit orchestration are identical for every DHCP server.
+
+A new driver (FortiGate, pfSense, …) gets `diff_firewall_rules`/
+`apply_firewall_ruleset` for free the moment it implements the three abstract
+methods — it never needs to reimplement the reconciliation logic itself.
+
+**Second worked example — ping sweeps** (`PingSweepMixin`, mixed into
+`DeviceTypeDriver`, so *every* device-type driver has it):
+
+```python
+class PingSweepMixin:
+    # Concrete — the loop, the reply parsing, the target cap and the progress
+    # reporting are the same for every device that can ping at all.
+    def ping_sweep(self, destinations, *, count=1, timeout=1, …) -> PingSweepResultDict:
+        ...  # calls NAPALM's standard ping() once per destination
+```
+
+A driver becomes a usable sweep source the moment it implements NAPALM's
+`ping()` — nothing else is required, and `driver_supports_ping(cls)` reports
+whether it did (introspection, not a hand-maintained list). A driver whose
+device offers something genuinely faster overrides `ping_sweep` and keeps the
+return shape: `napalm-opnsense` starts a batch of ping jobs over the
+diagnostics API, waits once for all of them, and reads every result with a
+single request — a per-host loop would be unusable there.
+
+This mirrors a similar split already documented on the consumer side, in NetOrk's
+`docs/ARCHITECTURE.md` ("Device Warnings — Trennung von Erkennung und
+Präsentation"): drivers return raw signals, the higher layer gives them meaning.
+Same shape of separation, different axis — device-specific vs. generic here,
+detection vs. presentation there.
+
 ## Installation
 
 ```bash
@@ -40,6 +186,13 @@ Requires Python ≥ 3.9 and NAPALM ≥ 4.0.
 | `HypervisorDriver` | Hypervisors & virtualisation platforms | Proxmox VE, VMware ESXi, KVM/libvirt |
 | `OSDriver` | General-purpose operating systems | Linux, BSD, macOS |
 | `StorageDriver` | Storage appliances & NAS/SAN | TrueNAS, Synology DSM, QNAP QTS |
+| `ResidentialGatewayDriver` | Router + firewall + AP in one box | OpenWrt, FritzBox |
+
+Mixins mixed into the classes above rather than used on their own:
+`ConfigLifecycleMixin` (config load/compare/commit/rollback), `PingSweepMixin`
+(subnet sweeps), and `DhcpServerMixin` (static DHCP reservations — mixed into
+`FirewallDriver` and `ResidentialGatewayDriver`, since both commonly run the
+DHCP server for their networks).
 
 ## Usage
 
@@ -89,6 +242,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
@@ -102,7 +260,13 @@ 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):
+        ...
+
+    def get_vm_cpu_types(self):
+        # optional — return List[VMCpuTypeDict]: the CPU models a new VM may get
+        # on this node, each with its cpuinfo flags and whether the node can run
+        # it; the name goes to create_vm_from_cloud_init(cpu_type=...)
         ...
 ```
 

napalm_device_types/__init__.py

@@ -4,16 +4,22 @@ napalm-device-types
 
 Abstract device-type base classes for NAPALM drivers.
 
-Instead of inheriting directly from ``napalm.base.NetworkDriver``, a driver
-can inherit from one of the device-type classes defined here to gain
-type-specific abstract methods and a clearer contract::
+A driver inherits one base per role its device fills, and **the order of those
+bases is the ranking** -- the first one is what netOrk shows as the device's
+class::
 
-    from napalm_device_types import AccessPointDriver
+    from napalm_device_types import StorageDriver, HypervisorDriver
 
-    class OpenWrtDriver(AccessPointDriver):
-        ...
+    class QnapQtsDriver(StorageDriver, HypervisorDriver, LinuxDriver):
+        ...            # a NAS that also runs VMs on a Linux userland
 
-Available base classes:
+That works because a role base *declares* its methods (under ``if
+TYPE_CHECKING``) and implements none of them. Nothing exists at runtime until a
+concrete driver provides it, so no base can shadow a working implementation
+inherited from a sibling, and ``hasattr`` is a truthful answer to "can this
+driver do X".
+
+Role bases -- what a device *is*:
 
 * :class:`~napalm_device_types.access_point.AccessPointDriver`
 * :class:`~napalm_device_types.switch.SwitchDriver`
@@ -21,21 +27,61 @@ Available base classes:
 * :class:`~napalm_device_types.hypervisor.HypervisorDriver`
 * :class:`~napalm_device_types.os.OSDriver`
 * :class:`~napalm_device_types.storage.StorageDriver`
+* :class:`~napalm_device_types.phone.PhoneDriver`
+* :class:`~napalm_device_types.media.MediaDriver`
 * :class:`~napalm_device_types.residential_gateway.ResidentialGatewayDriver`
 
-Also provided:
+Function classes -- what a device *can do*. Shared behaviour lives here once
+instead of being restated on every role that happens to need it:
 
-* :class:`~napalm_device_types.config_lifecycle.ConfigLifecycleMixin` --
-  stand-alone mixin to reduce duplication of config lifecycle methods across
-  drivers.
+* :class:`~napalm_device_types.config_lifecycle.ConfigLifecycleMixin`
+* :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.kernel.KernelFactsMixin`
+* :class:`~napalm_device_types.mac_acl.MacAclMixin`
+* :class:`~napalm_device_types.nat_vpn.NatVpnMixin`
+* :class:`~napalm_device_types.packages.PackageManagementMixin`
+* :class:`~napalm_device_types.ping_sweep.PingSweepMixin`
+* :class:`~napalm_device_types.services.ServiceControlMixin`
+* :class:`~napalm_device_types.systemd.SystemdServicesMixin`
+* :class:`~napalm_device_types.updates.UpdateMixin`
+
+Introspection -- :func:`~napalm_device_types.roles.roles_of`,
+:func:`~napalm_device_types.roles.role_keys_of` and
+:func:`~napalm_device_types.roles.primary_role_of`.
 """
 
 from napalm_device_types.base import DeviceTypeDriver, FingerprintRule, PortSpec
 from napalm_device_types.access_point import AccessPointDriver
 from napalm_device_types.config_lifecycle import ConfigLifecycleMixin
+from napalm_device_types.dhcp import DhcpServerMixin, normalize_cidr, normalize_mac
 from napalm_device_types.firewall import FirewallDriver
 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.kernel import KERNEL_FACTS_COMMAND, KernelFactsMixin, parse_kernel_facts
+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
+from napalm_device_types.packages import PackageManagementMixin
+from napalm_device_types.phone import PhoneDriver
+from napalm_device_types.ping_sweep import PingSweepMixin, driver_supports_ping
+from napalm_device_types.roles import primary_role_of, role_keys_of, roles_of
+from napalm_device_types.services import ServiceControlMixin
+from napalm_device_types.systemd import (
+    SYSTEMD_SERVICES_COMMAND,
+    SystemdServicesMixin,
+    SystemdUnavailable,
+    parse_systemd_services,
+)
+from napalm_device_types.updates import UpdateMixin
 from napalm_device_types.residential_gateway import ResidentialGatewayDriver
 from napalm_device_types.storage import StorageDriver
 from napalm_device_types.switch import SwitchDriver
@@ -44,12 +90,39 @@ __all__ = [
     "AccessPointDriver",
     "ConfigLifecycleMixin",
     "DeviceTypeDriver",
+    "DhcpServerMixin",
     "FingerprintRule",
     "FirewallDriver",
+    "FirewallRuleMixin",
+    "HealthMetricsMixin",
+    "HostRebootMixin",
     "HypervisorDriver",
+    "InterfaceFilterMixin",
+    "MacAclMixin",
+    "MediaDriver",
+    "NatVpnMixin",
     "OSDriver",
+    "PackageManagementMixin",
+    "KernelFactsMixin",
+    "KERNEL_FACTS_COMMAND",
+    "parse_kernel_facts",
+    "PhoneDriver",
+    "PingSweepMixin",
     "PortSpec",
     "ResidentialGatewayDriver",
+    "ServiceControlMixin",
     "StorageDriver",
     "SwitchDriver",
+    "SYSTEMD_SERVICES_COMMAND",
+    "SystemdServicesMixin",
+    "SystemdUnavailable",
+    "parse_systemd_services",
+    "UpdateMixin",
+    "add_lag_interfaces",
+    "driver_supports_ping",
+    "normalize_cidr",
+    "normalize_mac",
+    "primary_role_of",
+    "role_keys_of",
+    "roles_of",
 ]

napalm_device_types/access_point.py

@@ -10,30 +10,25 @@ Usage::
             ...
 """
 
-from typing import Any, Dict, List
+from typing import Any, Dict, List, TYPE_CHECKING
 from napalm_device_types.base import DeviceTypeDriver
-from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
+from napalm_device_types.mac_acl import MacAclMixin
+from napalm_device_types.updates import UpdateMixin
+from napalm_device_types.services import ServiceControlMixin
+from napalm_device_types.packages import PackageManagementMixin
+from napalm_device_types.health_metrics import HealthMetricsMixin
+from napalm_device_types.interface_filter import InterfaceFilterMixin
 from napalm_device_types.models import (
-    ChannelScanEntryDict,
     Dot1XConfigDict,
     FastTransitionConfigDict,
-    HealthMetricsDict,
-    MACACLDict,
     MeshConfigDict,
     MeshPeerDict,
-    PackageDict,
-    RadioStatusDict,
-    ServiceDict,
     SSIDBridgeDict,
-    SSIDDict,
-    UpdateDict,
-    WirelessClientDict,
     WirelessConfigDict,
 )
 
 
-class AccessPointDriver(DeviceTypeDriver):
-    TYPE_LABEL: str = "Access Point"
+class AccessPointDriver(MacAclMixin, UpdateMixin, ServiceControlMixin, PackageManagementMixin, HealthMetricsMixin, InterfaceFilterMixin, DeviceTypeDriver):
     """
     Abstract intermediate driver for wireless access points.
 
@@ -41,33 +36,16 @@ class AccessPointDriver(DeviceTypeDriver):
     access-point-specific operations that concrete drivers must implement.
     """
 
-    # Interfaces that carry no operational meaning on an access point and
-    # should be excluded from get_interfaces() / get_facts() interface_list.
-    _EXCLUDED_INTERFACES: frozenset = frozenset({"lo"})
+    #: Stable key netOrk exposes as ``device_class``. The order in which a
+    #: driver lists its role bases is the ranking; see
+    #: :func:`napalm_device_types.roles.primary_role_of`.
+    ROLE: str = "access_point"
+    TYPE_LABEL: str = "Access Point"
+
 
-    # Interface name *prefixes* to exclude (e.g. Linux phy* are raw radio
-    # devices and have no IP/Ethernet significance at the AP level).
-    _EXCLUDED_INTERFACE_PREFIXES: tuple = ("phy",)
 
-    _SNMP_SKIP_IF = IF_SKIP_DEFAULT
-    _SNMP_TX_ERR_IS_DROP: bool = False
 
-    @classmethod
-    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
-        return await collect_ucd_metrics(
-            snmp_get, snmp_walk,
-            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
-            if_skip=cls._SNMP_SKIP_IF,
-        )
 
-    def _filter_interfaces(self, interfaces: Dict[str, Any]) -> Dict[str, Any]:
-        """Remove loopback and radio-device (phy*) interfaces from an interface dict."""
-        return {
-            name: data
-            for name, data in interfaces.items()
-            if name not in self._EXCLUDED_INTERFACES
-            and not name.startswith(self._EXCLUDED_INTERFACE_PREFIXES)
-        }
 
     # AP-specific methods (get_wireless_clients, get_ssids, get_radio_status,
     # get_interfaces, get_vlans, …) are intentionally NOT defined here.
@@ -77,451 +55,225 @@ class AccessPointDriver(DeviceTypeDriver):
     # all standard NAPALM methods, so AccessPointDriver must come LAST to avoid
     # shadowing the mixin implementations.
 
-    def get_wireless_config(self) -> WirelessConfigDict:
-        """
-        Returns global wireless configuration parameters that apply across
-        all radios and SSIDs.
+    if TYPE_CHECKING:
 
-        * country_code (string) - ISO 3166-1 alpha-2 country code (e.g. ``"DE"``)
-        * regulatory_domain (string) - regulatory domain string (e.g. ``"ETSI"``)
-        * beacon_interval (int) - beacon interval in TUs (default 100)
-        * dtim_period (int) - DTIM period (default 2)
-        * rts_threshold (int) - RTS/CTS threshold in bytes (2347 = disabled)
-        * fragmentation_threshold (int) - fragmentation threshold in bytes
-        * short_preamble (bool) - whether short preamble is enabled
-        * wmm_enabled (bool) - whether WMM/QoS is enabled
+        def get_wireless_config(self) -> WirelessConfigDict:
+            """
+            Returns global wireless configuration parameters that apply across
+            all radios and SSIDs.
 
-        Example::
+            * country_code (string) - ISO 3166-1 alpha-2 country code (e.g. ``"DE"``)
+            * regulatory_domain (string) - regulatory domain string (e.g. ``"ETSI"``)
+            * beacon_interval (int) - beacon interval in TUs (default 100)
+            * dtim_period (int) - DTIM period (default 2)
+            * rts_threshold (int) - RTS/CTS threshold in bytes (2347 = disabled)
+            * fragmentation_threshold (int) - fragmentation threshold in bytes
+            * short_preamble (bool) - whether short preamble is enabled
+            * wmm_enabled (bool) - whether WMM/QoS is enabled
 
-            {
-                "country_code": "DE",
-                "regulatory_domain": "ETSI",
-                "beacon_interval": 100,
-                "dtim_period": 2,
-                "rts_threshold": 2347,
-                "fragmentation_threshold": 2346,
-                "short_preamble": True,
-                "wmm_enabled": True,
-            }
-        """
-        raise NotImplementedError
+            Example::
 
-    def get_fast_transition_config(self) -> Dict[str, FastTransitionConfigDict]:
-        """
-        Returns the 802.11r Fast BSS Transition (FT) configuration per SSID.
-
-        Keys are SSID names.  Each value contains:
-
-        * enabled (bool) - whether FT is active on this SSID
-        * ssid (string) - SSID name (repeated for convenience)
-        * mobility_domain (string) - 4-hex-digit Mobility Domain ID (MDID)
-        * reassociation_deadline (int) - FT reassociation deadline in TUs
-        * r0_key_lifetime (int) - PMK-R0 key lifetime in minutes
-        * r1_key_holder (string) - R1 Key Holder identifier (MAC-like string)
-        * pmk_r1_push (bool) - whether PMK-R1 is proactively pushed to neighbours
-        * over_ds (bool) - whether FT over DS (instead of FT over air) is used
-
-        Example::
-
-            {
-                "CorpWiFi": {
-                    "enabled": True,
-                    "ssid": "CorpWiFi",
-                    "mobility_domain": "a1b2",
-                    "reassociation_deadline": 1000,
-                    "r0_key_lifetime": 10000,
-                    "r1_key_holder": "00:11:22:33:44:55",
-                    "pmk_r1_push": True,
-                    "over_ds": False,
-                }
-            }
-        """
-        raise NotImplementedError
-
-    def get_mesh_config(self) -> Dict[str, MeshConfigDict]:
-        """
-        Returns the 802.11s mesh configuration per mesh interface.
-
-        Keys are mesh interface names (e.g. ``"mesh0"``).  Each value contains:
-
-        * enabled (bool) - whether the mesh interface is active
-        * radio (string) - underlying radio (e.g. ``"radio0"``)
-        * mesh_id (string) - 802.11s Mesh ID (analogous to SSID)
-        * path_metric (string) - path selection metric, e.g. ``"airtime"`` or ``"hopcount"``
-        * gate_announcements (bool) - whether gate announcements (GANN) are sent
-        * is_gate (bool) - whether this node acts as a mesh gate to the DS
-        * encryption (string) - e.g. ``"SAE"``, ``"open"``
-
-        Example::
-
-            {
-                "mesh0": {
-                    "enabled": True,
-                    "radio": "radio1",
-                    "mesh_id": "office-mesh",
-                    "path_metric": "airtime",
-                    "gate_announcements": True,
-                    "is_gate": True,
-                    "encryption": "SAE",
-                }
-            }
-        """
-        raise NotImplementedError
-
-    def get_mesh_peers(self) -> List[MeshPeerDict]:
-        """
-        Returns a list of currently active 802.11s mesh peers.
-
-        Each entry contains:
-
-        * mac (string) - peer MAC address
-        * radio (string) - radio on which the peering was established
-        * signal (int) - received signal strength in dBm
-        * tx_rate (float) - TX bitrate to peer in Mbit/s
-        * rx_rate (float) - RX bitrate from peer in Mbit/s
-        * uptime (int) - peering duration in seconds
-        * hop_count (int) - number of hops to the mesh gate (0 = this node is the gate)
-
-        Example::
-
-            [
                 {
-                    "mac": "AA:BB:CC:DD:EE:01",
-                    "radio": "radio1",
-                    "signal": -58,
-                    "tx_rate": 300.0,
-                    "rx_rate": 270.0,
-                    "uptime": 7200,
-                    "hop_count": 1,
+                    "country_code": "DE",
+                    "regulatory_domain": "ETSI",
+                    "beacon_interval": 100,
+                    "dtim_period": 2,
+                    "rts_threshold": 2347,
+                    "fragmentation_threshold": 2346,
+                    "short_preamble": True,
+                    "wmm_enabled": True,
                 }
-            ]
-        """
-        raise NotImplementedError
+            """
+            ...
 
-    def get_ssid_bridge_config(self) -> Dict[str, SSIDBridgeDict]:
-        """
-        Returns the Layer-2 bridging configuration for each SSID, i.e. which
-        bridge interface and VLAN each SSID is mapped to.
+        def get_fast_transition_config(self) -> Dict[str, FastTransitionConfigDict]:
+            """
+            Returns the 802.11r Fast BSS Transition (FT) configuration per SSID.
 
-        Keys are SSID names.  Each value contains:
+            Keys are SSID names.  Each value contains:
 
-        * ssid (string) - SSID name (repeated for convenience)
-        * bridge (string) - bridge interface the VAP is attached to (e.g. ``"br-lan"``, ``"br-guest"``)
-        * vlan_id (int) - 802.1Q VLAN ID (0 = untagged / no VLAN separation)
-        * tagged (bool) - whether traffic is 802.1Q-tagged on the uplink port
-        * client_isolation (bool) - whether clients on this SSID are isolated from each other
+            * enabled (bool) - whether FT is active on this SSID
+            * ssid (string) - SSID name (repeated for convenience)
+            * mobility_domain (string) - 4-hex-digit Mobility Domain ID (MDID)
+            * reassociation_deadline (int) - FT reassociation deadline in TUs
+            * r0_key_lifetime (int) - PMK-R0 key lifetime in minutes
+            * r1_key_holder (string) - R1 Key Holder identifier (MAC-like string)
+            * pmk_r1_push (bool) - whether PMK-R1 is proactively pushed to neighbours
+            * over_ds (bool) - whether FT over DS (instead of FT over air) is used
 
-        Example::
+            Example::
 
-            {
-                "CorpWiFi": {
-                    "ssid": "CorpWiFi",
-                    "bridge": "br-corp",
-                    "vlan_id": 10,
-                    "tagged": True,
-                    "client_isolation": False,
-                },
-                "GuestNet": {
-                    "ssid": "GuestNet",
-                    "bridge": "br-guest",
-                    "vlan_id": 20,
-                    "tagged": True,
-                    "client_isolation": True,
-                },
-                "IoT": {
-                    "ssid": "IoT",
-                    "bridge": "br-iot",
-                    "vlan_id": 30,
-                    "tagged": True,
-                    "client_isolation": True,
-                },
-            }
-        """
-        raise NotImplementedError
+                {
+                    "CorpWiFi": {
+                        "enabled": True,
+                        "ssid": "CorpWiFi",
+                        "mobility_domain": "a1b2",
+                        "reassociation_deadline": 1000,
+                        "r0_key_lifetime": 10000,
+                        "r1_key_holder": "00:11:22:33:44:55",
+                        "pmk_r1_push": True,
+                        "over_ds": False,
+                    }
+                }
+            """
+            ...
 
-    def get_mac_acl(self) -> Dict[str, MACACLDict]:
-        """
-        Returns the MAC-address-based access control lists configured per SSID.
+        def get_mesh_config(self) -> Dict[str, MeshConfigDict]:
+            """
+            Returns the 802.11s mesh configuration per mesh interface.
 
-        Keys are SSID names.  Each value contains:
+            Keys are mesh interface names (e.g. ``"mesh0"``).  Each value contains:
 
-        * ssid (string) - SSID name (repeated for convenience)
-        * policy (string) - ACL mode:
+            * enabled (bool) - whether the mesh interface is active
+            * radio (string) - underlying radio (e.g. ``"radio0"``)
+            * mesh_id (string) - 802.11s Mesh ID (analogous to SSID)
+            * path_metric (string) - path selection metric, e.g. ``"airtime"`` or ``"hopcount"``
+            * gate_announcements (bool) - whether gate announcements (GANN) are sent
+            * is_gate (bool) - whether this node acts as a mesh gate to the DS
+            * encryption (string) - e.g. ``"SAE"``, ``"open"``
 
-          * ``"allow"`` – whitelist: only listed MACs may associate
-          * ``"deny"``  – blacklist: listed MACs are blocked
-          * ``"disabled"`` – no MAC filtering active
+            Example::
 
-        * entries (list) - ACL entries, each with:
+                {
+                    "mesh0": {
+                        "enabled": True,
+                        "radio": "radio1",
+                        "mesh_id": "office-mesh",
+                        "path_metric": "airtime",
+                        "gate_announcements": True,
+                        "is_gate": True,
+                        "encryption": "SAE",
+                    }
+                }
+            """
+            ...
 
-          * mac (string) - MAC address (normalised, colon-separated)
-          * action (string) - ``"allow"`` or ``"deny"``
-          * description (string) - optional human-readable label
+        def get_mesh_peers(self) -> List[MeshPeerDict]:
+            """
+            Returns a list of currently active 802.11s mesh peers.
 
-        Example::
+            Each entry contains:
 
-            {
-                "CorpWiFi": {
-                    "name": "CorpWiFi",
-                    "policy": "allow",
-                    "entries": [
-                        {"mac": "AA:BB:CC:DD:EE:01", "action": "allow", "description": "CEO-Laptop"},
-                        {"mac": "AA:BB:CC:DD:EE:02", "action": "allow", "description": "CFO-Laptop"},
-                    ],
-                },
-                "GuestNet": {
-                    "name": "GuestNet",
-                    "policy": "deny",
-                    "entries": [
-                        {"mac": "DE:AD:BE:EF:00:01", "action": "deny", "description": "blocked device"},
-                    ],
-                },
-            }
-        """
-        raise NotImplementedError
+            * mac (string) - peer MAC address
+            * radio (string) - radio on which the peering was established
+            * signal (int) - received signal strength in dBm
+            * tx_rate (float) - TX bitrate to peer in Mbit/s
+            * rx_rate (float) - RX bitrate from peer in Mbit/s
+            * uptime (int) - peering duration in seconds
+            * hop_count (int) - number of hops to the mesh gate (0 = this node is the gate)
 
-    def get_dot1x_config(self) -> Dict[str, Dot1XConfigDict]:
-        """
-        Returns the 802.1X / WPA-Enterprise (RADIUS) configuration per SSID.
+            Example::
 
-        Keys are SSID names.  Each value contains:
+                [
+                    {
+                        "mac": "AA:BB:CC:DD:EE:01",
+                        "radio": "radio1",
+                        "signal": -58,
+                        "tx_rate": 300.0,
+                        "rx_rate": 270.0,
+                        "uptime": 7200,
+                        "hop_count": 1,
+                    }
+                ]
+            """
+            ...
 
-        * enabled (bool) - whether 802.1X authentication is active on this SSID
-        * ssid (string) - SSID name (repeated for convenience)
-        * auth_server (dict) - RADIUS authentication server:
+        def get_ssid_bridge_config(self) -> Dict[str, SSIDBridgeDict]:
+            """
+            Returns the Layer-2 bridging configuration for each SSID, i.e. which
+            bridge interface and VLAN each SSID is mapped to.
 
-          * host (string) - IP or FQDN of the RADIUS server
-          * port (int) - UDP port (default 1812)
-          * timeout (int) - request timeout in seconds
-          * retries (int) - number of retransmissions
+            Keys are SSID names.  Each value contains:
 
-        * acct_server (dict or None) - RADIUS accounting server (same keys as auth_server,
-          ``None`` if accounting is not configured)
-        * reauth_interval (int) - re-authentication interval in seconds (0 = disabled)
-        * pmksa_caching (bool) - whether PMKSA caching (opportunistic key caching) is enabled
+            * ssid (string) - SSID name (repeated for convenience)
+            * bridge (string) - bridge interface the VAP is attached to (e.g. ``"br-lan"``, ``"br-guest"``)
+            * vlan_id (int) - 802.1Q VLAN ID (0 = untagged / no VLAN separation)
+            * tagged (bool) - whether traffic is 802.1Q-tagged on the uplink port
+            * client_isolation (bool) - whether clients on this SSID are isolated from each other
 
-        Note: The RADIUS shared secret is intentionally omitted from the return
-        value for security reasons.
+            Example::
 
-        Example::
-
-            {
-                "CorpWiFi": {
-                    "enabled": True,
-                    "ssid": "CorpWiFi",
-                    "auth_server": {
-                        "host": "radius.corp.example",
-                        "port": 1812,
-                        "timeout": 5,
-                        "retries": 3,
+                {
+                    "CorpWiFi": {
+                        "ssid": "CorpWiFi",
+                        "bridge": "br-corp",
+                        "vlan_id": 10,
+                        "tagged": True,
+                        "client_isolation": False,
                     },
-                    "acct_server": {
-                        "host": "radius.corp.example",
-                        "port": 1813,
-                        "timeout": 5,
-                        "retries": 3,
+                    "GuestNet": {
+                        "ssid": "GuestNet",
+                        "bridge": "br-guest",
+                        "vlan_id": 20,
+                        "tagged": True,
+                        "client_isolation": True,
+                    },
+                    "IoT": {
+                        "ssid": "IoT",
+                        "bridge": "br-iot",
+                        "vlan_id": 30,
+                        "tagged": True,
+                        "client_isolation": True,
                     },
-                    "reauth_interval": 3600,
-                    "pmksa_caching": True,
                 }
-            }
-        """
-        raise NotImplementedError
+            """
+            ...
 
-    def get_packages(self) -> List[PackageDict]:
-        """
-        Returns all packages currently known to the device's package manager
-        (e.g. ``opkg`` on OpenWrt, ``apk`` on Alpine-based APs).
 
-        Each entry contains:
 
-        * name (string) - package name
-        * version (string) - installed or available version string
-        * installed (bool) - ``True`` if the package is currently installed
-        * description (string) - short package description
-        * size (int) - package size in bytes (0 if unknown)
-        * source (string) - repository / feed the package comes from
+        def get_dot1x_config(self) -> Dict[str, Dot1XConfigDict]:
+            """
+            Returns the 802.1X / WPA-Enterprise (RADIUS) configuration per SSID.
 
-        Example::
+            Keys are SSID names.  Each value contains:
+
+            * enabled (bool) - whether 802.1X authentication is active on this SSID
+            * ssid (string) - SSID name (repeated for convenience)
+            * auth_server (dict) - RADIUS authentication server:
+
+              * host (string) - IP or FQDN of the RADIUS server
+              * port (int) - UDP port (default 1812)
+              * timeout (int) - request timeout in seconds
+              * retries (int) - number of retransmissions
+
+            * acct_server (dict or None) - RADIUS accounting server (same keys as auth_server,
+              ``None`` if accounting is not configured)
+            * reauth_interval (int) - re-authentication interval in seconds (0 = disabled)
+            * pmksa_caching (bool) - whether PMKSA caching (opportunistic key caching) is enabled
+
+            Note: The RADIUS shared secret is intentionally omitted from the return
+            value for security reasons.
+
+            Example::
 
-            [
                 {
-                    "name": "luci-app-statistics",
-                    "version": "git-24.001.00000-1",
-                    "installed": True,
-                    "description": "LuCI Statistics application",
-                    "size": 20480,
-                    "source": "openwrt/packages",
-                },
-                {
-                    "name": "collectd-mod-wireless",
-                    "version": "5.12.0-24",
-                    "installed": False,
-                    "description": "Wireless statistics plugin for collectd",
-                    "size": 8192,
-                    "source": "openwrt/packages",
-                },
-            ]
-        """
-        raise NotImplementedError
+                    "CorpWiFi": {
+                        "enabled": True,
+                        "ssid": "CorpWiFi",
+                        "auth_server": {
+                            "host": "radius.corp.example",
+                            "port": 1812,
+                            "timeout": 5,
+                            "retries": 3,
+                        },
+                        "acct_server": {
+                            "host": "radius.corp.example",
+                            "port": 1813,
+                            "timeout": 5,
+                            "retries": 3,
+                        },
+                        "reauth_interval": 3600,
+                        "pmksa_caching": True,
+                    }
+                }
+            """
+            ...
 
-    def install_package(self, name: str, version: str = "") -> None:
-        """
-        Installs a package on the device.
 
-        The method blocks until the installation is complete.  After it returns
-        successfully the package is available for use without a reboot
-        (where the underlying package manager supports this).
 
-        :param name: Package name as known to the package manager.
-        :param version: Exact version to install.  An empty string (default)
-            installs the latest available version.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package name is unknown or the version does
-            not exist in any configured feed.
-        :raises RuntimeError: If the installation fails on the device side
-            (e.g. dependency conflict, disk full).
 
-        Example::
 
-            driver.install_package("luci-app-statistics")
-            driver.install_package("collectd", version="5.12.0-24")
-        """
-        raise NotImplementedError
 
-    def remove_package(self, name: str) -> None:
-        """
-        Removes an installed package from the device.
 
-        The method blocks until the removal is complete.
 
-        :param name: Package name to remove.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package is not currently installed.
-        :raises RuntimeError: If the removal fails on the device side
-            (e.g. other packages depend on it).
 
-        Example::
-
-            driver.remove_package("luci-app-statistics")
-        """
-        raise NotImplementedError
-
-    def get_package_config(self, name: str) -> Dict[str, Any]:
-        """
-        Returns the current configuration of an installed package as a
-        dictionary.  The structure is package-specific.
-
-        :param name: Package name.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package is not installed.
-
-        Example::
-
-            driver.get_package_config("luci-app-statistics")
-            # →
-            {
-                "collectd": {
-                    "enabled": True,
-                    "interval": 30,
-                },
-                "rrdtool": {
-                    "datadir": "/tmp/rrd",
-                    "stepsize": 30,
-                    "heartbeat": 60,
-                },
-            }
-        """
-        raise NotImplementedError
-
-    def get_services(self) -> List[ServiceDict]:
-        """
-        Returns the list of system services known to the device's init system.
-
-        Each entry contains:
-
-        * name (string)    - service name as registered with the init system
-        * running (bool)   - ``True`` if the service process is currently running
-        * enabled (bool)   - ``True`` if the service starts automatically at boot
-        * pid (int)        - process ID of the main service process; 0 if not running
-
-        Example::
-
-            [
-                {"name": "lldpd",   "running": True,  "enabled": True,  "pid": 2341},
-                {"name": "sshd",    "running": True,  "enabled": True,  "pid": 1198},
-                {"name": "cron",    "running": False, "enabled": False, "pid": 0},
-            ]
-        """
-        raise NotImplementedError
-
-    def manage_service(self, name: str, action: str) -> Dict[str, Any]:
-        """
-        Execute a lifecycle action on a named service.
-
-        :param name: Service name as returned by :meth:`get_services`.
-        :param action: One of ``start``, ``stop``, ``restart``, ``enable``, ``disable``.
-        :returns: ``{"success": bool, "output": str}``
-        :raises ValueError: If ``name`` or ``action`` is invalid.
-        :raises NotImplementedError: If the driver does not support service management.
-        """
-        raise NotImplementedError
-
-    def get_available_updates(self) -> List[UpdateDict]:
-        """
-        Returns the list of installed packages that have a newer version available.
-
-        Uses the local package manager cache — does not run ``opkg update`` / ``apk update``.
-
-        :returns: List of :class:`~napalm_device_types.models.UpdateDict`.
-        :raises NotImplementedError: If the driver does not support update listing.
-
-        Example::
-
-            [
-                {"name": "busybox",  "current_version": "1.36.1-1", "new_version": "1.37.0-1"},
-                {"name": "dropbear", "current_version": "2022.83-2", "new_version": "2024.86-1"},
-            ]
-        """
-        raise NotImplementedError
-
-    def apply_updates(self, packages: List[str]) -> Dict[str, Any]:
-        """
-        Upgrade one or more packages to their newest available version.
-
-        :param packages: List of package names to upgrade.
-        :returns: ``{"success": bool, "output": str}``
-        :raises NotImplementedError: If the driver does not support package upgrades.
-        """
-        raise NotImplementedError
-
-    def set_package_config(self, name: str, config: Dict[str, Any]) -> None:
-        """
-        Writes a new configuration for an installed package.
-
-        The ``config`` dict must match the structure returned by
-        :meth:`get_package_config`.  Unknown keys are ignored or raise a
-        ``ValueError`` depending on the driver implementation.
-
-        Changes take effect immediately where the package supports live
-        reload; otherwise a package restart or device reboot may be
-        required – behaviour is driver-specific.
-
-        :param name: Package name.
-        :param config: New configuration as a nested dictionary.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package is not installed or the configuration
-            contains invalid values.
-        :raises RuntimeError: If the device rejects the configuration.
-
-        Example::
-
-            driver.set_package_config(
-                "luci-app-statistics",
-                {
-                    "collectd": {"enabled": True, "interval": 60},
-                    "rrdtool": {"datadir": "/tmp/rrd", "stepsize": 60, "heartbeat": 120},
-                },
-            )
-        """
-        raise NotImplementedError

napalm_device_types/base.py

@@ -12,6 +12,9 @@ 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
+
 
 class FingerprintRule(NamedTuple):
     """Single pattern-matching rule for device fingerprinting.
@@ -45,13 +48,15 @@ class PortSpec(NamedTuple):
     mandatory: bool = False
 
 
-class DeviceTypeDriver(NetworkDriver):
+class DeviceTypeDriver(PingSweepMixin, HostRebootMixin, NetworkDriver):
     """Common base for all netOrk device-type drivers.
 
     Sits between napalm.base.NetworkDriver and the type-specific abstract
     classes (FirewallDriver, SwitchDriver, …). Adds the fingerprinting
-    interface consumed by the discovery subsystem; does not implement any
-    NAPALM abstract methods.
+    interface consumed by the discovery subsystem plus the generic
+    ``ping_sweep()`` from :class:`~napalm_device_types.ping_sweep.PingSweepMixin`
+    (usable by every driver that implements NAPALM's ``ping()``); does not
+    implement any NAPALM abstract methods.
 
     Override these class attributes in each concrete driver:
 
@@ -76,5 +81,15 @@ class DeviceTypeDriver(NetworkDriver):
     SSH_FINGERPRINT: list[FingerprintRule] = []
     HTTP_FINGERPRINT: list[FingerprintRule] = []
     OUI_PREFIXES: list[str] = []
+
+    #: Whether netOrk reaches this device over SSH. False for drivers that talk
+    #: to a REST API instead -- which decides whether an SSH credential is worth
+    #: asking for, and whether an SSH-shaped error message would even make sense.
+    USES_SSH: bool = True
+
+    #: How long a reboot takes before the device is worth polling again, in
+    #: seconds. Hypervisors and general-purpose OS hosts run through a full
+    #: init sequence; a switch or an access point is back in half the time.
+    REBOOT_SETTLE_SECONDS: int = 45
     # Format: "AA:BB:CC" — first 3 octets of MAC, uppercase, colon-separated.
     # A match contributes fixed weight 6.0 to the fingerprint score.

napalm_device_types/config_lifecycle.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import difflib
-from typing import ClassVar
+from typing import ClassVar, TYPE_CHECKING
 
 from napalm.base.exceptions import (
     CommandErrorException,
@@ -39,11 +39,13 @@ class ConfigLifecycleMixin:
 
     _comment_chars: ClassVar[tuple[str, ...]] = ("!", "#")
 
-    def _get_running_config(self) -> str:
-        raise NotImplementedError
+    if TYPE_CHECKING:
 
-    def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
-        raise NotImplementedError
+        def _get_running_config(self) -> str:
+            ...
+
+        def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
+            ...
 
     def load_merge_candidate(
         self, filename: str | None = None, config: str | None = None

napalm_device_types/dhcp.py

@@ -0,0 +1,400 @@
+# -*- coding: utf-8 -*-
+"""DHCP desired-state management, shared by firewalls and gateways.
+
+Covers two independent desired-state sets:
+
+* **reservations** -- static MAC -> IP bindings, matched on the MAC;
+* **subnets** -- the served ranges and their per-subnet DHCP options,
+  matched on the CIDR.
+
+Both :class:`~napalm_device_types.firewall.FirewallDriver` and
+:class:`~napalm_device_types.residential_gateway.ResidentialGatewayDriver`
+mix this in, because both device types commonly run the DHCP server for
+their networks.
+
+Only the six get/apply/commit methods are device-specific and must be
+implemented by a concrete driver; the diffs and the apply loops are
+vendor-neutral algorithms and live here -- see README.md "Design principle:
+generic vs. device-specific logic".
+
+Neither diff ever deletes. For subnets that is not just caution: removing one
+takes DHCP down for an entire VLAN.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Dict, Iterator, List, Optional, TYPE_CHECKING
+
+from napalm_device_types.models import (
+    DhcpReservationDiffDict,
+    DhcpReservationDict,
+    DhcpReservationUpdateDict,
+    DhcpSubnetDiffDict,
+    DhcpSubnetDict,
+    DhcpSubnetUpdateDict,
+)
+
+# `mac` is the identity, so it is matched rather than compared. `uuid` is
+# assigned by the device and never part of the desired state.
+_DHCP_RESERVATION_COMPARE_FIELDS = (
+    "ip",
+    "hostname",
+    "description",
+    "subnet",
+)
+
+# `subnet` (the CIDR) is the identity, so it is matched rather than compared.
+# `uuid` is assigned by the device and never part of the desired state.
+_DHCP_SUBNET_COMPARE_FIELDS = (
+    "description",
+    "pools",
+    "option_data",
+    "match_client_id",
+)
+
+_HEX_ONLY = re.compile(r"[^0-9a-f]")
+
+
+def normalize_mac(mac: Optional[str]) -> str:
+    """Reduces a MAC address to lowercase colon-separated form.
+
+    Devices report MACs in whatever form their config store happens to use --
+    ``AA-BB-CC-DD-EE-01``, ``aabb.ccdd.ee01``, ``AABBCCDDEE01``. Reservations
+    are matched on this value, so it has to be canonical before comparison.
+
+    A value that is not 12 hex digits is returned lowercased and stripped
+    instead of raising: a malformed device response should degrade to "this
+    entry never matches" rather than abort the whole diff.
+    """
+    if not mac:
+        return ""
+
+    lowered = mac.strip().lower()
+    hex_digits = _HEX_ONLY.sub("", lowered)
+    if len(hex_digits) != 12:
+        return lowered
+
+    return ":".join(hex_digits[i : i + 2] for i in range(0, 12, 2))
+
+
+def normalize_cidr(cidr: Optional[str]) -> str:
+    """Reduces a subnet CIDR to a canonical string for matching.
+
+    Only whitespace and case are normalised -- deliberately not the network
+    address itself. Rewriting ``10.10.20.5/24`` to ``10.10.20.0/24`` would
+    make a caller's typo silently match a real subnet and then apply that
+    caller's pools and options to it.
+    """
+    return (cidr or "").strip().lower()
+
+
+def _subnet_field_differs(
+    field: str, live: DhcpSubnetDict, desired: DhcpSubnetDict
+) -> bool:
+    """Compares one subnet field, with `option_data` handled specially.
+
+    For `option_data` only the options `desired` actually names are compared;
+    see ``diff_dhcp_subnets`` for why an unmentioned option must not count as
+    a difference.
+    """
+    if field != "option_data":
+        return live.get(field) != desired.get(field)
+
+    desired_options = desired.get("option_data") or {}
+    live_options = live.get("option_data") or {}
+    return any(
+        live_options.get(option) != value for option, value in desired_options.items()
+    )
+
+
+class DhcpServerMixin:
+    """Mixin providing DHCP reservation and subnet read/diff/apply.
+
+    Concrete drivers **must** provide ``get_dhcp_reservations()``,
+    ``apply_dhcp_reservation()`` and ``commit_dhcp_reservations()``; drivers
+    that also manage subnets provide ``get_dhcp_subnets()``,
+    ``apply_dhcp_subnet()`` and ``commit_dhcp_subnets()``.
+    """
+
+    # ------------------------------------------------------------------
+    # Device-specific -- must be implemented by the concrete driver.
+    # ------------------------------------------------------------------
+
+    if TYPE_CHECKING:
+
+        def get_dhcp_reservations(self) -> List[DhcpReservationDict]:
+            """
+            Returns all static DHCP reservations currently configured on the
+            device, across all subnets.
+
+            This is the *configured* state, not the observed leases -- see
+            ``get_dhcp_leases()`` for the latter.
+
+            :raises NotImplementedError: If the driver does not support reading
+                DHCP reservations.
+            """
+            ...
+
+        def apply_dhcp_reservation(
+            self, reservation: DhcpReservationDict, *, uuid: Optional[str] = None
+        ) -> Dict[str, Any]:
+            """
+            Creates or updates a single static DHCP reservation on the device.
+
+            :param reservation: The desired reservation state, vendor-neutral.
+            :param uuid: If given, update the existing reservation with this ID
+                in-place. If ``None``, create a new one.
+            :raises NotImplementedError: If the driver does not support writing
+                DHCP reservations.
+            :raises ValueError: If `reservation` names a subnet the device does
+                not serve.
+            :raises RuntimeError: If the device rejects the write.
+
+            :returns: A dict with at least ``{"success": bool}``.
+            """
+            ...
+
+        def get_dhcp_subnets(self) -> List[DhcpSubnetDict]:
+            """
+            Returns every DHCPv4 subnet the device serves, with its pools and
+            per-subnet options.
+
+            :raises NotImplementedError: If the driver does not support reading
+                DHCP subnets.
+            """
+            ...
+
+        def apply_dhcp_subnet(
+            self, subnet: DhcpSubnetDict, *, uuid: Optional[str] = None
+        ) -> Dict[str, Any]:
+            """
+            Creates or updates a single DHCPv4 subnet on the device.
+
+            Implementations must treat ``subnet["option_data"]`` as a partial
+            update: an option the caller did not name is left as the device has
+            it. Managing `domain_search` alone is the common case, and it must
+            not silently drop the `routers` the server autocollected.
+
+            :param subnet: The desired subnet state, vendor-neutral.
+            :param uuid: If given, update the existing subnet with this ID
+                in-place. If ``None``, create a new one.
+            :raises NotImplementedError: If the driver does not support writing
+                DHCP subnets.
+            :raises RuntimeError: If the device rejects the write.
+
+            :returns: A dict with at least ``{"success": bool}``.
+            """
+            ...
+
+        def commit_dhcp_subnets(self) -> Dict[str, Any]:
+            """
+            Applies pending subnet changes (e.g. Kea's ``service/reconfigure``).
+
+            Separate from ``commit_dhcp_reservations`` even where a driver
+            implements both with the same call: the two desired-state sets are
+            applied independently, and a caller that changed only subnets should
+            not have to know which reload the vendor happens to share.
+
+            :raises NotImplementedError: If the driver does not support this.
+
+            :returns: A dict with at least ``{"success": bool}``.
+            """
+            ...
+
+        def commit_dhcp_reservations(self) -> Dict[str, Any]:
+            """
+            Applies pending reservation changes (e.g. Kea's ``service/reconfigure``,
+            or a dnsmasq reload).
+
+            Call once after one or more `apply_dhcp_reservation()` calls -- not
+            after every single reservation, and not at all when nothing changed:
+            on most implementations this reloads the DHCP daemon.
+
+            :raises NotImplementedError: If the driver does not support this
+                (e.g. reservations take effect immediately on write).
+
+            :returns: A dict with at least ``{"success": bool}``.
+            """
+            ...
+
+    # ------------------------------------------------------------------
+    # Generic, vendor-neutral algorithms.
+    # ------------------------------------------------------------------
+
+    def diff_dhcp_reservations(
+        self, desired: List[DhcpReservationDict]
+    ) -> DhcpReservationDiffDict:
+        """
+        Compares `desired` against the device's current reservations and
+        returns what would need to change to reach that state.
+
+        Matches on the normalised MAC address. A desired reservation with no
+        live counterpart becomes an "add"; a live one whose MAC matches but
+        whose other fields differ becomes an "update". Live reservations with
+        no matching desired entry are **not** reported for deletion -- a DHCP
+        server routinely carries hand-created reservations that a caller's
+        `desired` set was never meant to describe, and this method cannot tell
+        those apart from ones simply no longer wanted. Callers wanting
+        delete/cleanup semantics must implement that themselves, deliberately.
+
+        :param desired: The complete desired reservation set.
+        :returns: ``{"add": [...], "update": [{"uuid", "reservation",
+            "changed_fields"}, ...]}``.
+        """
+        live_by_mac: Dict[str, DhcpReservationDict] = {
+            normalize_mac(reservation.get("mac")): reservation
+            for reservation in self.get_dhcp_reservations()
+        }
+
+        add: List[DhcpReservationDict] = []
+        update: List[DhcpReservationUpdateDict] = []
+
+        for desired_reservation in desired:
+            live = live_by_mac.get(normalize_mac(desired_reservation.get("mac")))
+            if live is None:
+                add.append(desired_reservation)
+                continue
+
+            changed_fields = [
+                field
+                for field in _DHCP_RESERVATION_COMPARE_FIELDS
+                if live.get(field) != desired_reservation.get(field)
+            ]
+            if changed_fields:
+                update.append(
+                    {
+                        "uuid": live["uuid"],
+                        "reservation": desired_reservation,
+                        "changed_fields": changed_fields,
+                    }
+                )
+
+        return {"add": add, "update": update}
+
+    def apply_dhcp_reservationset(
+        self, desired: List[DhcpReservationDict]
+    ) -> Iterator[str]:
+        """
+        Computes the diff against `desired` and applies it, yielding one
+        human-readable progress line per change, then commits.
+
+        Unlike ``apply_firewall_ruleset``, an empty diff does **not** commit:
+        committing reloads the DHCP daemon and drops in-flight requests, which
+        is too high a price for a no-op run.
+
+        :param desired: The complete desired reservation set.
+        :yields: Progress lines, one per applied add/update, plus a final
+            commit line.
+        """
+        diff = self.diff_dhcp_reservations(desired)
+
+        for reservation in diff["add"]:
+            self.apply_dhcp_reservation(reservation)
+            yield f"[add] {reservation['mac']} -> {reservation['ip']}"
+
+        for entry in diff["update"]:
+            self.apply_dhcp_reservation(entry["reservation"], uuid=entry["uuid"])
+            fields = ", ".join(entry["changed_fields"])
+            yield (
+                f"[update] {entry['reservation']['mac']} -> "
+                f"{entry['reservation']['ip']} ({fields})"
+            )
+
+        if not diff["add"] and not diff["update"]:
+            yield "[commit] no changes"
+            return
+
+        self.commit_dhcp_reservations()
+        yield (
+            f"[commit] applied {len(diff['add'])} add(s), "
+            f"{len(diff['update'])} update(s)"
+        )
+
+    def diff_dhcp_subnets(self, desired: List[DhcpSubnetDict]) -> DhcpSubnetDiffDict:
+        """
+        Compares `desired` against the device's current subnets and returns
+        what would need to change to reach that state.
+
+        Matches on the CIDR. A desired subnet with no live counterpart becomes
+        an "add"; a live one whose CIDR matches but whose other fields differ
+        becomes an "update".
+
+        `option_data` is compared **per option**, and only over the options
+        the caller named. An option the device carries but `desired` does not
+        mention is left out of the comparison entirely, because absent means
+        "not managed" rather than "should be empty". Without that rule a
+        caller managing only `domain_search` would diff against every option
+        Kea autocollects (`routers`, `domain_name_servers`, `ntp_servers`) and
+        reconfigure the DHCP daemon on every single run.
+
+        Live subnets with no matching desired entry are **not** reported for
+        deletion, and more emphatically than for reservations: removing a
+        subnet takes DHCP down for a whole VLAN, and this method cannot tell
+        "no longer wanted" from "was never netOrk's to describe".
+
+        :param desired: The complete desired subnet set.
+        :returns: ``{"add": [...], "update": [{"uuid", "subnet",
+            "changed_fields"}, ...]}``.
+        """
+        live_by_cidr: Dict[str, DhcpSubnetDict] = {
+            normalize_cidr(live.get("subnet")): live for live in self.get_dhcp_subnets()
+        }
+
+        add: List[DhcpSubnetDict] = []
+        update: List[DhcpSubnetUpdateDict] = []
+
+        for desired_subnet in desired:
+            live = live_by_cidr.get(normalize_cidr(desired_subnet.get("subnet")))
+            if live is None:
+                add.append(desired_subnet)
+                continue
+
+            changed_fields = [
+                field
+                for field in _DHCP_SUBNET_COMPARE_FIELDS
+                if _subnet_field_differs(field, live, desired_subnet)
+            ]
+            if changed_fields:
+                update.append(
+                    {
+                        "uuid": live["uuid"],
+                        "subnet": desired_subnet,
+                        "changed_fields": changed_fields,
+                    }
+                )
+
+        return {"add": add, "update": update}
+
+    def apply_dhcp_subnetset(self, desired: List[DhcpSubnetDict]) -> Iterator[str]:
+        """
+        Computes the diff against `desired` and applies it, yielding one
+        human-readable progress line per change, then commits.
+
+        As with ``apply_dhcp_reservationset``, an empty diff does **not**
+        commit: reconfiguring the DHCP daemon is too expensive for a no-op.
+
+        :param desired: The complete desired subnet set.
+        :yields: Progress lines, one per applied add/update, plus a final
+            commit line.
+        """
+        diff = self.diff_dhcp_subnets(desired)
+
+        for subnet in diff["add"]:
+            self.apply_dhcp_subnet(subnet)
+            yield f"[add] {subnet['subnet']}"
+
+        for entry in diff["update"]:
+            self.apply_dhcp_subnet(entry["subnet"], uuid=entry["uuid"])
+            fields = ", ".join(entry["changed_fields"])
+            yield f"[update] {entry['subnet']['subnet']} ({fields})"
+
+        if not diff["add"] and not diff["update"]:
+            yield "[commit] no changes"
+            return
+
+        self.commit_dhcp_subnets()
+        yield (
+            f"[commit] applied {len(diff['add'])} add(s), "
+            f"{len(diff['update'])} update(s)"
+        )

napalm_device_types/firewall.py

@@ -10,21 +10,21 @@ Usage::
             ...
 """
 
-from typing import Any, Dict, List
+from typing import Any, ClassVar, Dict, Iterator, List, Optional, TYPE_CHECKING
 from napalm_device_types.base import DeviceTypeDriver
-from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
+from napalm_device_types.nat_vpn import NatVpnMixin
+from napalm_device_types.packages import PackageManagementMixin
+from napalm_device_types.health_metrics import HealthMetricsMixin
+from napalm_device_types.dhcp import DhcpServerMixin
+from napalm_device_types.firewall_rules import FirewallRuleMixin
 from napalm_device_types.models import (
-    HealthMetricsDict,
-    NATTranslationDict,
-    PackageDict,
     SecurityZoneDict,
     SessionDict,
-    VPNTunnelDict,
 )
 
 
-class FirewallDriver(DeviceTypeDriver):
-    TYPE_LABEL: str = "Firewall"
+
+class FirewallDriver(NatVpnMixin, PackageManagementMixin, HealthMetricsMixin, FirewallRuleMixin, DhcpServerMixin, DeviceTypeDriver):
     """
     Abstract intermediate driver for firewall/security devices.
 
@@ -33,268 +33,128 @@ class FirewallDriver(DeviceTypeDriver):
     that concrete drivers must implement.
     """
 
-    _SNMP_SKIP_IF = IF_SKIP_DEFAULT
+    #: Stable key netOrk exposes as ``device_class``. The order in which a
+    #: driver lists its role bases is the ranking; see
+    #: :func:`napalm_device_types.roles.primary_role_of`.
+    ROLE: str = "firewall"
+    TYPE_LABEL: str = "Firewall"
+
     # OPNsense reports drops in the out-error counter.
-    _SNMP_TX_ERR_IS_DROP: bool = True
+    _SNMP_TX_ERR_IS_DROP: ClassVar[bool] = True
 
-    @classmethod
-    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
-        return await collect_ucd_metrics(
-            snmp_get, snmp_walk,
-            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
-            if_skip=cls._SNMP_SKIP_IF,
-        )
 
-    def get_nat_translations(self) -> List[NATTranslationDict]:
-        """
-        Returns a list of active NAT translation entries.
+    if TYPE_CHECKING:
 
-        Each entry contains:
 
-        * protocol (string) - ``"tcp"``, ``"udp"``, ``"icmp"``
-        * inside_local (string) - original source address (IP or IP:port)
-        * inside_global (string) - translated source address (IP or IP:port)
-        * outside_local (string) - destination as seen from inside
-        * outside_global (string) - actual destination address
-        * age (float) - translation entry age in seconds
+        def get_security_zones(self) -> Dict[str, SecurityZoneDict]:
+            """
+            Returns the security zone configuration.
 
-        Example::
+            Keys are zone names.  Each value contains:
+
+            * interfaces (list of strings) - interfaces assigned to this zone
+            * policy (string) - name of the security policy applied to this zone
+            * description (string) - zone description
+
+            Example::
 
-            [
                 {
-                    "protocol": "tcp",
-                    "inside_local": "192.168.1.10:54321",
-                    "inside_global": "203.0.113.1:54321",
-                    "outside_local": "1.1.1.1:443",
-                    "outside_global": "1.1.1.1:443",
-                    "age": 120.5,
+                    "LAN": {
+                        "interfaces": ["eth0", "eth1"],
+                        "policy": "LAN-policy",
+                        "description": "Internal LAN zone",
+                    },
+                    "WAN": {
+                        "interfaces": ["eth2"],
+                        "policy": "WAN-policy",
+                        "description": "Uplink to internet",
+                    },
                 }
-            ]
-        """
-        raise NotImplementedError
+            """
+            ...
 
-    def get_security_zones(self) -> Dict[str, SecurityZoneDict]:
-        """
-        Returns the security zone configuration.
+        def get_sessions(self) -> List[SessionDict]:
+            """
+            Returns a list of active connection sessions (stateful flows).
 
-        Keys are zone names.  Each value contains:
+            Each entry contains:
 
-        * interfaces (list of strings) - interfaces assigned to this zone
-        * policy (string) - name of the security policy applied to this zone
-        * description (string) - zone description
+            * protocol (string) - ``"tcp"``, ``"udp"``, ``"icmp"``
+            * src_ip (string) - source IP address
+            * src_port (int) - source port (0 for ICMP)
+            * dst_ip (string) - destination IP address
+            * dst_port (int) - destination port (0 for ICMP)
+            * state (string) - session state, e.g. ``"established"``, ``"syn_sent"``
+            * age (float) - session age in seconds
 
-        Example::
+            Example::
 
-            {
-                "LAN": {
-                    "interfaces": ["eth0", "eth1"],
-                    "policy": "LAN-policy",
-                    "description": "Internal LAN zone",
-                },
-                "WAN": {
-                    "interfaces": ["eth2"],
-                    "policy": "WAN-policy",
-                    "description": "Uplink to internet",
-                },
-            }
-        """
-        raise NotImplementedError
+                [
+                    {
+                        "protocol": "tcp",
+                        "src_ip": "192.168.1.10",
+                        "src_port": 54321,
+                        "dst_ip": "1.1.1.1",
+                        "dst_port": 443,
+                        "state": "established",
+                        "age": 30.2,
+                    }
+                ]
+            """
+            ...
 
-    def get_sessions(self) -> List[SessionDict]:
-        """
-        Returns a list of active connection sessions (stateful flows).
 
-        Each entry contains:
+        def send_wake_on_lan(self, mac_address: str, interface: str = "") -> Dict[str, Any]:
+            """
+            Sends a Wake-on-LAN "magic packet" to wake a host on the network.
 
-        * protocol (string) - ``"tcp"``, ``"udp"``, ``"icmp"``
-        * src_ip (string) - source IP address
-        * src_port (int) - source port (0 for ICMP)
-        * dst_ip (string) - destination IP address
-        * dst_port (int) - destination port (0 for ICMP)
-        * state (string) - session state, e.g. ``"established"``, ``"syn_sent"``
-        * age (float) - session age in seconds
+            :param mac_address: Target host's MAC address (colon-separated,
+                case-insensitive, e.g. ``"AA:BB:CC:DD:EE:FF"``).
+            :param interface: Driver-specific interface identifier to broadcast the
+                magic packet from. Required by drivers that scope WOL per interface
+                (e.g. OPNsense); an empty string means "use the driver's default/
+                only broadcast domain." Consult the concrete driver's docstring for
+                the exact expected format.
+            :raises NotImplementedError: If the driver does not support Wake-on-LAN.
+            :raises ValueError: If ``mac_address`` is malformed, or ``interface`` is
+                required by this driver but was not provided.
 
-        Example::
+            :returns: A dict with:
 
-            [
-                {
-                    "protocol": "tcp",
-                    "src_ip": "192.168.1.10",
-                    "src_port": 54321,
-                    "dst_ip": "1.1.1.1",
-                    "dst_port": 443,
-                    "state": "established",
-                    "age": 30.2,
-                }
-            ]
-        """
-        raise NotImplementedError
+                * success (bool) - ``True`` if the magic packet was sent without error
+                * output (string) - human-readable status message
 
-    def get_vpn_tunnels(self) -> Dict[str, VPNTunnelDict]:
-        """
-        Returns the status of VPN tunnels.
+            Example::
 
-        Keys are tunnel names or identifiers.  Each value contains:
+                driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
+                # → {"success": True, "output": "Magic packet sent to AA:BB:CC:DD:EE:FF via lan"}
 
-        * type (string) - tunnel type: ``"IPsec"``, ``"SSL"``, ``"GRE"``, ``"WireGuard"``
-        * local_endpoint (string) - local tunnel endpoint IP
-        * remote_endpoint (string) - remote tunnel endpoint IP
-        * is_up (bool) - whether the tunnel is operationally up
-        * uptime (int) - tunnel uptime in seconds (0 if down)
-        * bytes_in (int) - total bytes received through the tunnel
-        * bytes_out (int) - total bytes sent through the tunnel
+            .. note::
 
-        Example::
+                A driver whose ``interface`` is *not* the name :meth:`get_interfaces`
+                is keyed by must expose the name it does expect as an ``identifier``
+                key on each ``get_interfaces()`` entry. Without it a caller has no
+                way to offer a valid choice: OPNsense, for instance, keys interfaces
+                by the physical device ("em0") but wakes by the assigned name
+                ("lan"), and rejects the former. ``identifier`` is a non-standard
+                NAPALM key, so it reaches consumers through the usual passthrough
+                for extra interface data.
+            """
+            ...
 
-            {
-                "vpn-to-branch": {
-                    "type": "IPsec",
-                    "local_endpoint": "203.0.113.1",
-                    "remote_endpoint": "198.51.100.1",
-                    "is_up": True,
-                    "uptime": 86400,
-                    "bytes_in": 104857600,
-                    "bytes_out": 52428800,
-                }
-            }
-        """
-        raise NotImplementedError
 
-    def get_packages(self) -> List[PackageDict]:
-        """
-        Returns all packages / plugins currently known to the firewall's
-        package manager (e.g. ``pkg`` on pfSense/OPNsense, ``FortiGate
-        License`` add-ons, ``apt`` on Debian-based firewalls).
 
-        Each entry contains:
 
-        * name (string) - package name
-        * version (string) - installed or available version string
-        * installed (bool) - ``True`` if the package is currently installed
-        * description (string) - short package description
-        * size (int) - package size in bytes (0 if unknown)
-        * source (string) - repository / channel the package comes from
 
-        Example::
 
-            [
-                {
-                    "name": "pfBlockerNG",
-                    "version": "3.2.0_4",
-                    "installed": True,
-                    "description": "IP and DNS blocking for pfSense",
-                    "size": 2097152,
-                    "source": "pfSense-pkg",
-                },
-                {
-                    "name": "suricata",
-                    "version": "7.0.3_1",
-                    "installed": False,
-                    "description": "High-performance Network IDS/IPS",
-                    "size": 51380224,
-                    "source": "pfSense-pkg",
-                },
-            ]
-        """
-        raise NotImplementedError
+        # ------------------------------------------------------------------
+        # Firewall rule diff/apply. get_firewall_rules/apply_firewall_rule/
+        # commit_firewall_rules are abstract (device communication); everything
+        # else here is a concrete, vendor-neutral algorithm -- see README.md
+        # "Design principle: generic vs. device-specific logic".
+        # ------------------------------------------------------------------
 
-    def install_package(self, name: str, version: str = "") -> None:
-        """
-        Installs a package or plugin on the firewall.
 
-        The method blocks until the installation is complete.  Whether a
-        reboot is required afterwards depends on the device; check the vendor
-        documentation.
 
-        :param name: Package name as known to the package manager.
-        :param version: Exact version to install.  An empty string (default)
-            installs the latest available version.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package name is unknown or the requested
-            version is not available.
-        :raises RuntimeError: If the installation fails on the device side
-            (e.g. license missing, dependency conflict, disk full).
 
-        Example::
 
-            driver.install_package("pfBlockerNG")
-            driver.install_package("suricata", version="7.0.3_1")
-        """
-        raise NotImplementedError
-
-    def remove_package(self, name: str) -> None:
-        """
-        Removes an installed package or plugin from the firewall.
-
-        The method blocks until the removal is complete.
-
-        :param name: Package name to remove.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package is not currently installed.
-        :raises RuntimeError: If the removal fails on the device side
-            (e.g. the package is a system dependency).
-
-        Example::
-
-            driver.remove_package("pfBlockerNG")
-        """
-        raise NotImplementedError
-
-    def get_package_config(self, name: str) -> Dict[str, Any]:
-        """
-        Returns the current configuration of an installed package or plugin
-        as a dictionary.  The structure is package-specific.
-
-        :param name: Package name.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package is not installed.
-
-        Example::
-
-            driver.get_package_config("pfBlockerNG")
-            # →
-            {
-                "enable": True,
-                "maxmind_key": "",
-                "blocklists": [
-                    {"name": "PRI1", "action": "Deny_Both", "enabled": True},
-                    {"name": "DNSBL_ADs", "action": "Unbound", "enabled": True},
-                ],
-                "update_interval": "Once a day",
-            }
-        """
-        raise NotImplementedError
-
-    def set_package_config(self, name: str, config: Dict[str, Any]) -> None:
-        """
-        Writes a new configuration for an installed package or plugin.
-
-        The ``config`` dict must match the structure returned by
-        :meth:`get_package_config`.  Unknown keys are ignored or raise a
-        ``ValueError`` depending on the driver implementation.
-
-        Changes take effect immediately where the package supports live
-        reload; otherwise a package restart or device reboot may be
-        required – behaviour is driver-specific.
-
-        :param name: Package name.
-        :param config: New configuration as a nested dictionary.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package is not installed or the configuration
-            contains invalid values.
-        :raises RuntimeError: If the device rejects the configuration.
-
-        Example::
-
-            driver.set_package_config(
-                "pfBlockerNG",
-                {
-                    "enable": True,
-                    "blocklists": [
-                        {"name": "PRI1", "action": "Deny_Both", "enabled": True},
-                    ],
-                    "update_interval": "Twice a day",
-                },
-            )
-        """
-        raise NotImplementedError

napalm_device_types/firewall_rules.py

@@ -0,0 +1,161 @@
+# -*- coding: utf-8 -*-
+"""Firewall rule reconciliation: generic here, device communication in the driver.
+
+The split follows the rule in this package's README: matching, diffing and the
+apply-then-commit sequence would be identical for any vendor's firewall, so they
+live here as concrete methods. Only the three hooks below touch the device, and
+a concrete driver supplies those.
+
+Declared under ``if TYPE_CHECKING``, the hooks do not exist at runtime until a
+driver implements them -- so ``hasattr`` stays an honest answer to "can this
+driver manage firewall rules", and mixing this class in can never shadow a
+working implementation inherited from elsewhere.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Iterator, List, Optional, TYPE_CHECKING
+
+from napalm_device_types.models import (
+    FirewallRuleDict,
+    FirewallRuleDiffDict,
+    FirewallRuleUpdateDict,
+)
+
+_FIREWALL_RULE_COMPARE_FIELDS = (
+    "action",
+    "interface",
+    "direction",
+    "protocol",
+    "source_net",
+    "source_port",
+    "destination_net",
+    "destination_port",
+    "log",
+    "quick",
+    "enabled",
+)
+
+
+class FirewallRuleMixin:
+    """Adds generic firewall-rule diff and apply on top of three device hooks."""
+
+    if TYPE_CHECKING:
+
+        def get_firewall_rules(self) -> List[FirewallRuleDict]:
+            """
+            Returns all firewall filter rules currently configured on the device.
+
+            `description` must be a stable, human-assigned identifier -- it is
+            the key used to match rules across calls (most firewall vendors
+            don't expose an ID a caller can pre-assign).
+
+            :raises NotImplementedError: If the driver does not support reading
+                firewall rules.
+            """
+            ...
+        def apply_firewall_rule(
+            self, rule: FirewallRuleDict, *, uuid: Optional[str] = None
+        ) -> Dict[str, Any]:
+            """
+            Creates or updates a single firewall filter rule on the device.
+
+            :param rule: The desired rule state, in vendor-neutral form.
+            :param uuid: If given, update the existing rule with this ID
+                in-place. If ``None``, create a new rule.
+            :raises NotImplementedError: If the driver does not support writing
+                firewall rules.
+            :raises ValueError: If `rule` references an alias/interface the
+                device doesn't know about.
+            :raises RuntimeError: If the device rejects the write.
+
+            :returns: A dict with at least ``{"success": bool}``.
+            """
+            ...
+        def commit_firewall_rules(self) -> Dict[str, Any]:
+            """
+            Applies pending firewall filter rule changes (e.g. reloads pf/pfctl,
+            or whatever the device's equivalent of "Apply Changes" is).
+
+            Call once after one or more `apply_firewall_rule()` calls -- not
+            after every single rule.
+
+            :raises NotImplementedError: If the driver does not support this
+                (e.g. rules take effect immediately on write).
+
+            :returns: A dict with at least ``{"success": bool}``.
+            """
+            ...
+
+    def diff_firewall_rules(self, desired: List[FirewallRuleDict]) -> FirewallRuleDiffDict:
+        """
+        Compares `desired` against the device's current rules and returns
+        what would need to change to reach that state.
+
+        Matches rules by `description`. A desired rule with no live
+        counterpart becomes an "add"; a live rule whose description matches
+        but whose other fields differ becomes an "update". Live rules with
+        no matching desired entry are **not** reported for deletion -- this
+        is intentionally conservative: a firewall may carry manually-created
+        or otherwise unmanaged rules that a caller's `desired` set was never
+        meant to describe, and this method has no way to distinguish those
+        from ones simply no longer wanted. Callers wanting delete/cleanup
+        semantics must implement that themselves, deliberately.
+
+        :param desired: The complete desired rule set.
+        :returns: ``{"add": [...], "update": [{"uuid", "rule",
+            "changed_fields"}, ...]}``.
+        """
+        live_by_description: Dict[str, FirewallRuleDict] = {
+            rule["description"]: rule for rule in self.get_firewall_rules()
+        }
+
+        add: List[FirewallRuleDict] = []
+        update: List[FirewallRuleUpdateDict] = []
+
+        for desired_rule in desired:
+            live_rule = live_by_description.get(desired_rule["description"])
+            if live_rule is None:
+                add.append(desired_rule)
+                continue
+
+            changed_fields = [
+                field
+                for field in _FIREWALL_RULE_COMPARE_FIELDS
+                if live_rule.get(field) != desired_rule.get(field)
+            ]
+            if changed_fields:
+                update.append(
+                    {
+                        "uuid": live_rule["uuid"],
+                        "rule": desired_rule,
+                        "changed_fields": changed_fields,
+                    }
+                )
+
+        return {"add": add, "update": update}
+    def apply_firewall_ruleset(self, desired: List[FirewallRuleDict]) -> Iterator[str]:
+        """
+        Computes the diff against `desired` and applies it, yielding one
+        human-readable progress line per change, then commits.
+
+        Intended for streaming to a caller (e.g. an SSE endpoint) that wants
+        live progress while writing to a real device.
+
+        :param desired: The complete desired rule set.
+        :yields: Progress lines, one per applied add/update, plus a final
+            commit line.
+        """
+        diff = self.diff_firewall_rules(desired)
+
+        for rule in diff["add"]:
+            self.apply_firewall_rule(rule)
+            yield f"[add] {rule['description']}"
+
+        for entry in diff["update"]:
+            self.apply_firewall_rule(entry["rule"], uuid=entry["uuid"])
+            fields = ", ".join(entry["changed_fields"])
+            yield f"[update] {entry['rule']['description']} ({fields})"
+
+        self.commit_firewall_rules()
+        yield f"[commit] applied {len(diff['add'])} add(s), {len(diff['update'])} update(s)"

napalm_device_types/health_metrics.py

@@ -0,0 +1,53 @@
+# -*- coding: utf-8 -*-
+"""SNMP health collection, shared by every device type that speaks UCD-MIB.
+
+The collection itself is generic: CPU, memory, uptime, load and per-interface
+counters come from the same standard OIDs on any UCD-MIB/IF-MIB capable device.
+Only two details vary by device type, and both are class attributes rather than
+code -- which is why this was five byte-identical copies of the same method
+before it moved here.
+
+A device type whose vendor publishes the same data under proprietary OIDs does
+**not** mix this in; ``SwitchDriver`` is the example, and its concrete drivers
+implement ``get_health_metrics`` themselves.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Awaitable, Callable, ClassVar, Dict, cast
+
+from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
+from napalm_device_types.models import HealthMetricsDict
+
+
+class HealthMetricsMixin:
+    """Adds the standard UCD-MIB/IF-MIB ``get_health_metrics`` collector."""
+
+    #: Interfaces whose counters are noise rather than signal. Loopback,
+    #: tunnels and container bridges inflate error rates without meaning.
+    _SNMP_SKIP_IF: ClassVar[re.Pattern[str]] = IF_SKIP_DEFAULT
+
+    #: Whether the device reports discards in the TX-error counter. Firewalls
+    #: do -- a dropped packet is the point, not a fault -- so counting those as
+    #: errors would make a healthy firewall look broken.
+    _SNMP_TX_ERR_IS_DROP: ClassVar[bool] = False
+
+    @classmethod
+    async def get_health_metrics(
+        cls,
+        snmp_get: Callable[..., Awaitable[Any]],
+        snmp_walk: Callable[..., Awaitable[Dict[str, Any]]],
+    ) -> HealthMetricsDict:
+        """Collect SNMP health metrics (CPU, memory, uptime, load, interfaces)."""
+        # collect_ucd_metrics is untyped shared plumbing; the shape it returns is
+        # the contract HealthMetricsDict describes.
+        return cast(
+            HealthMetricsDict,
+            await collect_ucd_metrics(
+                snmp_get,
+                snmp_walk,
+                tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
+                if_skip=cls._SNMP_SKIP_IF,
+            ),
+        )

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/interface_filter.py

@@ -0,0 +1,31 @@
+# -*- coding: utf-8 -*-
+"""Dropping interfaces that are noise rather than signal.
+
+An access point exposes radio PHYs and a loopback alongside its real
+interfaces. Reporting them inflates every interface count and every error rate,
+so drivers filter them out -- identically, whatever the vendor. The two class
+attributes are the customisation point.
+"""
+
+from __future__ import annotations
+
+from typing import Any, ClassVar, Dict, FrozenSet, Tuple
+
+
+class InterfaceFilterMixin:
+    """Adds :meth:`_filter_interfaces` for drivers that report raw interface dicts."""
+
+    #: Interface names dropped outright.
+    _EXCLUDED_INTERFACES: ClassVar[FrozenSet[str]] = frozenset({"lo"})
+
+    #: Name prefixes dropped -- ``phy*`` are radio devices, not links.
+    _EXCLUDED_INTERFACE_PREFIXES: ClassVar[Tuple[str, ...]] = ("phy",)
+
+    def _filter_interfaces(self, interfaces: Dict[str, Any]) -> Dict[str, Any]:
+        """Remove loopback and radio-device (phy*) interfaces from an interface dict."""
+        return {
+            name: data
+            for name, data in interfaces.items()
+            if name not in self._EXCLUDED_INTERFACES
+            and not name.startswith(self._EXCLUDED_INTERFACE_PREFIXES)
+        }

napalm_device_types/kernel.py

@@ -0,0 +1,168 @@
+# -*- coding: utf-8 -*-
+"""What the running kernel has built and loaded.
+
+A kernel CVE's exploitability often hangs on code that is simply not there --
+a module that is neither loaded nor shipped, a subsystem the kernel was built
+without. Reading that is identical on every Linux host, so the command and its
+parse live here once and a driver only carries the command across: SSH,
+an API's exec endpoint, whatever it has.
+
+The command is read-only and needs no privileges. It frames its report and
+sends it gzipped and base64-encoded, for two reasons: nothing in the payload can
+then look like a shell prompt to a screen-scraping transport, and a kernel's
+build configuration (~300 kB on a distribution kernel) crosses as a fifth of
+that.
+
+What the four lists mean for a module, and why "not loaded" alone is never
+"absent": a module that is not loaded can still be loaded on demand -- by an
+attacker too, where autoloading reaches it. Only a module that is neither
+loaded, nor compiled in, nor shipped for this kernel is one it cannot have.
+"""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import gzip
+import zlib
+from typing import Dict, List, Optional, TYPE_CHECKING
+
+from napalm_device_types.models import KernelFactsDict
+
+_BEGIN = "KFACTS_BEGIN"
+_END = "KFACTS_END"
+
+#: One line, POSIX ``sh``, read-only. The frame markers are printed in two
+#: halves so that a transport which echoes the command does not show them early.
+KERNEL_FACTS_COMMAND = (
+    "r=$(uname -r); m=/lib/modules/$r; "
+    "printf '%s%s\\n' KFACTS_ BEGIN; "
+    "{ echo '[release]'; echo \"$r\"; "
+    "if [ -r /proc/modules ]; then echo '[loaded]'; cut -d' ' -f1 /proc/modules; fi; "
+    "if [ -r $m/modules.builtin ]; then echo '[builtin]'; cat $m/modules.builtin; fi; "
+    "if [ -r $m/modules.dep ]; then echo '[available]'; cut -d: -f1 $m/modules.dep; fi; "
+    "if [ -r /boot/config-$r ]; then echo '[config]'; grep '^CONFIG_' /boot/config-$r; "
+    "elif [ -r /proc/config.gz ]; then echo '[config]'; zcat /proc/config.gz | grep '^CONFIG_'; fi; "
+    "} 2>/dev/null | gzip -c | base64; "
+    "printf '%s%s\\n' KFACTS_ END"
+)
+
+
+def module_name(raw: str) -> str:
+    """A module as the kernel names it: no path, no ``.ko`` suffix, ``_`` for ``-``.
+
+    ``kernel/net/can/can-raw.ko.zst`` and ``can_raw`` are the same module; the
+    kernel itself treats dash and underscore alike.
+    """
+    base = raw.strip().rsplit("/", 1)[-1]
+    suffix = base.find(".ko")
+    if suffix != -1:
+        base = base[:suffix]
+    return base.replace("-", "_").lower()
+
+
+def _report(output: str) -> str:
+    lines = [line.strip() for line in output.splitlines()]
+    try:
+        start = lines.index(_BEGIN)
+        end = lines.index(_END, start)
+    except ValueError:
+        raise ValueError("no kernel facts in the output") from None
+    try:
+        packed = base64.b64decode("".join(lines[start + 1 : end]), validate=True)
+        return gzip.decompress(packed).decode()
+    except (binascii.Error, OSError, EOFError, zlib.error, UnicodeDecodeError) as exc:
+        raise ValueError(f"the kernel facts could not be decoded: {exc}") from exc
+
+
+def _sections(report: str) -> Dict[str, List[str]]:
+    sections: Dict[str, List[str]] = {}
+    current: Optional[List[str]] = None
+    for line in report.splitlines():
+        line = line.strip()
+        if line.startswith("[") and line.endswith("]"):
+            current = sections.setdefault(line[1:-1], [])
+        elif line and current is not None:
+            current.append(line)
+    return sections
+
+
+def _config(lines: List[str]) -> Dict[str, str]:
+    config: Dict[str, str] = {}
+    for line in lines:
+        option, sep, value = line.partition("=")
+        if not sep:
+            continue
+        if len(value) >= 2 and value[0] == value[-1] == '"':
+            value = value[1:-1]
+        config[option] = value
+    return config
+
+
+def parse_kernel_facts(output: str) -> KernelFactsDict:
+    """Parse what :data:`KERNEL_FACTS_COMMAND` printed.
+
+    A section the command did not print -- the file was missing or unreadable --
+    comes back ``None``, never empty.
+
+    :raises ValueError: when the output carries no intact report.
+    """
+    sections = _sections(_report(output))
+
+    def names(key: str) -> Optional[List[str]]:
+        if key not in sections:
+            return None
+        return sorted({module_name(line) for line in sections[key]})
+
+    release = sections.get("release") or [""]
+    return {
+        "release": release[0],
+        "loaded": names("loaded"),
+        "builtin": names("builtin"),
+        "available": names("available"),
+        "config": _config(sections["config"]) if "config" in sections else None,
+    }
+
+
+class KernelFactsMixin:
+    """Adds :meth:`get_kernel_facts` to a driver that can run a command on a Linux host.
+
+    The template form (README, "Function classes"): the reading and its parse are
+    the same everywhere, so they are concrete here, and a driver supplies only
+    :meth:`_run_kernel_facts_command` -- how a command reaches its host. Mixed in
+    by the drivers that can, not by :class:`~napalm_device_types.os.OSDriver`:
+    a Windows host is an OS driver too and has no Linux kernel to read, and
+    ``hasattr(driver, "get_kernel_facts")`` has to stay a truthful answer.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_kernel_facts_command(self, command: str) -> str:
+            """Run *command* on the host with ``sh`` and return what it printed."""
+            ...
+
+    def get_kernel_facts(self) -> KernelFactsDict:
+        """
+        Returns what the running kernel has built and loaded.
+
+        * release (string) - ``uname -r``
+        * loaded (list or None) - loaded modules, from ``/proc/modules``
+        * builtin (list or None) - modules compiled into the kernel image
+        * available (list or None) - modules shipped for this kernel
+        * config (dict or None) - the build configuration's set options
+
+        ``None`` means the source could not be read.
+
+        Example::
+
+            {
+                "release": "6.1.0-25-amd64",
+                "loaded": ["nf_tables", "tipc"],
+                "builtin": ["tcp_cubic"],
+                "available": ["can_raw", "nf_tables", "tipc"],
+                "config": {"CONFIG_TIPC": "m", "CONFIG_HZ": "250"},
+            }
+
+        :raises ValueError: if the host's output carried no intact report.
+        """
+        return parse_kernel_facts(self._run_kernel_facts_command(KERNEL_FACTS_COMMAND))

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/mac_acl.py

@@ -0,0 +1,79 @@
+# -*- coding: utf-8 -*-
+"""MAC-based access control, per SSID on an AP and per port on a switch.
+
+The reader is identical either way; only the writer is AP-specific so far.
+
+Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
+Nothing exists at runtime until a concrete driver implements it, so mixing
+this class in can never shadow a working implementation from a sibling base.
+"""
+
+from __future__ import annotations
+
+from typing import Dict, List, TYPE_CHECKING
+
+from napalm_device_types.models import MACACLDict
+
+
+class MacAclMixin:
+    if TYPE_CHECKING:
+
+        def get_mac_acl(self) -> Dict[str, MACACLDict]:
+            """
+            Returns the MAC-address-based access control lists configured per SSID.
+
+            Keys are SSID names.  Each value contains:
+
+            * ssid (string) - SSID name (repeated for convenience)
+            * policy (string) - ACL mode:
+
+              * ``"allow"`` – whitelist: only listed MACs may associate
+              * ``"deny"``  – blacklist: listed MACs are blocked
+              * ``"disabled"`` – no MAC filtering active
+
+            * entries (list) - ACL entries, each with:
+
+              * mac (string) - MAC address (normalised, colon-separated)
+              * action (string) - ``"allow"`` or ``"deny"``
+              * description (string) - optional human-readable label
+
+            Example::
+
+                {
+                    "CorpWiFi": {
+                        "name": "CorpWiFi",
+                        "policy": "allow",
+                        "entries": [
+                            {"mac": "AA:BB:CC:DD:EE:01", "action": "allow", "description": "CEO-Laptop"},
+                            {"mac": "AA:BB:CC:DD:EE:02", "action": "allow", "description": "CFO-Laptop"},
+                        ],
+                    },
+                    "GuestNet": {
+                        "name": "GuestNet",
+                        "policy": "deny",
+                        "entries": [
+                            {"mac": "DE:AD:BE:EF:00:01", "action": "deny", "description": "blocked device"},
+                        ],
+                    },
+                }
+            """
+            ...
+
+        def push_mac_acl(self, ssid_name: str, mode: str, macs: List[str]) -> None:
+            """
+            Rewrites the MAC-address access control list for a single SSID.
+
+            Full-rebuild semantics: replaces whatever ACL state currently exists
+            for *ssid_name* with *mode* + *macs* — not a diff/patch.
+
+            :param ssid_name: SSID name to apply the ACL to.
+            :param mode: ``"off"`` | ``"whitelist"`` | ``"blacklist"``.
+            :param macs: MAC addresses for the active list. Ignored when ``mode == "off"``.
+            :raises NotImplementedError: If the driver does not support MAC ACL push.
+
+            Example::
+
+                driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"])
+                driver.push_mac_acl("GuestNet", "off", [])
+            """
+            ...

napalm_device_types/media.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+"""
+Abstract base class for networked media players.
+
+Usage::
+
+    from napalm_device_types import MediaDriver
+
+    class SonosDriver(MediaDriver):
+        ...
+"""
+
+from typing import TYPE_CHECKING, Any, Dict, List
+
+from napalm_device_types.base import DeviceTypeDriver
+
+
+class MediaDriver(DeviceTypeDriver):
+    """
+    Abstract intermediate driver for speakers, streamers and media renderers
+    (e.g. Sonos, Chromecast, Squeezebox, UPnP/DLNA renderers).
+
+    Like :class:`~napalm_device_types.phone.PhoneDriver`, this exists so an
+    endpoint stops being filed under ``AccessPointDriver`` for want of anywhere
+    better. A speaker has no SSIDs, no radio configuration and no AP profile; it
+    has a transport state and a volume.
+    """
+
+    #: Stable key netOrk exposes as ``device_class``. The order in which a
+    #: driver lists its role bases is the ranking; see
+    #: :func:`napalm_device_types.roles.primary_role_of`.
+    ROLE: str = "media"
+    TYPE_LABEL: str = "Media"
+
+    if TYPE_CHECKING:
+
+        def get_playback_state(self) -> Dict[str, Any]:
+            """
+            Returns the transport state of the renderer.
+
+            * state (string) - ``"playing"``, ``"paused"``, ``"stopped"``, or ``"transitioning"``
+            * source (string) - input or service currently selected
+            """
+            ...
+
+        def get_volume(self) -> int:
+            """Returns the current output volume, 0-100."""
+            ...
+
+        def set_volume(self, level: int) -> None:
+            """Sets the output volume, 0-100."""
+            ...
+
+        def get_zone_info(self) -> List[Dict[str, Any]]:
+            """
+            Returns the zones or groups this renderer participates in.
+
+            Each entry contains:
+
+            * name (string) - zone/room name
+            * coordinator (bool) - whether this device leads the group
+            * members (list) - names of the other devices in the group
+            """
+            ...

napalm_device_types/models.py

@@ -298,6 +298,39 @@ class NATTranslationDict(TypedDict):
     age: float
 
 
+class KernelFactsDict(TypedDict):
+    """What the running kernel has built and loaded (``KernelFactsMixin.get_kernel_facts``).
+
+    ``None`` means *could not be read*; an empty list means *read, and there is
+    nothing*. The difference is what lets a consumer say "this module cannot be
+    loaded on this kernel" rather than "we did not look".
+
+    Module names are normalised by :func:`napalm_device_types.kernel.module_name`:
+    no path, no ``.ko`` suffix, ``-`` folded to ``_``.
+    """
+
+    release: str  # uname -r
+    loaded: Optional[List[str]]  # /proc/modules
+    builtin: Optional[List[str]]  # modules.builtin -- compiled into the kernel image
+    available: Optional[List[str]]  # modules.dep -- shipped as loadable modules
+    config: Optional[Dict[str, str]]  # build configuration, set options only; quotes stripped
+
+
+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
@@ -314,6 +347,133 @@ class SessionDict(TypedDict):
     age: float
 
 
+class FirewallRuleDict(TypedDict):
+    """A single firewall filter rule, in vendor-neutral form.
+
+    `description` is the stable matching key across get_firewall_rules()/
+    diff_firewall_rules()/apply_firewall_rule() -- firewall vendors
+    generally don't expose an ID a caller can pre-assign, so the rule's
+    human description is what ties a "desired" rule to its "live"
+    counterpart. `source_net`/`source_port`/`destination_net`/
+    `destination_port` are plain strings (comma-joined by the caller if a
+    rule references multiple aliases) -- driver methods never expand or
+    split them.
+    """
+
+    uuid: str
+    description: str
+    action: str
+    interface: str
+    direction: str
+    protocol: str
+    source_net: str
+    source_port: str
+    destination_net: str
+    destination_port: str
+    enabled: bool
+    quick: bool
+    log: bool
+
+
+class FirewallRuleUpdateDict(TypedDict):
+    uuid: str
+    rule: FirewallRuleDict
+    changed_fields: List[str]
+
+
+class FirewallRuleDiffDict(TypedDict):
+    add: List[FirewallRuleDict]
+    update: List[FirewallRuleUpdateDict]
+
+
+class DhcpReservationDict(TypedDict):
+    """A single static DHCP reservation (MAC -> IP), in vendor-neutral form.
+
+    `mac` is the stable matching key across get_dhcp_reservations()/
+    diff_dhcp_reservations()/apply_dhcp_reservation() -- unlike firewall
+    rules, a reservation has a natural identity, and it is the MAC address.
+    It is compared after normalisation (see ``dhcp.normalize_mac``), so
+    devices that report ``AA-BB-CC-DD-EE-01`` still match a desired
+    ``aa:bb:cc:dd:ee:01``.
+
+    `subnet` is the CIDR the reservation lives in. It is a *compared* field,
+    not part of the key: a host that moves to another VLAN keeps its MAC, and
+    that is an update of the existing reservation rather than a second one.
+    """
+
+    uuid: str
+    mac: str
+    ip: str
+    hostname: str
+    description: str
+    subnet: str
+
+
+class DhcpReservationUpdateDict(TypedDict):
+    uuid: str
+    reservation: DhcpReservationDict
+    changed_fields: List[str]
+
+
+class DhcpReservationDiffDict(TypedDict):
+    add: List[DhcpReservationDict]
+    update: List[DhcpReservationUpdateDict]
+
+
+class DhcpOptionDataDict(TypedDict, total=False):
+    """DHCPv4 options carried by a subnet, by their RFC/Kea names.
+
+    Only options netOrk actually models are listed. `domain_search` (option
+    119) is the reason this type exists: it is the one option that cannot be
+    expressed anywhere else in the stack, and no DHCP server autocollects it.
+
+    Every field is optional and an absent key means "do not manage this
+    option" -- distinct from an empty list, which means "manage it, and the
+    desired value is empty". A driver must preserve options it was not given.
+    """
+
+    routers: List[str]
+    domain_name_servers: List[str]
+    domain_name: str
+    domain_search: List[str]
+    ntp_servers: List[str]
+
+
+class DhcpSubnetDict(TypedDict):
+    """A DHCPv4 subnet served by the device, in vendor-neutral form.
+
+    `subnet` (the CIDR) is the stable matching key, the way `mac` is for a
+    reservation. Renaming is not a thing a subnet does; changing its CIDR
+    makes it a different subnet.
+
+    `pools` are address ranges in ``"start-end"`` form, the shape both Kea
+    and ISC DHCP use.
+
+    `option_data` carries the per-subnet DHCP options. Servers that
+    autocollect some of them (Kea fills `routers`, `domain_name_servers` and
+    `ntp_servers` when ``option_data_autocollect`` is on) still never
+    autocollect `domain_name` or `domain_search`.
+    """
+
+    uuid: str
+    subnet: str
+    description: str
+    pools: List[str]
+    option_data: DhcpOptionDataDict
+    match_client_id: bool
+
+
+class DhcpSubnetUpdateDict(TypedDict):
+    uuid: str
+    subnet: DhcpSubnetDict
+    changed_fields: List[str]
+
+
+class DhcpSubnetDiffDict(TypedDict):
+    add: List[DhcpSubnetDict]
+    update: List[DhcpSubnetUpdateDict]
+
+
 class VPNTunnelDict(TypedDict):
     type: str
     local_endpoint: str
@@ -343,16 +503,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
@@ -385,7 +535,7 @@ class VMNICDict(TypedDict):
 
 class VMDict(TypedDict):
     name: str
-    vmid: int
+    vmid: str
     status: str
     vcpus: int
     memory: int
@@ -395,9 +545,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
@@ -406,6 +564,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):
@@ -713,6 +879,7 @@ class NICConfigDict(TypedDict):
     vlan_tag: NotRequired[int | None]  # Access VLAN (None = untagged)
     trunk_vlan_tags: NotRequired[list[int]]  # Trunk VLAN list (alternative to vlan_tag)
     dhcp: NotRequired[bool]  # Enable DHCP (default True for first NIC, False for others)
+    mac: NotRequired[str]  # Explicit MAC address (omit to let the hypervisor auto-generate one)
 
 
 class VMProvisionResultDict(TypedDict):
@@ -730,3 +897,75 @@ class VMStatusDict(TypedDict):
     ip_address: NotRequired[str]  # Management NIC IP (absent if VM has no IP or is stopped)
     hostname: NotRequired[str]  # Hostname resolved from IP (if available)
     mac_address: NotRequired[str]  # MAC of management NIC
+
+
+class NetworkTargetDict(TypedDict):
+    """A selectable network target for a new VM's NIC (``NICConfigDict.bridge``).
+
+    Distinguishes real bridges (Linux or OVS) from SDN network segments (vnets),
+    and tells the caller whether a separate ``vlan_tag`` may be applied on top:
+    - Linux bridge: vlan_aware reflects the bridge's own ``bridge_vlan_aware`` flag.
+    - OVS bridge: always vlan_aware (OVS bridges tag per-port regardless of a
+      dedicated "VLAN aware" setting).
+    - SDN vnet: never vlan_aware — the VLAN is already fixed by the vnet's zone/tag,
+      so a NIC attached to it must not also carry a ``vlan_tag``. That fixed VLAN
+      is surfaced via ``fixed_vlan_tag`` instead, for display purposes.
+    """
+
+    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
+
+
+class StorageTargetDict(TypedDict):
+    """A selectable storage pool for a new VM's root disk (``create_vm_from_cloud_init``'s
+    ``storage`` argument), scoped to the specific node the VM will be created on —
+    a storage restricted to other cluster nodes must not appear here.
+    """
+
+    name: str  # Storage pool name, usable directly as create_vm_from_cloud_init(storage=...)
+    type: str  # Backend type: "dir", "lvmthin", "zfspool", "nfs", etc.
+    total_gb: float  # Total capacity in gigabytes
+    available_gb: float  # Free capacity in gigabytes
+
+
+class VMCpuTypeDict(TypedDict):
+    """A virtual CPU model a new VM may be given (``create_vm_from_cloud_init``'s
+    ``cpu_type`` argument), judged against the specific node the VM will be
+    created on.
+
+    ``features`` uses the flag names of Linux's ``/proc/cpuinfo`` (``avx``,
+    ``avx2``, ``aes`` ...), so a caller can ask "does this model give the guest
+    AVX?" without knowing the hypervisor's model names. A model that passes the
+    host CPU through lists that CPU's own flags.
+    """
+
+    name: str  # Model name, usable directly as create_vm_from_cloud_init(cpu_type=...)
+    description: str  # One line on what the model is for, for a picker
+    features: List[str]  # cpuinfo flags the guest is guaranteed to see
+    available: bool  # False when this node's CPU cannot run the model
+    default: bool  # The model create_vm_from_cloud_init uses when cpu_type is None
+
+
+# ---------------------------------------------------------------------------
+# Ping sweep  (shared across device types)
+# ---------------------------------------------------------------------------
+
+
+class PingSweepEntryDict(TypedDict):
+    """Outcome of a single ``ping`` inside a sweep (see ``PingSweepMixin``)."""
+
+    ip: str                  # destination that was probed
+    alive: bool              # True if at least one probe was answered
+    rtt_ms: Optional[float]  # average round-trip time in ms; None if unreachable
+    error: NotRequired[str]  # driver/transport error for this destination
+
+
+class PingSweepResultDict(TypedDict):
+    """Result of a ``ping_sweep()`` call."""
+
+    entries: List[PingSweepEntryDict]  # one entry per probed destination, in input order
+    scanned: int                       # destinations actually probed
+    alive_count: int                   # entries with alive=True
+    truncated: bool                    # True if targets were dropped at the sweep cap

napalm_device_types/nat_vpn.py

@@ -0,0 +1,118 @@
+# -*- coding: utf-8 -*-
+"""Address translation and VPN tunnels.
+
+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
+this class in can never shadow a working implementation from a sibling base.
+"""
+
+from __future__ import annotations
+
+from typing import Dict, List, TYPE_CHECKING
+
+from napalm_device_types.models import NATTranslationDict, PortForwardDict, VPNTunnelDict
+
+
+class NatVpnMixin:
+    if TYPE_CHECKING:
+
+        def get_nat_translations(self) -> List[NATTranslationDict]:
+            """
+            Returns a list of active NAT translation entries.
+
+            Each entry contains:
+
+            * protocol (string) - ``"tcp"``, ``"udp"``, ``"icmp"``
+            * inside_local (string) - original source address (IP or IP:port)
+            * inside_global (string) - translated source address (IP or IP:port)
+            * outside_local (string) - destination as seen from inside
+            * outside_global (string) - actual destination address
+            * age (float) - translation entry age in seconds
+
+            Example::
+
+                [
+                    {
+                        "protocol": "tcp",
+                        "inside_local": "192.168.1.10:54321",
+                        "inside_global": "203.0.113.1:54321",
+                        "outside_local": "1.1.1.1:443",
+                        "outside_global": "1.1.1.1:443",
+                        "age": 120.5,
+                    }
+                ]
+            """
+            ...
+
+        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.
+
+            Keys are tunnel names or identifiers.  Each value contains:
+
+            * type (string) - tunnel type: ``"IPsec"``, ``"SSL"``, ``"GRE"``, ``"WireGuard"``
+            * local_endpoint (string) - local tunnel endpoint IP
+            * remote_endpoint (string) - remote tunnel endpoint IP
+            * is_up (bool) - whether the tunnel is operationally up
+            * uptime (int) - tunnel uptime in seconds (0 if down)
+            * bytes_in (int) - total bytes received through the tunnel
+            * bytes_out (int) - total bytes sent through the tunnel
+
+            Example::
+
+                {
+                    "vpn-to-branch": {
+                        "type": "IPsec",
+                        "local_endpoint": "203.0.113.1",
+                        "remote_endpoint": "198.51.100.1",
+                        "is_up": True,
+                        "uptime": 86400,
+                        "bytes_in": 104857600,
+                        "bytes_out": 52428800,
+                    }
+                }
+            """
+            ...

napalm_device_types/os.py

@@ -10,27 +10,23 @@ Usage::
             ...
 """
 
-import re
-from typing import List, Optional
+from typing import List, Optional, TYPE_CHECKING
 from napalm_device_types.base import DeviceTypeDriver
-from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
+from napalm_device_types.updates import UpdateMixin
+from napalm_device_types.services import ServiceControlMixin
+from napalm_device_types.packages import PackageManagementMixin
+from napalm_device_types.health_metrics import HealthMetricsMixin
 from napalm_device_types.models import (
-    ApplyUpdatesResultDict,
     CronJobDict,
     DeviceActionResultDict,
     DockerInfoDict,
-    HealthMetricsDict,
-    PackageDict,
     ProcessDict,
-    ServiceDict,
     SNMPConfigDict,
-    UpdateDict,
     UserDict,
 )
 
 
-class OSDriver(DeviceTypeDriver):
-    TYPE_LABEL: str = "OS"
+class OSDriver(UpdateMixin, ServiceControlMixin, PackageManagementMixin, HealthMetricsMixin, DeviceTypeDriver):
     """
     Abstract intermediate driver for general-purpose operating systems
     (e.g. Linux, BSD, macOS).
@@ -39,391 +35,280 @@ class OSDriver(DeviceTypeDriver):
     operations that concrete drivers must implement.
     """
 
+    #: Stable key netOrk exposes as ``device_class``. The order in which a
+    #: driver lists its role bases is the ranking; see
+    #: :func:`napalm_device_types.roles.primary_role_of`.
+    ROLE: str = "linux"
+    TYPE_LABEL: str = "OS"
+
     # Interfaces matching this pattern are excluded from health-metric collection.
-    _SNMP_SKIP_IF: re.Pattern = IF_SKIP_DEFAULT
     # Set True for drivers where the out-error counter actually reports drops (e.g. OPNsense).
-    _SNMP_TX_ERR_IS_DROP: bool = False
 
-    @classmethod
-    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
-        """Collect SNMP health metrics (CPU, memory, uptime, load, interfaces).
-
-        :param snmp_get:  async callable ``(oid: str) -> Optional[str]``
-        :param snmp_walk: async callable ``(oid: str) -> Dict[str, str]``
-        """
-        return await collect_ucd_metrics(
-            snmp_get, snmp_walk,
-            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
-            if_skip=cls._SNMP_SKIP_IF,
-        )
 
     # ------------------------------------------------------------------
     # Package management
     # ------------------------------------------------------------------
 
-    def get_packages(self) -> List[PackageDict]:
-        """
-        Returns a list of all installed software packages.
+    if TYPE_CHECKING:
 
-        Each entry contains:
 
-        * name (string) - package name
-        * version (string) - installed version string
-        * installed (bool) - always ``True`` for this method
-        * description (string) - short package description
-        * size (int) - installed size in bytes; ``0`` if unavailable
-        * source (string) - package source / repository name; empty string if unavailable
 
-        Example::
 
-            [
-                {
-                    "name": "openssh-server",
-                    "version": "1:9.2p1-2+deb12u2",
-                    "installed": True,
-                    "description": "secure shell (SSH) server, for secure access from remote machines",
-                    "size": 524288,
-                    "source": "Debian",
-                },
-            ]
-        """
-        raise NotImplementedError
+        # ------------------------------------------------------------------
+        # Service management
+        # ------------------------------------------------------------------
 
-    def get_pending_updates(self) -> List[UpdateDict]:
-        """
-        Returns a list of packages that have a newer version available.
 
-        Each entry contains:
+        # ------------------------------------------------------------------
+        # Users
+        # ------------------------------------------------------------------
 
-        * name (string) - package name
-        * current_version (string) - currently installed version
-        * new_version (string) - version available in the repository
+        def get_users(self) -> List[UserDict]:
+            """
+            Returns a list of local OS user accounts.
 
-        Example::
+            Each entry contains:
 
-            [
-                {
-                    "name": "openssh-server",
-                    "current_version": "1:9.2p1-2+deb12u1",
-                    "new_version": "1:9.2p1-2+deb12u2",
-                },
-            ]
-        """
-        raise NotImplementedError
+            * username (string) - login name
+            * uid (int) - numeric user ID
+            * gid (int) - primary group ID
+            * home (string) - home directory path
+            * shell (string) - login shell path
+            * groups (list of strings) - all supplementary group names
 
-    def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict:
-        """
-        Upgrades the given packages to the newest available version.
+            Example::
 
-        Only packages that are already installed may be upgraded; this method
-        does **not** install new packages.  Pass an empty list to upgrade
-        **all** packages that have pending updates.
-
-        :param packages: List of package names to upgrade.  Each name must
-            match ``^[a-zA-Z0-9_\\-\\+\\.]+$``; a :exc:`ValueError` is raised
-            for any name that does not conform.
-        :returns: A dict with:
-
-            * success (bool) – ``True`` if the package manager exited without error
-            * output (string) – combined stdout / stderr from the package manager
-            * error (string, optional) – short error message when *success* is ``False``
-
-        :raises ValueError: If any package name fails the safety check.
-
-        Example::
-
-            result = driver.apply_updates(["openssh-server", "curl"])
-            # → {"success": True, "output": "Reading package lists...\\n..."}
-
-            # Upgrade everything:
-            result = driver.apply_updates([])
-        """
-        raise NotImplementedError
-
-    # ------------------------------------------------------------------
-    # Service management
-    # ------------------------------------------------------------------
-
-    def get_services(self) -> List[ServiceDict]:
-        """
-        Returns a list of system services and their current state.
-
-        Each entry contains:
-
-        * name (string) - service unit name (without ``.service`` suffix)
-        * running (bool) - ``True`` if the service is currently active
-        * enabled (bool) - ``True`` if the service starts automatically on boot
-        * pid (int) - main process ID; ``0`` if not running
-
-        Example::
-
-            [
-                {
-                    "name": "ssh",
-                    "running": True,
-                    "enabled": True,
-                    "pid": 1234,
-                },
-                {
-                    "name": "cron",
-                    "running": True,
-                    "enabled": True,
-                    "pid": 5678,
-                },
-            ]
-        """
-        raise NotImplementedError
-
-    # ------------------------------------------------------------------
-    # Users
-    # ------------------------------------------------------------------
-
-    def get_users(self) -> List[UserDict]:
-        """
-        Returns a list of local OS user accounts.
-
-        Each entry contains:
-
-        * username (string) - login name
-        * uid (int) - numeric user ID
-        * gid (int) - primary group ID
-        * home (string) - home directory path
-        * shell (string) - login shell path
-        * groups (list of strings) - all supplementary group names
-
-        Example::
-
-            [
-                {
-                    "username": "admin",
-                    "uid": 1000,
-                    "gid": 1000,
-                    "home": "/home/admin",
-                    "shell": "/bin/bash",
-                    "groups": ["sudo", "docker", "adm"],
-                },
-            ]
-        """
-        raise NotImplementedError
-
-    # ------------------------------------------------------------------
-    # Processes
-    # ------------------------------------------------------------------
-
-    def get_processes(self) -> List[ProcessDict]:
-        """
-        Returns a snapshot of currently running processes.
-
-        Each entry contains:
-
-        * pid (int) - process ID
-        * ppid (int) - parent process ID
-        * user (string) - effective user name
-        * cpu (float) - CPU utilisation percentage
-        * memory (float) - RSS as a percentage of total RAM
-        * vsz (int) - virtual memory size in KiB
-        * rss (int) - resident set size in KiB
-        * tty (string) - controlling terminal; empty string if none
-        * state (string) - process state: ``"R"`` running, ``"S"`` sleeping,
-          ``"D"`` uninterruptible, ``"Z"`` zombie, ``"T"`` stopped, etc.
-        * started (string) - start time as printed by ``ps`` (e.g. ``"12:34"`` or ``"May28"``)
-        * command (string) - full command line
-
-        Example::
-
-            [
-                {
-                    "pid": 1,
-                    "ppid": 0,
-                    "user": "root",
-                    "cpu": 0.0,
-                    "memory": 0.1,
-                    "vsz": 168576,
-                    "rss": 13312,
-                    "tty": "",
-                    "state": "S",
-                    "started": "May28",
-                    "command": "/sbin/init",
-                },
-            ]
-        """
-        raise NotImplementedError
-
-    # ------------------------------------------------------------------
-    # Cron jobs
-    # ------------------------------------------------------------------
-
-    def get_cron_jobs(self) -> List[CronJobDict]:
-        """
-        Returns scheduled cron tasks from all user crontabs and ``/etc/cron.d``.
-
-        Each entry contains:
-
-        * user (string) - owner of the crontab entry
-        * schedule (string) - five-field cron expression (e.g. ``"0 * * * *"``)
-        * command (string) - shell command to execute
-        * description (string, optional) - inline comment text if present
-
-        Example::
-
-            [
-                {
-                    "user": "root",
-                    "schedule": "0 4 * * *",
-                    "command": "/usr/local/bin/backup.sh",
-                    "description": "nightly backup",
-                },
-            ]
-        """
-        raise NotImplementedError
-
-    # ------------------------------------------------------------------
-    # SNMP
-    # ------------------------------------------------------------------
-
-    def get_snmp_config(self) -> Optional[SNMPConfigDict]:
-        """
-        Returns the SNMP agent configuration currently active on the device,
-        or ``None`` if no SNMP daemon is running or detectable.
-
-        The returned dictionary contains:
-
-        * running (bool) - whether the SNMP daemon is currently active
-        * community (string) - the read community string (e.g. ``"public"``)
-        * port (int) - the UDP port the agent listens on (default ``161``)
-        * version (string) - highest supported SNMP version: ``"1"``, ``"2c"``, or ``"3"``
-
-        Example::
-
-            # snmpd running with community "public":
-            {
-                "running": True,
-                "community": "public",
-                "port": 161,
-                "version": "2c",
-            }
-
-            # snmpd not installed / not running:
-            None
-        """
-        raise NotImplementedError
-
-    # ------------------------------------------------------------------
-    # Docker
-    # ------------------------------------------------------------------
-
-    def get_docker_info(self) -> DockerInfoDict:
-        """
-        Returns information about the local Docker environment.
-
-        If Docker is not installed or the current user lacks access to the
-        Docker socket, returns ``{"available": False}``.  When the user has
-        no socket permission, ``permission_denied`` is additionally set to
-        ``True``.
-
-        When Docker is available the dict contains:
-
-        * available (bool) - always ``True``
-        * version (string) - Docker Engine version string
-        * containers (list) - all containers (running and stopped), each with:
-
-          * id (string) - short container ID
-          * name (string) - container name(s)
-          * image (string) - image reference
-          * image_version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
-          * command (string) - entrypoint / command string
-          * created (string) - creation timestamp string
-          * status (string) - human-readable status (e.g. ``"Up 3 hours"``)
-          * ports (string) - port mapping string
-          * state (string) - ``"running"``, ``"exited"``, ``"paused"``, etc.
-
-        * images (list) - local images, each with:
-
-          * id (string) - short image ID
-          * repository (string) - image repository
-          * tag (string) - image tag
-          * size (string) - human-readable size string (e.g. ``"187MB"``)
-          * created (string) - creation timestamp string
-          * version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
-
-        * volumes (list) - Docker volumes, each with:
-
-          * name (string) - volume name
-          * driver (string) - volume driver
-          * mountpoint (string) - host filesystem path
-          * scope (string) - ``"local"`` or ``"global"``
-
-        * networks (list) - Docker networks, each with:
-
-          * id (string) - short network ID
-          * name (string) - network name
-          * driver (string) - network driver (e.g. ``"bridge"``, ``"host"``, ``"overlay"``)
-          * scope (string) - network scope
-          * ipv6 (string) - ``"true"`` if IPv6 is enabled
-          * internal (string) - ``"true"`` if the network is internal
-
-        * outdated_images (list of strings) - image names where the local digest
-          differs from the latest remote digest; empty list if all images are
-          current or update checks could not be performed.
-
-        Example::
-
-            # Docker not installed:
-            {"available": False}
-
-            # Docker installed, no socket permission:
-            {"available": False, "permission_denied": True}
-
-            # Docker available:
-            {
-                "available": True,
-                "version": "Docker version 27.3.1, build ce12230",
-                "containers": [
+                [
                     {
-                        "id": "a1b2c3d4e5f6",
-                        "name": "my-app",
-                        "image": "nginx:latest",
-                        "image_version": "1.27.0",
-                        "command": "nginx -g 'daemon off;'",
-                        "created": "2026-05-28 10:00:00 +0000 UTC",
-                        "status": "Up 3 days",
-                        "ports": "0.0.0.0:80->80/tcp",
-                        "state": "running",
+                        "username": "admin",
+                        "uid": 1000,
+                        "gid": 1000,
+                        "home": "/home/admin",
+                        "shell": "/bin/bash",
+                        "groups": ["sudo", "docker", "adm"],
                     },
-                ],
-                "images": [...],
-                "volumes": [],
-                "networks": [...],
-                "outdated_images": ["nginx:latest"],
-            }
-        """
-        raise NotImplementedError
+                ]
+            """
+            ...
 
-    # ------------------------------------------------------------------
-    # Generic device actions
-    # ------------------------------------------------------------------
+        # ------------------------------------------------------------------
+        # Processes
+        # ------------------------------------------------------------------
 
-    def run_device_action(self, action: str) -> DeviceActionResultDict:
-        """
-        Executes a named administrative action on the device.
+        def get_processes(self) -> List[ProcessDict]:
+            """
+            Returns a snapshot of currently running processes.
 
-        This method is an extensibility point for driver-specific one-off
-        operations that do not fit any other NAPALM API method.  Each driver
-        documents the action names it supports.
+            Each entry contains:
 
-        :param action: Action identifier string (e.g. ``"fix_docker_permissions"``).
-        :raises NotImplementedError: If the driver does not implement this method.
-        :raises ValueError: If ``action`` is not a recognised action name for
-            this driver.
+            * pid (int) - process ID
+            * ppid (int) - parent process ID
+            * user (string) - effective user name
+            * cpu (float) - CPU utilisation percentage
+            * memory (float) - RSS as a percentage of total RAM
+            * vsz (int) - virtual memory size in KiB
+            * rss (int) - resident set size in KiB
+            * tty (string) - controlling terminal; empty string if none
+            * state (string) - process state: ``"R"`` running, ``"S"`` sleeping,
+              ``"D"`` uninterruptible, ``"Z"`` zombie, ``"T"`` stopped, etc.
+            * started (string) - start time as printed by ``ps`` (e.g. ``"12:34"`` or ``"May28"``)
+            * command (string) - full command line
 
-        :returns: A dict with:
+            Example::
 
-            * success (bool) – ``True`` if the action completed without error
-            * output (string) – human-readable output or status message
+                [
+                    {
+                        "pid": 1,
+                        "ppid": 0,
+                        "user": "root",
+                        "cpu": 0.0,
+                        "memory": 0.1,
+                        "vsz": 168576,
+                        "rss": 13312,
+                        "tty": "",
+                        "state": "S",
+                        "started": "May28",
+                        "command": "/sbin/init",
+                    },
+                ]
+            """
+            ...
 
-        Example::
+        # ------------------------------------------------------------------
+        # Cron jobs
+        # ------------------------------------------------------------------
 
-            result = driver.run_device_action("fix_docker_permissions")
-            # → {"success": True, "output": "Added 'pi' to the docker group. Reconnect..."}
-        """
-        raise NotImplementedError
+        def get_cron_jobs(self) -> List[CronJobDict]:
+            """
+            Returns scheduled cron tasks from all user crontabs and ``/etc/cron.d``.
+
+            Each entry contains:
+
+            * user (string) - owner of the crontab entry
+            * schedule (string) - five-field cron expression (e.g. ``"0 * * * *"``)
+            * command (string) - shell command to execute
+            * description (string, optional) - inline comment text if present
+
+            Example::
+
+                [
+                    {
+                        "user": "root",
+                        "schedule": "0 4 * * *",
+                        "command": "/usr/local/bin/backup.sh",
+                        "description": "nightly backup",
+                    },
+                ]
+            """
+            ...
+
+        # ------------------------------------------------------------------
+        # SNMP
+        # ------------------------------------------------------------------
+
+        def get_snmp_config(self) -> Optional[SNMPConfigDict]:
+            """
+            Returns the SNMP agent configuration currently active on the device,
+            or ``None`` if no SNMP daemon is running or detectable.
+
+            The returned dictionary contains:
+
+            * running (bool) - whether the SNMP daemon is currently active
+            * community (string) - the read community string (e.g. ``"public"``)
+            * port (int) - the UDP port the agent listens on (default ``161``)
+            * version (string) - highest supported SNMP version: ``"1"``, ``"2c"``, or ``"3"``
+
+            Example::
+
+                # snmpd running with community "public":
+                {
+                    "running": True,
+                    "community": "public",
+                    "port": 161,
+                    "version": "2c",
+                }
+
+                # snmpd not installed / not running:
+                None
+            """
+            ...
+
+        # ------------------------------------------------------------------
+        # Docker
+        # ------------------------------------------------------------------
+
+        def get_docker_info(self) -> DockerInfoDict:
+            """
+            Returns information about the local Docker environment.
+
+            If Docker is not installed or the current user lacks access to the
+            Docker socket, returns ``{"available": False}``.  When the user has
+            no socket permission, ``permission_denied`` is additionally set to
+            ``True``.
+
+            When Docker is available the dict contains:
+
+            * available (bool) - always ``True``
+            * version (string) - Docker Engine version string
+            * containers (list) - all containers (running and stopped), each with:
+
+              * id (string) - short container ID
+              * name (string) - container name(s)
+              * image (string) - image reference
+              * image_version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
+              * command (string) - entrypoint / command string
+              * created (string) - creation timestamp string
+              * status (string) - human-readable status (e.g. ``"Up 3 hours"``)
+              * ports (string) - port mapping string
+              * state (string) - ``"running"``, ``"exited"``, ``"paused"``, etc.
+
+            * images (list) - local images, each with:
+
+              * id (string) - short image ID
+              * repository (string) - image repository
+              * tag (string) - image tag
+              * size (string) - human-readable size string (e.g. ``"187MB"``)
+              * created (string) - creation timestamp string
+              * version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
+
+            * volumes (list) - Docker volumes, each with:
+
+              * name (string) - volume name
+              * driver (string) - volume driver
+              * mountpoint (string) - host filesystem path
+              * scope (string) - ``"local"`` or ``"global"``
+
+            * networks (list) - Docker networks, each with:
+
+              * id (string) - short network ID
+              * name (string) - network name
+              * driver (string) - network driver (e.g. ``"bridge"``, ``"host"``, ``"overlay"``)
+              * scope (string) - network scope
+              * ipv6 (string) - ``"true"`` if IPv6 is enabled
+              * internal (string) - ``"true"`` if the network is internal
+
+            * outdated_images (list of strings) - image names where the local digest
+              differs from the latest remote digest; empty list if all images are
+              current or update checks could not be performed.
+
+            Example::
+
+                # Docker not installed:
+                {"available": False}
+
+                # Docker installed, no socket permission:
+                {"available": False, "permission_denied": True}
+
+                # Docker available:
+                {
+                    "available": True,
+                    "version": "Docker version 27.3.1, build ce12230",
+                    "containers": [
+                        {
+                            "id": "a1b2c3d4e5f6",
+                            "name": "my-app",
+                            "image": "nginx:latest",
+                            "image_version": "1.27.0",
+                            "command": "nginx -g 'daemon off;'",
+                            "created": "2026-05-28 10:00:00 +0000 UTC",
+                            "status": "Up 3 days",
+                            "ports": "0.0.0.0:80->80/tcp",
+                            "state": "running",
+                        },
+                    ],
+                    "images": [...],
+                    "volumes": [],
+                    "networks": [...],
+                    "outdated_images": ["nginx:latest"],
+                }
+            """
+            ...
+
+        # ------------------------------------------------------------------
+        # Generic device actions
+        # ------------------------------------------------------------------
+
+        def run_device_action(self, action: str) -> DeviceActionResultDict:
+            """
+            Executes a named administrative action on the device.
+
+            This method is an extensibility point for driver-specific one-off
+            operations that do not fit any other NAPALM API method.  Each driver
+            documents the action names it supports.
+
+            :param action: Action identifier string (e.g. ``"fix_docker_permissions"``).
+            :raises NotImplementedError: If the driver does not implement this method.
+            :raises ValueError: If ``action`` is not a recognised action name for
+                this driver.
+
+            :returns: A dict with:
+
+                * success (bool) – ``True`` if the action completed without error
+                * output (string) – human-readable output or status message
+
+            Example::
+
+                result = driver.run_device_action("fix_docker_permissions")
+                # → {"success": True, "output": "Added 'pi' to the docker group. Reconnect..."}
+            """
+            ...

napalm_device_types/packages.py

@@ -0,0 +1,158 @@
+# -*- coding: utf-8 -*-
+"""Listing and managing installed software.
+
+Identical on every device type that has a package manager -- this was five
+byte-identical copies before it moved here. A driver whose device offers no
+package management simply implements none of it.
+
+Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
+Nothing exists at runtime until a concrete driver implements it, so mixing
+this class in can never shadow a working implementation from a sibling base.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, TYPE_CHECKING
+
+from napalm_device_types.models import PackageDict
+
+
+class PackageManagementMixin:
+    if TYPE_CHECKING:
+
+        def get_packages(self) -> List[PackageDict]:
+            """
+            Returns all packages currently known to the device's package manager
+            (e.g. ``opkg`` on OpenWrt, ``apk`` on Alpine-based APs).
+
+            Each entry contains:
+
+            * name (string) - package name
+            * version (string) - installed or available version string
+            * installed (bool) - ``True`` if the package is currently installed
+            * description (string) - short package description
+            * size (int) - package size in bytes (0 if unknown)
+            * source (string) - repository / feed the package comes from
+
+            Example::
+
+                [
+                    {
+                        "name": "luci-app-statistics",
+                        "version": "git-24.001.00000-1",
+                        "installed": True,
+                        "description": "LuCI Statistics application",
+                        "size": 20480,
+                        "source": "openwrt/packages",
+                    },
+                    {
+                        "name": "collectd-mod-wireless",
+                        "version": "5.12.0-24",
+                        "installed": False,
+                        "description": "Wireless statistics plugin for collectd",
+                        "size": 8192,
+                        "source": "openwrt/packages",
+                    },
+                ]
+            """
+            ...
+
+        def install_package(self, name: str, version: str = "") -> None:
+            """
+            Installs a package on the device.
+
+            The method blocks until the installation is complete.  After it returns
+            successfully the package is available for use without a reboot
+            (where the underlying package manager supports this).
+
+            :param name: Package name as known to the package manager.
+            :param version: Exact version to install.  An empty string (default)
+                installs the latest available version.
+            :raises NotImplementedError: If the driver does not support package management.
+            :raises ValueError: If the package name is unknown or the version does
+                not exist in any configured feed.
+            :raises RuntimeError: If the installation fails on the device side
+                (e.g. dependency conflict, disk full).
+
+            Example::
+
+                driver.install_package("luci-app-statistics")
+                driver.install_package("collectd", version="5.12.0-24")
+            """
+            ...
+
+        def uninstall_package(self, name: str) -> None:
+            """
+            Removes an installed package from the device.
+
+            The method blocks until the removal is complete.
+
+            :param name: Package name to remove.
+            :raises NotImplementedError: If the driver does not support package management.
+            :raises ValueError: If the package is not currently installed.
+            :raises RuntimeError: If the removal fails on the device side
+                (e.g. other packages depend on it).
+
+            Example::
+
+                driver.uninstall_package("luci-app-statistics")
+            """
+            ...
+
+        def get_package_config(self, name: str) -> Dict[str, Any]:
+            """
+            Returns the current configuration of an installed package as a
+            dictionary.  The structure is package-specific.
+
+            :param name: Package name.
+            :raises NotImplementedError: If the driver does not support package management.
+            :raises ValueError: If the package is not installed.
+
+            Example::
+
+                driver.get_package_config("luci-app-statistics")
+                # →
+                {
+                    "collectd": {
+                        "enabled": True,
+                        "interval": 30,
+                    },
+                    "rrdtool": {
+                        "datadir": "/tmp/rrd",
+                        "stepsize": 30,
+                        "heartbeat": 60,
+                    },
+                }
+            """
+            ...
+
+        def set_package_config(self, name: str, config: Dict[str, Any]) -> None:
+            """
+            Writes a new configuration for an installed package.
+
+            The ``config`` dict must match the structure returned by
+            :meth:`get_package_config`.  Unknown keys are ignored or raise a
+            ``ValueError`` depending on the driver implementation.
+
+            Changes take effect immediately where the package supports live
+            reload; otherwise a package restart or device reboot may be
+            required – behaviour is driver-specific.
+
+            :param name: Package name.
+            :param config: New configuration as a nested dictionary.
+            :raises NotImplementedError: If the driver does not support package management.
+            :raises ValueError: If the package is not installed or the configuration
+                contains invalid values.
+            :raises RuntimeError: If the device rejects the configuration.
+
+            Example::
+
+                driver.set_package_config(
+                    "luci-app-statistics",
+                    {
+                        "collectd": {"enabled": True, "interval": 60},
+                        "rrdtool": {"datadir": "/tmp/rrd", "stepsize": 60, "heartbeat": 120},
+                    },
+                )
+            """
+            ...

napalm_device_types/phone.py

@@ -0,0 +1,71 @@
+# -*- coding: utf-8 -*-
+"""
+Abstract base class for IP telephony endpoints.
+
+Usage::
+
+    from napalm_device_types import PhoneDriver
+
+    class YealinkDriver(PhoneDriver):
+        ...
+"""
+
+from typing import TYPE_CHECKING, Any, Dict, List
+
+from napalm_device_types.base import DeviceTypeDriver
+
+
+class PhoneDriver(DeviceTypeDriver):
+    """
+    Abstract intermediate driver for desk phones, conference units and DECT
+    bases (e.g. Yealink T-series, Snom, Grandstream, Fanvil).
+
+    A phone is an endpoint, not infrastructure. It was worth separating from
+    ``AccessPointDriver`` — which some phone drivers used to inherit for want of
+    anywhere better — because netOrk offers access points for AP profiles,
+    wireless management and SSID drift, none of which a desk phone should appear
+    in. A WiFi-capable phone still reports its radio through its own methods; it
+    just is not an access point.
+    """
+
+    #: Stable key netOrk exposes as ``device_class``. The order in which a
+    #: driver lists its role bases is the ranking; see
+    #: :func:`napalm_device_types.roles.primary_role_of`.
+    ROLE: str = "phone"
+    TYPE_LABEL: str = "Phone"
+
+    if TYPE_CHECKING:
+
+        def get_sip_accounts(self) -> List[Dict[str, Any]]:
+            """
+            Returns the SIP registrations configured on the phone.
+
+            Each entry contains:
+
+            * account (string) - display label or line key
+            * user (string) - SIP user / extension
+            * server (string) - registrar host
+            * registered (bool) - whether the registration is currently active
+
+            Example::
+
+                [
+                    {
+                        "account": "Line 1",
+                        "user": "201",
+                        "server": "pbx.example.com",
+                        "registered": True,
+                    }
+                ]
+            """
+            ...
+
+        def get_call_status(self) -> Dict[str, Any]:
+            """
+            Returns what the phone is doing right now.
+
+            * state (string) - ``"idle"``, ``"ringing"``, ``"talking"``, or ``"held"``
+            * remote (string) - remote party, empty string when idle
+            * duration (int) - seconds in the current state; ``0`` when idle
+            """
+            ...

napalm_device_types/ping_sweep.py

@@ -0,0 +1,172 @@
+# -*- coding: utf-8 -*-
+"""Generic ICMP sweep on top of the NAPALM-standard ``ping()``.
+
+Sweeping a range of addresses is orchestration, not device mechanics: the
+only vendor-specific part is how a single ``ping`` is executed, and NAPALM
+already standardises that. So the loop, the reply parsing, the target cap
+and the progress reporting live here once, and a concrete driver only has
+to implement ``ping()`` to become a usable sweep source.
+
+A driver whose device offers a *faster* sweep mechanism (a batch API, a
+single shell command that pings many hosts in parallel, an ARP-assisted
+scan) overrides :meth:`PingSweepMixin.ping_sweep` and keeps the same return
+shape — see ``napalm-opnsense`` for an example.
+
+The generic implementation is deliberately **sequential**: a NAPALM
+connection is a single session (SSH channel, HTTP client) and is not safe to
+drive from several threads at once. Callers that need many addresses covered
+quickly should either use a driver with its own parallel override or cap the
+target list (see ``PING_SWEEP_MAX_TARGETS``).
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Callable, ClassVar, Dict, Iterable, List, Optional
+
+from napalm.base import NetworkDriver
+
+from napalm_device_types.models import PingSweepEntryDict, PingSweepResultDict
+
+
+def driver_supports_ping(driver_cls: type) -> bool:
+    """Whether *driver_cls* can actually execute ``ping()``.
+
+    True when the class provides its own ``ping`` implementation instead of
+    inheriting NAPALM's ``NotImplementedError`` stub. A driver that inherits
+    a working ``ping`` but cannot use it (unsupported firmware, disabled
+    service) opts out by setting ``SUPPORTS_PING = False``.
+    """
+    if getattr(driver_cls, "SUPPORTS_PING", None) is False:
+        return False
+    ping_impl = getattr(driver_cls, "ping", None)
+    if ping_impl is None:
+        return False
+    return ping_impl is not getattr(NetworkDriver, "ping", None)
+
+
+class PingSweepMixin:
+    """Adds :meth:`ping_sweep` to any driver that implements ``ping()``.
+
+    Mixed into :class:`~napalm_device_types.base.DeviceTypeDriver`, so every
+    device-type driver inherits it; drivers without a ``ping()`` of their own
+    simply report ``supports_ping() is False`` and raise from ``ping_sweep``.
+    """
+
+    #: Upper bound on destinations probed in one sweep. The sequential
+    #: default implementation costs roughly ``timeout`` seconds per silent
+    #: host, so an uncapped /24 would keep a device session busy for minutes.
+    #: Drivers with a parallel mechanism raise this.
+    PING_SWEEP_MAX_TARGETS: ClassVar[int] = 256
+
+    #: ``False`` opts a driver out of ping sweeps even though it implements
+    #: ``ping()``. ``None`` (the default) means "decide by introspection".
+    SUPPORTS_PING: ClassVar[Optional[bool]] = None
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def ping(
+            self,
+            destination: str,
+            source: str = "",
+            ttl: int = 255,
+            timeout: int = 2,
+            size: int = 100,
+            count: int = 5,
+            vrf: str = "",
+        ) -> Dict[str, Any]: ...
+
+    @classmethod
+    def supports_ping(cls) -> bool:
+        """Whether this driver class can be used as a ping-sweep source."""
+        return driver_supports_ping(cls)
+
+    def ping_sweep(
+        self,
+        destinations: Iterable[str],
+        *,
+        count: int = 1,
+        timeout: int = 1,
+        max_targets: Optional[int] = None,
+        on_progress: Optional[Callable[[int, int], None]] = None,
+        should_stop: Optional[Callable[[], bool]] = None,
+    ) -> PingSweepResultDict:
+        """Ping every address in *destinations* and report who answered.
+
+        :param destinations: IP addresses / hostnames to probe, in order.
+        :param count: probes per destination — 1 is enough for liveness.
+        :param timeout: seconds to wait for a reply per destination.
+        :param max_targets: cap for this call; defaults to
+            ``PING_SWEEP_MAX_TARGETS``. Excess destinations are dropped and
+            ``truncated`` is set in the result.
+        :param on_progress: called as ``(done, total)`` after each probe.
+        :param should_stop: polled before each probe; returning True ends the
+            sweep early (cancelled job, shutting-down worker).
+        :raises NotImplementedError: if the driver has no ``ping()``.
+        """
+        if not self.supports_ping():
+            raise NotImplementedError(
+                f"{type(self).__name__} does not implement ping(); cannot run a ping sweep"
+            )
+
+        limit = self.PING_SWEEP_MAX_TARGETS if max_targets is None else max_targets
+        targets = list(destinations)
+        truncated = len(targets) > limit
+        if truncated:
+            targets = targets[:limit]
+
+        total = len(targets)
+        entries: List[PingSweepEntryDict] = []
+        for done, destination in enumerate(targets, start=1):
+            if should_stop is not None and should_stop():
+                break
+            entries.append(self._ping_once(destination, count=count, timeout=timeout))
+            if on_progress is not None:
+                on_progress(done, total)
+
+        return {
+            "entries": entries,
+            "scanned": len(entries),
+            "alive_count": sum(1 for entry in entries if entry["alive"]),
+            "truncated": truncated,
+        }
+
+    # ── internals ────────────────────────────────────────────────────────────
+
+    def _ping_once(self, destination: str, *, count: int, timeout: int) -> PingSweepEntryDict:
+        """One probe, never raising — a dead session must not abort the sweep."""
+        try:
+            reply = self.ping(destination, count=count, timeout=timeout)
+        except Exception as exc:  # noqa: BLE001 - any driver error is just "no answer"
+            return {"ip": destination, "alive": False, "rtt_ms": None, "error": str(exc)}
+        return self._parse_ping_reply(destination, reply)
+
+    @staticmethod
+    def _parse_ping_reply(destination: str, reply: Any) -> PingSweepEntryDict:
+        """Map a NAPALM ``ping()`` reply onto a sweep entry."""
+        if not isinstance(reply, dict) or "success" not in reply:
+            error = "malformed ping reply"
+            if isinstance(reply, dict) and reply.get("error"):
+                error = str(reply["error"])
+            return {"ip": destination, "alive": False, "rtt_ms": None, "error": error}
+
+        success = reply.get("success") or {}
+        probes_sent = _as_int(success.get("probes_sent"))
+        packet_loss = _as_int(success.get("packet_loss"), default=probes_sent)
+        alive = bool(success.get("results")) or probes_sent > packet_loss
+        if not alive:
+            return {"ip": destination, "alive": False, "rtt_ms": None}
+        return {"ip": destination, "alive": True, "rtt_ms": _as_float(success.get("rtt_avg"))}
+
+
+def _as_int(value: Any, default: int = 0) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def _as_float(value: Any) -> Optional[float]:
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None

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::
 
@@ -18,24 +19,21 @@ Usage::
             ...
 """
 
-from typing import Dict, List
+from typing import Dict, List, TYPE_CHECKING
 from napalm_device_types.base import DeviceTypeDriver
-from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
+from napalm_device_types.nat_vpn import NatVpnMixin
+from napalm_device_types.health_metrics import HealthMetricsMixin
+from napalm_device_types.dhcp import DhcpServerMixin
 from napalm_device_types.models import (
-    HealthMetricsDict,
     HostDict,
-    NATTranslationDict,
-    PortForwardDict,
     RadioStatusDict,
     SSIDDict,
-    VPNTunnelDict,
     WANStatusDict,
     WirelessClientDict,
 )
 
 
-class ResidentialGatewayDriver(DeviceTypeDriver):
-    TYPE_LABEL: str = "Gateway"
+class ResidentialGatewayDriver(NatVpnMixin, HealthMetricsMixin, DhcpServerMixin, DeviceTypeDriver):
     """
     Abstract intermediate driver for residential gateways (router + firewall + AP).
 
@@ -44,154 +42,103 @@ class ResidentialGatewayDriver(DeviceTypeDriver):
     drivers must implement.
     """
 
-    _SNMP_SKIP_IF = IF_SKIP_DEFAULT
-    _SNMP_TX_ERR_IS_DROP: bool = False
+    #: Stable key netOrk exposes as ``device_class``. The order in which a
+    #: driver lists its role bases is the ranking; see
+    #: :func:`napalm_device_types.roles.primary_role_of`.
+    ROLE: str = "residential_gateway"
+    TYPE_LABEL: str = "Gateway"
 
-    @classmethod
-    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
-        return await collect_ucd_metrics(
-            snmp_get, snmp_walk,
-            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
-            if_skip=cls._SNMP_SKIP_IF,
-        )
 
-    def get_wan_status(self) -> WANStatusDict:
-        """
-        Returns the status of the device's internet (WAN) uplink.
 
-        Contains:
+    if TYPE_CHECKING:
 
-        * connection_type (string) - e.g. ``"DSL"``, ``"Cable"``, ``"PPPoE"``, ``"DHCP"``
-        * is_connected (bool) - whether the WAN connection is currently established
-        * external_ip (string) - the public IPv4 address assigned to the WAN interface
-        * uptime (int) - seconds since the WAN connection was last (re-)established
-        * bytes_sent (int) - total bytes transmitted on the WAN interface
-        * bytes_received (int) - total bytes received on the WAN interface
-        * max_bitrate_up (int) - upstream sync rate in kbit/s
-        * max_bitrate_down (int) - downstream sync rate in kbit/s
-        * external_ipv6 (string, optional) - the public IPv6 address, if any
-        * link_status (string, optional) - physical line state, e.g. ``"Up"`` / ``"Down"``
+        def get_wan_status(self) -> WANStatusDict:
+            """
+            Returns the status of the device's internet (WAN) uplink.
 
-        Example::
+            Contains:
 
-            {
-                "connection_type": "DSL",
-                "is_connected": True,
-                "external_ip": "203.0.113.7",
-                "uptime": 345600,
-                "bytes_sent": 1234567890,
-                "bytes_received": 9876543210,
-                "max_bitrate_up": 40000,
-                "max_bitrate_down": 250000,
-                "link_status": "Up",
-            }
-        """
-        raise NotImplementedError
+            * connection_type (string) - e.g. ``"DSL"``, ``"Cable"``, ``"PPPoE"``, ``"DHCP"``
+            * is_connected (bool) - whether the WAN connection is currently established
+            * external_ip (string) - the public IPv4 address assigned to the WAN interface
+            * uptime (int) - seconds since the WAN connection was last (re-)established
+            * bytes_sent (int) - total bytes transmitted on the WAN interface
+            * bytes_received (int) - total bytes received on the WAN interface
+            * max_bitrate_up (int) - upstream sync rate in kbit/s
+            * max_bitrate_down (int) - downstream sync rate in kbit/s
+            * external_ipv6 (string, optional) - the public IPv6 address, if any
+            * link_status (string, optional) - physical line state, e.g. ``"Up"`` / ``"Down"``
 
-    def get_port_forwards(self) -> List[PortForwardDict]:
-        """
-        Returns the configured port forwarding (port mapping) rules.
+            Example::
 
-        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,
+                    "connection_type": "DSL",
+                    "is_connected": True,
+                    "external_ip": "203.0.113.7",
+                    "uptime": 345600,
+                    "bytes_sent": 1234567890,
+                    "bytes_received": 9876543210,
+                    "max_bitrate_up": 40000,
+                    "max_bitrate_down": 250000,
+                    "link_status": "Up",
                 }
-            ]
-        """
-        raise NotImplementedError
+            """
+            ...
 
-    def get_hosts(self) -> List[HostDict]:
-        """
-        Returns the list of hosts known to the gateway (LAN clients).
+        def get_hosts(self) -> List[HostDict]:
+            """
+            Returns the list of hosts known to the gateway (LAN clients).
 
-        Each entry contains:
+            Each entry contains:
 
-        * mac (string) - the host's MAC address
-        * ip (string) - the host's current IP address
-        * hostname (string) - the host's reported hostname (empty if unknown)
-        * interface_type (string) - how the host is connected, e.g. ``"LAN"``, ``"WLAN"``
-        * is_active (bool) - whether the host is currently online
-        * lease_time_remaining (int, optional) - remaining DHCP lease time in seconds
+            * mac (string) - the host's MAC address
+            * ip (string) - the host's current IP address
+            * hostname (string) - the host's reported hostname (empty if unknown)
+            * interface_type (string) - how the host is connected, e.g. ``"LAN"``, ``"WLAN"``
+            * is_active (bool) - whether the host is currently online
+            * lease_time_remaining (int, optional) - remaining DHCP lease time in seconds
 
-        Example::
+            Example::
 
-            [
-                {
-                    "mac": "AA:BB:CC:DD:EE:FF",
-                    "ip": "192.168.1.42",
-                    "hostname": "laptop",
-                    "interface_type": "WLAN",
-                    "is_active": True,
-                    "lease_time_remaining": 3600,
-                }
-            ]
-        """
-        raise NotImplementedError
+                [
+                    {
+                        "mac": "AA:BB:CC:DD:EE:FF",
+                        "ip": "192.168.1.42",
+                        "hostname": "laptop",
+                        "interface_type": "WLAN",
+                        "is_active": True,
+                        "lease_time_remaining": 3600,
+                    }
+                ]
+            """
+            ...
 
-    def get_nat_translations(self) -> List[NATTranslationDict]:
-        """
-        Returns a list of active NAT translation entries.
 
-        See :meth:`napalm_device_types.firewall.FirewallDriver.get_nat_translations`
-        for the entry format. Residential gateways typically derive this from
-        the active port-forwarding/NAT-PT table rather than a live connection
-        tracker; drivers that cannot provide this should return an empty list.
-        """
-        raise NotImplementedError
 
-    def get_vpn_tunnels(self) -> Dict[str, VPNTunnelDict]:
-        """
-        Returns the status of VPN tunnels (e.g. WireGuard road-warrior
-        access, IPsec site-to-site).
+        def get_wireless_clients(self) -> List[WirelessClientDict]:
+            """
+            Returns the list of wireless clients currently associated with the
+            device's built-in access point(s).
 
-        See :meth:`napalm_device_types.firewall.FirewallDriver.get_vpn_tunnels`
-        for the entry format. Drivers that cannot provide this should return
-        an empty dict.
-        """
-        raise NotImplementedError
+            See :meth:`napalm_device_types.access_point.AccessPointDriver.get_wireless_clients`
+            for the entry format.
+            """
+            ...
 
-    def get_wireless_clients(self) -> List[WirelessClientDict]:
-        """
-        Returns the list of wireless clients currently associated with the
-        device's built-in access point(s).
+        def get_ssids(self) -> Dict[str, SSIDDict]:
+            """
+            Returns the configured wireless networks (SSIDs).
 
-        See :meth:`napalm_device_types.access_point.AccessPointDriver.get_wireless_clients`
-        for the entry format.
-        """
-        raise NotImplementedError
+            See :meth:`napalm_device_types.access_point.AccessPointDriver.get_ssids`
+            for the entry format.
+            """
+            ...
 
-    def get_ssids(self) -> Dict[str, SSIDDict]:
-        """
-        Returns the configured wireless networks (SSIDs).
+        def get_radio_status(self) -> Dict[str, RadioStatusDict]:
+            """
+            Returns the status of the device's wireless radios.
 
-        See :meth:`napalm_device_types.access_point.AccessPointDriver.get_ssids`
-        for the entry format.
-        """
-        raise NotImplementedError
-
-    def get_radio_status(self) -> Dict[str, RadioStatusDict]:
-        """
-        Returns the status of the device's wireless radios.
-
-        See :meth:`napalm_device_types.access_point.AccessPointDriver.get_radio_status`
-        for the entry format.
-        """
-        raise NotImplementedError
+            See :meth:`napalm_device_types.access_point.AccessPointDriver.get_radio_status`
+            for the entry format.
+            """
+            ...

napalm_device_types/roles.py

@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+"""Which roles a driver class fills, and which of them is primary.
+
+A *role* is what a device is: a firewall, a switch, a NAS. Real devices are
+often several at once -- a QNAP is storage, hypervisor and Linux host -- so a
+driver declares every role it fills by inheriting the matching base, and the
+**order of those bases is the ranking**::
+
+    class QnapQtsDriver(StorageDriver, HypervisorDriver, LinuxDriver):
+        ...        # primary_role_of(...) == "storage"
+
+That replaces the older arrangement, where an ``issubclass`` chain in netOrk
+picked a single winner by a fixed precedence and a ``DEVICE_CLASS`` attribute
+existed to overrule it. The author already states the ranking in the class
+definition; nothing needs to restate it.
+
+Whether a driver can perform a *specific operation* is a separate question with
+a separate answer: ``hasattr``. Role bases declare their methods under
+``if TYPE_CHECKING`` and implement nothing, so a method exists at runtime only
+when a concrete driver provided it.
+"""
+
+from __future__ import annotations
+
+from typing import Any, List, Optional, Type
+
+
+def _is_role_base(cls: type) -> bool:
+    """True for a class that introduces a role, not one that merely inherits one.
+
+    Keyed on ``ROLE`` appearing in the class's own ``__dict__``: a concrete
+    driver inherits the attribute but does not define it, and must not be
+    mistaken for a role base of its own.
+    """
+    role = vars(cls).get("ROLE")
+    return isinstance(role, str) and bool(role)
+
+
+def roles_of(driver_cls: type) -> List[Type[Any]]:
+    """Every role base in ``driver_cls``'s MRO, most significant first.
+
+    MRO order is inheritance order, so the list reflects exactly what the driver
+    author wrote in the class definition.
+    """
+    return [cls for cls in driver_cls.__mro__ if _is_role_base(cls)]
+
+
+def role_keys_of(driver_cls: type) -> List[str]:
+    """The stable string keys of :func:`roles_of`, e.g. ``["storage", "linux"]``."""
+    return [vars(cls)["ROLE"] for cls in roles_of(driver_cls)]
+
+
+def primary_role_of(driver_cls: type) -> Optional[str]:
+    """The driver's headline role, or ``None`` when it declares no role at all.
+
+    This is the value netOrk exposes as ``device_class``.
+    """
+    keys = role_keys_of(driver_cls)
+    return keys[0] if keys else None

napalm_device_types/services.py

@@ -0,0 +1,55 @@
+# -*- coding: utf-8 -*-
+"""Init-system services: listing them and starting/stopping them.
+
+Deliberately *not* the same thing as a NAS's exported shares, which live on
+``StorageServiceCapability`` as ``get_storage_services``. Those two used to
+share the name ``get_services`` with incompatible return types, which is why
+a QNAP -- storage and OS at once -- could not satisfy both.
+
+Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
+Nothing exists at runtime until a concrete driver implements it, so mixing
+this class in can never shadow a working implementation from a sibling base.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, TYPE_CHECKING
+
+from napalm_device_types.models import ServiceDict
+
+
+class ServiceControlMixin:
+    if TYPE_CHECKING:
+
+        def get_services(self) -> List[ServiceDict]:
+            """
+            Returns the list of system services known to the device's init system.
+
+            Each entry contains:
+
+            * name (string)    - service name as registered with the init system
+            * running (bool)   - ``True`` if the service process is currently running
+            * enabled (bool)   - ``True`` if the service starts automatically at boot
+            * pid (int)        - process ID of the main service process; 0 if not running
+
+            Example::
+
+                [
+                    {"name": "lldpd",   "running": True,  "enabled": True,  "pid": 2341},
+                    {"name": "sshd",    "running": True,  "enabled": True,  "pid": 1198},
+                    {"name": "cron",    "running": False, "enabled": False, "pid": 0},
+                ]
+            """
+            ...
+
+        def manage_service(self, name: str, action: str) -> Dict[str, Any]:
+            """
+            Execute a lifecycle action on a named service.
+
+            :param name: Service name as returned by :meth:`get_services`.
+            :param action: One of ``start``, ``stop``, ``restart``, ``enable``, ``disable``.
+            :returns: ``{"success": bool, "output": str}``
+            :raises ValueError: If ``name`` or ``action`` is invalid.
+            :raises NotImplementedError: If the driver does not support service management.
+            """
+            ...

napalm_device_types/switch.py

@@ -10,13 +10,13 @@ Usage::
             ...
 """
 
-from typing import Dict, List
+from typing import Any, Awaitable, Callable, Dict, List, TYPE_CHECKING
 from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types.mac_acl import MacAclMixin
 from napalm_device_types.models import (
     Dot1XPortDict,
     HealthMetricsDict,
     InterfaceConfigDict,
-    MACACLDict,
     PoESummaryDict,
     PortChannelDict,
     SpanningTreeDict,
@@ -24,8 +24,7 @@ from napalm_device_types.models import (
 )
 
 
-class SwitchDriver(DeviceTypeDriver):
-    TYPE_LABEL: str = "Switch"
+class SwitchDriver(MacAclMixin, DeviceTypeDriver):
     """
     Abstract intermediate driver for Ethernet switches.
 
@@ -34,403 +33,378 @@ class SwitchDriver(DeviceTypeDriver):
     operations that concrete drivers must implement.
     """
 
-    @classmethod
-    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
-        """Collect SNMP health metrics for this switch type.
+    #: Stable key netOrk exposes as ``device_class``. The order in which a
+    #: driver lists its role bases is the ranking; see
+    #: :func:`napalm_device_types.roles.primary_role_of`.
+    ROLE: str = "switch"
+    TYPE_LABEL: str = "Switch"
 
-        Switch vendors use proprietary OIDs — each concrete driver must
-        override this classmethod.
-        """
-        raise NotImplementedError
+    if TYPE_CHECKING:
 
-    def get_spanning_tree(self) -> Dict[str, SpanningTreeDict]:
-        """
-        Returns spanning tree status for each STP instance.
+        @classmethod
+        async def get_health_metrics(
+            cls,
+            snmp_get: Callable[..., Awaitable[Any]],
+            snmp_walk: Callable[..., Awaitable[Dict[str, Any]]],
+        ) -> HealthMetricsDict:
+            """Collect SNMP health metrics for this switch type.
 
-        Keys are STP instance identifiers (e.g. VLAN IDs for PVST,
-        ``"MST0"`` for MSTP, or ``"0"`` for a single instance).
-        Each value contains:
+            Switch vendors use proprietary OIDs — each concrete driver must
+            override this classmethod.
+            """
+            ...
 
-        * mode (string) - STP variant: ``"STP"``, ``"RSTP"``, ``"MSTP"``, ``"PVST"``
-        * root_bridge (bool) - whether this device is the root bridge
-        * root_id (string) - root bridge MAC address
-        * root_priority (int) - root bridge priority
-        * bridge_id (string) - this bridge's MAC address
-        * bridge_priority (int) - this bridge's priority
-        * interfaces (dict) - per-interface STP state:
+        def get_spanning_tree(self) -> Dict[str, SpanningTreeDict]:
+            """
+            Returns spanning tree status for each STP instance.
 
-            * role (string) - ``"root"``, ``"designated"``, ``"alternate"``, ``"backup"``
-            * state (string) - ``"forwarding"``, ``"blocking"``, ``"learning"``, ``"listening"``
-            * cost (int) - port path cost
-            * port_priority (int) - port priority
+            Keys are STP instance identifiers (e.g. VLAN IDs for PVST,
+            ``"MST0"`` for MSTP, or ``"0"`` for a single instance).
+            Each value contains:
 
-        Example::
+            * mode (string) - STP variant: ``"STP"``, ``"RSTP"``, ``"MSTP"``, ``"PVST"``
+            * root_bridge (bool) - whether this device is the root bridge
+            * root_id (string) - root bridge MAC address
+            * root_priority (int) - root bridge priority
+            * bridge_id (string) - this bridge's MAC address
+            * bridge_priority (int) - this bridge's priority
+            * interfaces (dict) - per-interface STP state:
 
-            {
-                "1": {
-                    "mode": "RSTP",
-                    "root_bridge": False,
-                    "root_id": "00:11:22:33:44:55",
-                    "root_priority": 4096,
-                    "bridge_id": "AA:BB:CC:DD:EE:FF",
-                    "bridge_priority": 32768,
-                    "interfaces": {
-                        "GigabitEthernet0/1": {
-                            "role": "root",
-                            "state": "forwarding",
-                            "cost": 4,
-                            "port_priority": 128,
+                * role (string) - ``"root"``, ``"designated"``, ``"alternate"``, ``"backup"``
+                * state (string) - ``"forwarding"``, ``"blocking"``, ``"learning"``, ``"listening"``
+                * cost (int) - port path cost
+                * port_priority (int) - port priority
+
+            Example::
+
+                {
+                    "1": {
+                        "mode": "RSTP",
+                        "root_bridge": False,
+                        "root_id": "00:11:22:33:44:55",
+                        "root_priority": 4096,
+                        "bridge_id": "AA:BB:CC:DD:EE:FF",
+                        "bridge_priority": 32768,
+                        "interfaces": {
+                            "GigabitEthernet0/1": {
+                                "role": "root",
+                                "state": "forwarding",
+                                "cost": 4,
+                                "port_priority": 128,
+                            },
+                            "GigabitEthernet0/2": {
+                                "role": "designated",
+                                "state": "forwarding",
+                                "cost": 4,
+                                "port_priority": 128,
+                            },
                         },
-                        "GigabitEthernet0/2": {
-                            "role": "designated",
-                            "state": "forwarding",
-                            "cost": 4,
-                            "port_priority": 128,
-                        },
-                    },
+                    }
                 }
-            }
-        """
-        raise NotImplementedError
+            """
+            ...
 
-    def get_port_channels(self) -> Dict[str, PortChannelDict]:
-        """
-        Returns port-channel (LAG) configuration and status.
+        def get_port_channels(self) -> Dict[str, PortChannelDict]:
+            """
+            Returns port-channel (LAG) configuration and status.
 
-        Keys are port-channel interface names (e.g. ``"Port-Channel1"``).
-        Each value contains:
+            Keys are port-channel interface names (e.g. ``"Port-Channel1"``).
+            Each value contains:
 
-        * members (list of strings) - names of member interfaces
-        * protocol (string) - aggregation protocol: ``"LACP"``, ``"PAgP"``, ``"static"``
-        * min_links (int) - minimum number of active members required
-        * is_up (bool) - whether the LAG is operationally up
+            * members (list of strings) - names of member interfaces
+            * protocol (string) - aggregation protocol: ``"LACP"``, ``"PAgP"``, ``"static"``
+            * min_links (int) - minimum number of active members required
+            * is_up (bool) - whether the LAG is operationally up
 
-        Example::
+            Example::
 
-            {
-                "Port-Channel1": {
-                    "members": ["GigabitEthernet0/1", "GigabitEthernet0/2"],
-                    "protocol": "LACP",
-                    "min_links": 1,
-                    "is_up": True,
+                {
+                    "Port-Channel1": {
+                        "members": ["GigabitEthernet0/1", "GigabitEthernet0/2"],
+                        "protocol": "LACP",
+                        "min_links": 1,
+                        "is_up": True,
+                    }
                 }
-            }
-        """
-        raise NotImplementedError
+            """
+            ...
 
-    def get_mac_acl(self) -> Dict[str, MACACLDict]:
-        """
-        Returns the MAC-address-based access control lists configured per port.
 
-        Keys are interface names.  Each value contains:
+        def get_dot1x_ports(self) -> Dict[str, Dot1XPortDict]:
+            """
+            Returns the 802.1X / NAC configuration per switch port.
 
-        * name (string) - interface name (repeated for convenience)
-        * policy (string) - ACL mode:
+            Keys are interface names.  Each value contains:
 
-          * ``"allow"`` – whitelist: only listed MACs may use this port
-          * ``"deny"``  – blacklist: listed MACs are blocked
-          * ``"disabled"`` – no MAC filtering active
+            * enabled (bool) - whether 802.1X is active on this port
+            * port_control (string) - authentication mode:
 
-        * entries (list) - ACL entries, each with:
+              * ``"auto"`` – port authenticates normally
+              * ``"force-authorized"`` – port always passes traffic (bypass)
+              * ``"force-unauthorized"`` – port always blocks traffic
 
-          * mac (string) - MAC address (normalised, colon-separated)
-          * action (string) - ``"allow"`` or ``"deny"``
-          * description (string) - optional human-readable label
+            * host_mode (string) - how many identities are authenticated per port:
 
-        Example::
+              * ``"single-host"`` – one device, then port is locked
+              * ``"multi-host"`` – first auth unlocks port for all devices
+              * ``"multi-domain"`` – one data + one voice device (IP phone scenario)
+              * ``"multi-auth"`` – each device authenticates individually
 
-            {
-                "GigabitEthernet0/1": {
-                    "name": "GigabitEthernet0/1",
-                    "policy": "allow",
-                    "entries": [
-                        {"mac": "AA:BB:CC:DD:EE:01", "action": "allow", "description": "printer"},
-                    ],
-                },
-                "GigabitEthernet0/2": {
-                    "name": "GigabitEthernet0/2",
-                    "policy": "disabled",
-                    "entries": [],
-                },
-            }
-        """
-        raise NotImplementedError
+            * auth_server (dict) - RADIUS authentication server (host, port, timeout, retries)
+            * acct_server (dict or None) - RADIUS accounting server, ``None`` if unused
+            * reauthentication (bool) - whether periodic re-authentication is enabled
+            * reauth_interval (int) - re-authentication interval in seconds (0 = disabled)
+            * guest_vlan (int) - VLAN ID for unauthenticated clients (0 = disabled)
+            * auth_fail_vlan (int) - VLAN ID for clients that fail authentication (0 = disabled)
 
-    def get_dot1x_config(self) -> Dict[str, Dot1XPortDict]:
-        """
-        Returns the 802.1X / NAC configuration per switch port.
+            Note: RADIUS shared secrets are intentionally omitted.
 
-        Keys are interface names.  Each value contains:
+            Example::
 
-        * enabled (bool) - whether 802.1X is active on this port
-        * port_control (string) - authentication mode:
-
-          * ``"auto"`` – port authenticates normally
-          * ``"force-authorized"`` – port always passes traffic (bypass)
-          * ``"force-unauthorized"`` – port always blocks traffic
-
-        * host_mode (string) - how many identities are authenticated per port:
-
-          * ``"single-host"`` – one device, then port is locked
-          * ``"multi-host"`` – first auth unlocks port for all devices
-          * ``"multi-domain"`` – one data + one voice device (IP phone scenario)
-          * ``"multi-auth"`` – each device authenticates individually
-
-        * auth_server (dict) - RADIUS authentication server (host, port, timeout, retries)
-        * acct_server (dict or None) - RADIUS accounting server, ``None`` if unused
-        * reauthentication (bool) - whether periodic re-authentication is enabled
-        * reauth_interval (int) - re-authentication interval in seconds (0 = disabled)
-        * guest_vlan (int) - VLAN ID for unauthenticated clients (0 = disabled)
-        * auth_fail_vlan (int) - VLAN ID for clients that fail authentication (0 = disabled)
-
-        Note: RADIUS shared secrets are intentionally omitted.
-
-        Example::
-
-            {
-                "GigabitEthernet0/1": {
-                    "enabled": True,
-                    "port_control": "auto",
-                    "host_mode": "multi-domain",
-                    "auth_server": {
-                        "host": "radius.corp.example",
-                        "port": 1812,
-                        "timeout": 5,
-                        "retries": 3,
-                    },
-                    "acct_server": None,
-                    "reauthentication": True,
-                    "reauth_interval": 3600,
-                    "guest_vlan": 99,
-                    "auth_fail_vlan": 999,
-                },
-            }
-        """
-        raise NotImplementedError
-
-    def get_poe_status(self) -> PoESummaryDict:
-        """
-        Returns the PoE status of the switch as a whole and per port.
-
-        The returned dictionary contains:
-
-        * total_power_budget (float) - total PoE power available in watts
-        * total_power_draw (float) - total PoE power currently consumed in watts
-        * ports (dict) - per-interface PoE state, keyed by interface name:
-
-          * enabled (bool) - whether PoE is configured on this port
-          * status (string) - operational state:
-
-            * ``"delivering"`` – power is being delivered to a PD
-            * ``"searching"``  – port is looking for a powered device
-            * ``"fault"``      – an error condition was detected
-            * ``"disabled"``   – PoE is administratively off
-            * ``"denied"``     – PD detected but power budget exceeded
-
-          * poe_class (string) - IEEE 802.3 class: ``"Class 0"`` … ``"Class 8"``
-            (``"unknown"`` if not yet negotiated)
-          * power_draw (float) - current power consumption in watts
-          * power_budget (float) - per-port power limit in watts
-          * voltage (float) - measured port voltage in volts
-          * current (float) - measured port current in milliamps
-
-        Example::
-
-            {
-                "total_power_budget": 740.0,
-                "total_power_draw": 43.2,
-                "ports": {
+                {
                     "GigabitEthernet0/1": {
                         "enabled": True,
-                        "status": "delivering",
-                        "poe_class": "Class 3",
-                        "power_draw": 12.4,
-                        "power_budget": 30.0,
-                        "voltage": 53.5,
-                        "current": 231.0,
+                        "port_control": "auto",
+                        "host_mode": "multi-domain",
+                        "auth_server": {
+                            "host": "radius.corp.example",
+                            "port": 1812,
+                            "timeout": 5,
+                            "retries": 3,
+                        },
+                        "acct_server": None,
+                        "reauthentication": True,
+                        "reauth_interval": 3600,
+                        "guest_vlan": 99,
+                        "auth_fail_vlan": 999,
                     },
-                    "GigabitEthernet0/2": {
+                }
+            """
+            ...
+
+        def get_poe_status(self) -> PoESummaryDict:
+            """
+            Returns the PoE status of the switch as a whole and per port.
+
+            The returned dictionary contains:
+
+            * total_power_budget (float) - total PoE power available in watts
+            * total_power_draw (float) - total PoE power currently consumed in watts
+            * ports (dict) - per-interface PoE state, keyed by interface name:
+
+              * enabled (bool) - whether PoE is configured on this port
+              * status (string) - operational state:
+
+                * ``"delivering"`` – power is being delivered to a PD
+                * ``"searching"``  – port is looking for a powered device
+                * ``"fault"``      – an error condition was detected
+                * ``"disabled"``   – PoE is administratively off
+                * ``"denied"``     – PD detected but power budget exceeded
+
+              * poe_class (string) - IEEE 802.3 class: ``"Class 0"`` … ``"Class 8"``
+                (``"unknown"`` if not yet negotiated)
+              * power_draw (float) - current power consumption in watts
+              * power_budget (float) - per-port power limit in watts
+              * voltage (float) - measured port voltage in volts
+              * current (float) - measured port current in milliamps
+
+            Example::
+
+                {
+                    "total_power_budget": 740.0,
+                    "total_power_draw": 43.2,
+                    "ports": {
+                        "GigabitEthernet0/1": {
+                            "enabled": True,
+                            "status": "delivering",
+                            "poe_class": "Class 3",
+                            "power_draw": 12.4,
+                            "power_budget": 30.0,
+                            "voltage": 53.5,
+                            "current": 231.0,
+                        },
+                        "GigabitEthernet0/2": {
+                            "enabled": True,
+                            "status": "searching",
+                            "poe_class": "unknown",
+                            "power_draw": 0.0,
+                            "power_budget": 30.0,
+                            "voltage": 0.0,
+                            "current": 0.0,
+                        },
+                        "GigabitEthernet0/3": {
+                            "enabled": False,
+                            "status": "disabled",
+                            "poe_class": "unknown",
+                            "power_draw": 0.0,
+                            "power_budget": 0.0,
+                            "voltage": 0.0,
+                            "current": 0.0,
+                        },
+                    },
+                }
+            """
+            ...
+
+        def set_vlan(self, vlan_id: int, config: VlanConfigDict) -> None:
+            """
+            Creates or updates a VLAN on the switch.
+
+            If the VLAN does not yet exist it is created first.  Only the keys
+            present in *config* are applied; omitted keys leave the existing VLAN
+            configuration untouched.
+
+            :param vlan_id: VLAN ID (1–4094).
+            :param config: A (partial) :class:`~napalm_device_types.models.VlanConfigDict`
+                containing the fields to set.  Supported keys:
+
+                * ``name`` (str) – human-readable VLAN name
+                * ``active`` (bool) – ``True`` = active, ``False`` = suspended
+                * ``interfaces`` (list of str) – access-port names to assign to
+                  this VLAN (replaces the current membership list)
+
+            :raises NotImplementedError: If the driver does not implement this method.
+            :raises ValueError: If *vlan_id* is out of range or a field value is invalid.
+
+            Example – create VLAN 10 with a name::
+
+                driver.set_vlan(10, {"name": "Workstations", "active": True})
+
+            Example – assign ports to an existing VLAN::
+
+                driver.set_vlan(
+                    10,
+                    {"interfaces": ["GigabitEthernet0/1", "GigabitEthernet0/2"]},
+                )
+            """
+            ...
+
+        def delete_vlan(self, vlan_id: int) -> None:
+            """
+            Removes a VLAN from the switch.
+
+            All ports that were assigned to this VLAN as their access VLAN are
+            moved to the default VLAN (1) by the driver before deletion.  Trunk
+            ports that carry this VLAN will have it removed from their allowed
+            VLAN list.
+
+            :param vlan_id: VLAN ID (1–4094) to delete.
+            :raises NotImplementedError: If the driver does not implement this method.
+            :raises ValueError: If *vlan_id* is out of range or the VLAN does not
+                exist on the device.
+
+            Example::
+
+                driver.delete_vlan(10)
+            """
+            ...
+
+        def set_interface(self, interface: str, config: InterfaceConfigDict) -> None:
+            """
+            Applies configuration to a single switch interface.
+
+            Only the keys present in *config* are changed; omitted keys leave the
+            current device configuration untouched.
+
+            :param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
+            :param config: A (partial) :class:`~napalm_device_types.models.InterfaceConfigDict`
+                containing the fields to update.  Supported keys:
+
+                * ``description`` (str) – human-readable port label
+                * ``enabled`` (bool) – administrative state
+                * ``speed`` (int) – link speed in Mbps; ``0`` = auto-negotiate
+                * ``duplex`` (str) – ``"full"``, ``"half"``, or ``"auto"``
+                * ``mtu`` (int) – maximum transmission unit in bytes
+                * ``mode`` (str) – ``"access"``, ``"trunk"``, or ``"routed"``
+                * ``access_vlan`` (int) – untagged VLAN; relevant when mode is ``"access"``
+                * ``voice_vlan`` (int) – voice VLAN ID (``0`` = disabled)
+                * ``trunk_vlans`` (list of int) – tagged VLANs; empty = allow all
+                * ``native_vlan`` (int) – native VLAN on trunk ports
+
+            :raises NotImplementedError: If the driver does not implement this method.
+            :raises ValueError: If *interface* does not exist or an invalid value is
+                supplied for a configuration field.
+
+            Example – convert port to access VLAN 10 and add a description::
+
+                driver.set_interface(
+                    "GigabitEthernet0/1",
+                    {
+                        "description": "Workstation port",
                         "enabled": True,
-                        "status": "searching",
-                        "poe_class": "unknown",
-                        "power_draw": 0.0,
-                        "power_budget": 30.0,
-                        "voltage": 0.0,
-                        "current": 0.0,
+                        "mode": "access",
+                        "access_vlan": 10,
                     },
-                    "GigabitEthernet0/3": {
-                        "enabled": False,
-                        "status": "disabled",
-                        "poe_class": "unknown",
-                        "power_draw": 0.0,
-                        "power_budget": 0.0,
-                        "voltage": 0.0,
-                        "current": 0.0,
+                )
+
+            Example – configure a trunk port::
+
+                driver.set_interface(
+                    "GigabitEthernet0/2",
+                    {
+                        "mode": "trunk",
+                        "trunk_vlans": [10, 20, 30],
+                        "native_vlan": 1,
                     },
-                },
-            }
-        """
-        raise NotImplementedError
+                )
+            """
+            ...
 
-    def set_vlan(self, vlan_id: int, config: VlanConfigDict) -> None:
-        """
-        Creates or updates a VLAN on the switch.
+        def set_lag_members(self, lag_name: str, members: List[str]) -> None:
+            """
+            Sets the full member-port list of a LAG/trunk interface.
 
-        If the VLAN does not yet exist it is created first.  Only the keys
-        present in *config* are applied; omitted keys leave the existing VLAN
-        configuration untouched.
+            Diffs *members* against the LAG's current members (as reported by
+            ``get_interfaces()``'s ``lag_members`` field) and adds/removes ports
+            on the device to match. The LAG itself must already exist on the
+            device; this method only manages its membership.
 
-        :param vlan_id: VLAN ID (1–4094).
-        :param config: A (partial) :class:`~napalm_device_types.models.VlanConfigDict`
-            containing the fields to set.  Supported keys:
+            :param lag_name: Name of the LAG/trunk interface as returned by
+                ``get_interfaces()`` (e.g. ``"Trk1"``, ``"ch1"``, ``"Lag1"``).
+            :param members: Full desired list of member port names.
+            :raises NotImplementedError: If the driver does not implement this method.
+            :raises ValueError: If *lag_name* does not refer to a valid LAG.
 
-            * ``name`` (str) – human-readable VLAN name
-            * ``active`` (bool) – ``True`` = active, ``False`` = suspended
-            * ``interfaces`` (list of str) – access-port names to assign to
-              this VLAN (replaces the current membership list)
+            Example::
 
-        :raises NotImplementedError: If the driver does not implement this method.
-        :raises ValueError: If *vlan_id* is out of range or a field value is invalid.
+                driver.set_lag_members("Lag1", ["GigabitEthernet0/1", "GigabitEthernet0/2"])
+            """
+            ...
 
-        Example – create VLAN 10 with a name::
+        def set_poe_enabled(self, interface: str, enabled: bool) -> None:
+            """
+            Administratively enables or disables PoE on a single port.
 
-            driver.set_vlan(10, {"name": "Workstations", "active": True})
+            :param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
+            :param enabled: ``True`` to enable PoE, ``False`` to disable it.
+            :raises NotImplementedError: If the driver does not support PoE control.
+            :raises ValueError: If the interface does not exist or does not support PoE.
 
-        Example – assign ports to an existing VLAN::
+            Example::
 
-            driver.set_vlan(
-                10,
-                {"interfaces": ["GigabitEthernet0/1", "GigabitEthernet0/2"]},
-            )
-        """
-        raise NotImplementedError
+                driver.set_poe_enabled("GigabitEthernet0/1", False)  # cut power
+                driver.set_poe_enabled("GigabitEthernet0/1", True)   # restore
+            """
+            ...
 
-    def delete_vlan(self, vlan_id: int) -> None:
-        """
-        Removes a VLAN from the switch.
+        def power_cycle_port(self, interface: str, delay: int = 5) -> None:
+            """
+            Power-cycles the PoE port: cuts power, waits ``delay`` seconds, then
+            restores power.  Useful for rebooting a hung IP camera, AP, or IP phone
+            without physical access.
 
-        All ports that were assigned to this VLAN as their access VLAN are
-        moved to the default VLAN (1) by the driver before deletion.  Trunk
-        ports that carry this VLAN will have it removed from their allowed
-        VLAN list.
+            The method blocks until the full cycle (off → wait → on) is complete.
+            After it returns the port is back in the delivering/searching state.
 
-        :param vlan_id: VLAN ID (1–4094) to delete.
-        :raises NotImplementedError: If the driver does not implement this method.
-        :raises ValueError: If *vlan_id* is out of range or the VLAN does not
-            exist on the device.
+            :param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
+            :param delay: Seconds to keep the port powered off (default: 5).
+            :raises NotImplementedError: If the driver does not support PoE control.
+            :raises ValueError: If the interface does not exist, does not support PoE,
+                or PoE is administratively disabled on the port.
 
-        Example::
+            Example::
 
-            driver.delete_vlan(10)
-        """
-        raise NotImplementedError
-
-    def set_interface(self, interface: str, config: InterfaceConfigDict) -> None:
-        """
-        Applies configuration to a single switch interface.
-
-        Only the keys present in *config* are changed; omitted keys leave the
-        current device configuration untouched.
-
-        :param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
-        :param config: A (partial) :class:`~napalm_device_types.models.InterfaceConfigDict`
-            containing the fields to update.  Supported keys:
-
-            * ``description`` (str) – human-readable port label
-            * ``enabled`` (bool) – administrative state
-            * ``speed`` (int) – link speed in Mbps; ``0`` = auto-negotiate
-            * ``duplex`` (str) – ``"full"``, ``"half"``, or ``"auto"``
-            * ``mtu`` (int) – maximum transmission unit in bytes
-            * ``mode`` (str) – ``"access"``, ``"trunk"``, or ``"routed"``
-            * ``access_vlan`` (int) – untagged VLAN; relevant when mode is ``"access"``
-            * ``voice_vlan`` (int) – voice VLAN ID (``0`` = disabled)
-            * ``trunk_vlans`` (list of int) – tagged VLANs; empty = allow all
-            * ``native_vlan`` (int) – native VLAN on trunk ports
-
-        :raises NotImplementedError: If the driver does not implement this method.
-        :raises ValueError: If *interface* does not exist or an invalid value is
-            supplied for a configuration field.
-
-        Example – convert port to access VLAN 10 and add a description::
-
-            driver.set_interface(
-                "GigabitEthernet0/1",
-                {
-                    "description": "Workstation port",
-                    "enabled": True,
-                    "mode": "access",
-                    "access_vlan": 10,
-                },
-            )
-
-        Example – configure a trunk port::
-
-            driver.set_interface(
-                "GigabitEthernet0/2",
-                {
-                    "mode": "trunk",
-                    "trunk_vlans": [10, 20, 30],
-                    "native_vlan": 1,
-                },
-            )
-        """
-        raise NotImplementedError
-
-    def set_lag_members(self, lag_name: str, members: List[str]) -> None:
-        """
-        Sets the full member-port list of a LAG/trunk interface.
-
-        Diffs *members* against the LAG's current members (as reported by
-        ``get_interfaces()``'s ``lag_members`` field) and adds/removes ports
-        on the device to match. The LAG itself must already exist on the
-        device; this method only manages its membership.
-
-        :param lag_name: Name of the LAG/trunk interface as returned by
-            ``get_interfaces()`` (e.g. ``"Trk1"``, ``"ch1"``, ``"Lag1"``).
-        :param members: Full desired list of member port names.
-        :raises NotImplementedError: If the driver does not implement this method.
-        :raises ValueError: If *lag_name* does not refer to a valid LAG.
-
-        Example::
-
-            driver.set_lag_members("Lag1", ["GigabitEthernet0/1", "GigabitEthernet0/2"])
-        """
-        raise NotImplementedError
-
-    def set_poe_enabled(self, interface: str, enabled: bool) -> None:
-        """
-        Administratively enables or disables PoE on a single port.
-
-        :param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
-        :param enabled: ``True`` to enable PoE, ``False`` to disable it.
-        :raises NotImplementedError: If the driver does not support PoE control.
-        :raises ValueError: If the interface does not exist or does not support PoE.
-
-        Example::
-
-            driver.set_poe_enabled("GigabitEthernet0/1", False)  # cut power
-            driver.set_poe_enabled("GigabitEthernet0/1", True)   # restore
-        """
-        raise NotImplementedError
-
-    def power_cycle_port(self, interface: str, delay: int = 5) -> None:
-        """
-        Power-cycles the PoE port: cuts power, waits ``delay`` seconds, then
-        restores power.  Useful for rebooting a hung IP camera, AP, or IP phone
-        without physical access.
-
-        The method blocks until the full cycle (off → wait → on) is complete.
-        After it returns the port is back in the delivering/searching state.
-
-        :param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
-        :param delay: Seconds to keep the port powered off (default: 5).
-        :raises NotImplementedError: If the driver does not support PoE control.
-        :raises ValueError: If the interface does not exist, does not support PoE,
-            or PoE is administratively disabled on the port.
-
-        Example::
-
-            driver.power_cycle_port("GigabitEthernet0/1")        # 5 s off
-            driver.power_cycle_port("GigabitEthernet0/1", delay=15)  # 15 s off
-        """
-        raise NotImplementedError
+                driver.power_cycle_port("GigabitEthernet0/1")        # 5 s off
+                driver.power_cycle_port("GigabitEthernet0/1", delay=15)  # 15 s off
+            """
+            ...

napalm_device_types/systemd.py

@@ -0,0 +1,331 @@
+# -*- coding: utf-8 -*-
+"""systemd services: listing them in one round trip, and starting and stopping them.
+
+What systemd reports about its services, and how one is started or stopped, is
+the same on every host that runs it. So the command, its parse, the check of a
+unit name and the reading of an action's exit status live here once, and a
+driver only carries a command across: SSH, an API's exec endpoint, whatever it
+has.
+
+**Listing.** One command prints the installed unit files and, for every loaded
+service unit, what ``systemctl show`` knows about it -- state, boot state and
+main PID together, instead of asking ``systemctl is-enabled`` and ``systemctl
+show`` once per unit (two hundred round trips on an ordinary Linux host). The
+report is framed, and a report whose end is missing raises: a list cut short
+must never read as services that went away.
+
+**What counts as enabled.** A unit file state of ``enabled`` or
+``enabled-runtime``. ``static`` does not: such a unit starts only when
+something else pulls it in, and calling it enabled made every one of them look
+like a service of the host. The state is read from ``UnitFileState``, never
+from a column of ``list-unit-files``, whose second column has been followed by
+a preset column since systemd 245.
+
+**Starting and stopping.** ``systemctl`` runs bounded by ``timeout`` and never
+asks for a password, and its exit status is printed after it. The marker also
+keeps the output from ever being empty, which a transport that retries on an
+empty answer would otherwise take as a reason to run the action twice.
+"""
+
+from __future__ import annotations
+
+import re
+from shlex import quote
+from typing import Any, Dict, List, Set, Tuple, TYPE_CHECKING
+
+from napalm_device_types.models import ServiceDict
+from napalm_device_types.services import ServiceControlMixin
+
+_BEGIN = "SVC_BEGIN"
+_END = "SVC_END"
+_NO_SYSTEMD = "no-systemd"
+_SUFFIX = ".service"
+
+#: What ``systemctl show`` prints per unit. It prints them in its own order.
+_PROPERTIES = "Id,Names,LoadState,ActiveState,SubState,UnitFileState,MainPID"
+
+#: Picks the units whose file state is ``generated`` out of ``systemctl show``'s
+#: output, whatever order it prints the properties in.
+_GENERATED_AWK = (
+    'awk -F= \'NF<2{id="";g=0;next} $1=="Id"{id=$2} '
+    '$1=="UnitFileState"{g=($2=="generated")} id!=""&&g{print id;id="";g=0}\''
+)
+
+#: One line, POSIX ``sh``, read-only. The frame markers are printed in two
+#: halves so that a transport which echoes the command does not show them early.
+#: ``xargs -0`` passes escaped names such as ``foo\x2dbar.service`` unchanged.
+#: A generated unit -- the wrapper systemd makes for a SysV script -- has no unit
+#: file whose state says whether it starts at boot; ``systemctl is-enabled``
+#: asks the script's rc links instead, for those few units only.
+SYSTEMD_SERVICES_COMMAND = (
+    "printf '%s%s\\n' SVC_ BEGIN; "
+    "[ -d /run/systemd/system ] || echo '[no-systemd]'; "
+    "echo '[files]'; systemctl list-unit-files --type=service --no-legend --no-pager 2>/dev/null; "
+    "echo '[units]'; s=$(systemctl list-units --type=service --all --no-legend --no-pager --plain "
+    "2>/dev/null | awk '{print $1}' | tr '\\n' '\\0' | xargs -0 -r systemctl show --no-pager "
+    f"-p {_PROPERTIES} -- 2>/dev/null); printf '%s\\n' \"$s\"; "
+    f"echo '[generated]'; printf '%s\\n' \"$s\" | {_GENERATED_AWK} | while read -r u; do "
+    'printf \'%s %s\\n\' "$u" "$(systemctl is-enabled -- "$u" 2>/dev/null)"; done; '
+    "printf '%s%s\\n' SVC_ END"
+)
+
+#: The lifecycle actions :meth:`SystemdServicesMixin.manage_service` accepts.
+SERVICE_ACTIONS = ("start", "stop", "restart", "enable", "disable")
+
+#: Seconds an action may run on the host before ``timeout`` stops waiting for
+#: it. systemd itself carries on with the job.
+ACTION_TIMEOUT = 45
+
+#: What a transport should allow for one command: the action's own bound plus
+#: the round trip around it.
+_TRANSPORT_TIMEOUT = ACTION_TIMEOUT + 15
+
+_TIMED_OUT = 124  # timeout(1)'s exit status when the time ran out
+_RC_MARKER = "__SVC_RC="
+_RC_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE)
+
+#: The characters systemd allows in a unit name, with ``\xHH`` for any other byte.
+_UNIT_RE = re.compile(r"(?:[A-Za-z0-9_.:@-]|\\x[0-9A-Fa-f]{2})+")
+_MAX_UNIT_LENGTH = 255
+
+#: Terminal colour codes, which systemctl adds when a transport gives it a terminal.
+_ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]")
+
+_ENABLED = frozenset({"enabled", "enabled-runtime"})
+#: Unit file states of a service that is installed but need not be loaded.
+_INSTALLED = frozenset({"enabled", "enabled-runtime", "disabled", "indirect"})
+
+
+class SystemdUnavailable(NotImplementedError):
+    """The host does not run systemd; a driver may fall back to another init system."""
+
+
+def unit_name(name: str) -> str:
+    """*name* as a service unit's name without ``.service``, or ``ValueError``.
+
+    Accepts template instances (``wg-quick@wg0``), dots (``snapd.apparmor``),
+    colons and systemd's ``\\xHH`` escapes. Refuses a bare template
+    (``getty@``), a leading ``-`` that a command would read as an option, and
+    anything a shell would read.
+    """
+    base = name[: -len(_SUFFIX)] if name.endswith(_SUFFIX) else name
+    if (
+        not _UNIT_RE.fullmatch(base)
+        or base.startswith("-")
+        or base.endswith("@")
+        or len(base) + len(_SUFFIX) > _MAX_UNIT_LENGTH
+    ):
+        raise ValueError(f"Invalid service name: {name!r}")
+    return base
+
+
+def service_action_command(name: str, action: str) -> str:
+    """The shell command that applies *action* to the service *name*.
+
+    :raises ValueError: for an unknown action or an invalid name.
+    """
+    if action not in SERVICE_ACTIONS:
+        raise ValueError(f"Invalid action {action!r}; use one of {', '.join(SERVICE_ACTIONS)}")
+    unit = quote(unit_name(name) + _SUFFIX)
+    return (
+        f"timeout {ACTION_TIMEOUT} systemctl --no-ask-password {action} -- {unit} 2>&1; "
+        f"echo {_RC_MARKER}$?"
+    )
+
+
+def parse_action_result(output: str) -> Dict[str, Any]:
+    """``{"success", "output"}`` from what :func:`service_action_command` printed.
+
+    Only the exit status decides. A job still running when ``timeout`` gave up
+    is not reported as done, and output without a status is no success.
+    """
+    output = _ANSI_RE.sub("", output)
+    statuses = _RC_RE.findall(output)
+    text = _RC_RE.sub("", output).strip()
+    if not statuses:
+        return {"success": False, "output": text or "No exit status came back from the host."}
+    status = int(statuses[-1])
+    if status == 0:
+        return {"success": True, "output": text}
+    if status == _TIMED_OUT:
+        note = f"Still running after {ACTION_TIMEOUT} s; systemd carries on with the job."
+        return {"success": False, "output": f"{text}\n{note}".strip()}
+    return {"success": False, "output": text or f"systemctl exited with status {status}."}
+
+
+def _frame(output: str) -> List[str]:
+    lines = [line.strip() for line in _ANSI_RE.sub("", output).splitlines()]
+    try:
+        start = lines.index(_BEGIN)
+        end = lines.index(_END, start)
+    except ValueError:
+        raise ValueError("no intact systemd service report in the output") from None
+    return lines[start + 1 : end]
+
+
+def _sections(lines: List[str]) -> Dict[str, List[str]]:
+    sections: Dict[str, List[str]] = {}
+    current: List[str] = []
+    for line in lines:
+        if line.startswith("[") and line.endswith("]"):
+            current = sections.setdefault(line[1:-1], [])
+        else:
+            current.append(line)
+    return sections
+
+
+def _unit_blocks(lines: List[str]) -> List[Dict[str, str]]:
+    """``systemctl show``'s output, one dict per unit.
+
+    Units are separated by a blank line -- except where ``xargs`` split the
+    list over two runs and the blocks meet, so a key seen twice starts the next
+    unit as well.
+    """
+    blocks: List[Dict[str, str]] = []
+    current: Dict[str, str] = {}
+    for line in lines:
+        key, sep, value = line.partition("=")
+        if not sep or key in current:
+            if current:
+                blocks.append(current)
+            current = {}
+        if sep:
+            current[key] = value
+    if current:
+        blocks.append(current)
+    return blocks
+
+
+def _base(unit: str) -> str:
+    return unit[: -len(_SUFFIX)]
+
+
+def _main_pid(block: Dict[str, str]) -> int:
+    try:
+        return int(block.get("MainPID") or 0)
+    except ValueError:
+        return 0
+
+
+def _loaded(blocks: List[Dict[str, str]]) -> Tuple[Dict[str, ServiceDict], Set[str]]:
+    """The loaded services, and every name they go by (aliases included)."""
+    services: Dict[str, ServiceDict] = {}
+    names: Set[str] = set()
+    for block in blocks:
+        unit = block.get("Id", "")
+        if not unit.endswith(_SUFFIX) or block.get("LoadState") == "not-found":
+            continue
+        names.update(block.get("Names", unit).split())
+        running = block.get("ActiveState") == "active" and block.get("SubState") == "running"
+        services[_base(unit)] = {
+            "name": _base(unit),
+            "running": running,
+            "enabled": block.get("UnitFileState") in _ENABLED,
+            "pid": _main_pid(block) if running else 0,
+        }
+    return services, names
+
+
+def _installed(lines: List[str], known: Set[str]) -> Dict[str, ServiceDict]:
+    """Installed services that are not loaded: neither running nor starting now.
+
+    Templates, static units and aliases are left out -- the last also when an
+    older systemd lists an alias as ``enabled``, which is why every name a
+    loaded unit goes by is skipped.
+    """
+    services: Dict[str, ServiceDict] = {}
+    for line in lines:
+        parts = line.split()
+        if len(parts) < 2:
+            continue
+        unit, state = parts[0], parts[1]
+        if (
+            not unit.endswith(_SUFFIX)
+            or unit.endswith("@" + _SUFFIX)
+            or unit in known
+            or state not in _INSTALLED
+        ):
+            continue
+        services[_base(unit)] = {
+            "name": _base(unit),
+            "running": False,
+            "enabled": state in _ENABLED,
+            "pid": 0,
+        }
+    return services
+
+
+def _apply_generated(services: Dict[str, ServiceDict], lines: List[str]) -> None:
+    """Take a generated unit's boot state from ``is-enabled``'s answer."""
+    for line in lines:
+        parts = line.split()
+        if len(parts) == 2 and parts[0].endswith(_SUFFIX) and _base(parts[0]) in services:
+            services[_base(parts[0])]["enabled"] = parts[1] in _ENABLED
+
+
+def parse_systemd_services(output: str) -> List[ServiceDict]:
+    """Parse what :data:`SYSTEMD_SERVICES_COMMAND` printed, sorted by name.
+
+    Lists every loaded service unit but those that are not found, and every
+    installed one that is not loaded.
+
+    :raises SystemdUnavailable: when the host does not run systemd.
+    :raises ValueError: when the output carries no intact report.
+    """
+    sections = _sections(_frame(output))
+    if _NO_SYSTEMD in sections:
+        raise SystemdUnavailable("the host does not run systemd")
+    loaded, known = _loaded(_unit_blocks(sections.get("units", [])))
+    _apply_generated(loaded, sections.get("generated", []))
+    merged = {**_installed(sections.get("files", []), known), **loaded}
+    return [merged[name] for name in sorted(merged)]
+
+
+class SystemdServicesMixin(ServiceControlMixin):
+    """Implements :class:`ServiceControlMixin` for a driver whose host runs systemd.
+
+    The template form (README, "Function classes"): the command, the parse,
+    the check of the name and the reading of the exit status are the same
+    everywhere, so they are concrete here, and a driver supplies only
+    :meth:`_run_service_command` -- how a command reaches its host, and how it
+    gains root there when it needs to.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str:
+            """Run *command* with ``sh`` on the host and return what it printed.
+
+            *privileged* commands change the system and need root; *timeout*
+            is how long the transport should wait for the output, in seconds.
+            """
+            ...
+
+    def get_services(self) -> List[ServiceDict]:
+        """
+        Returns the services systemd knows, in one round trip.
+
+        * name (string) - the unit name without ``.service``
+        * running (bool) - active and running
+        * enabled (bool) - the unit file is enabled
+        * pid (int) - the main process; 0 when not running
+
+        :raises SystemdUnavailable: if the host does not run systemd.
+        :raises ValueError: if the host's output carried no intact report.
+        """
+        output = self._run_service_command(
+            SYSTEMD_SERVICES_COMMAND, privileged=False, timeout=_TRANSPORT_TIMEOUT
+        )
+        return parse_systemd_services(output)
+
+    def manage_service(self, name: str, action: str) -> Dict[str, Any]:
+        """
+        Applies *action* (start, stop, restart, enable, disable) to the service *name*.
+
+        :returns: ``{"success": bool, "output": str}``
+        :raises ValueError: for an unknown action or an invalid name, before
+            anything is sent.
+        """
+        command = service_action_command(name, action)
+        output = self._run_service_command(command, privileged=True, timeout=_TRANSPORT_TIMEOUT)
+        return parse_action_result(output)

napalm_device_types/updates.py

@@ -0,0 +1,73 @@
+# -*- coding: utf-8 -*-
+"""Pending package updates and applying them.
+
+The two device types that offer this used to declare it under two different
+names -- ``get_available_updates`` and ``get_pending_updates`` -- with
+``napalm-linux`` carrying an alias between them so netOrk could call either.
+One name now, the one netOrk and four drivers already used.
+
+Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
+Nothing exists at runtime until a concrete driver implements it, so mixing
+this class in can never shadow a working implementation from a sibling base.
+"""
+
+from __future__ import annotations
+
+from typing import List, TYPE_CHECKING
+
+from napalm_device_types.models import ApplyUpdatesResultDict, UpdateDict
+
+
+class UpdateMixin:
+    if TYPE_CHECKING:
+
+        def get_available_updates(self) -> List[UpdateDict]:
+            """
+            Returns a list of packages that have a newer version available.
+
+            Each entry contains:
+
+            * name (string) - package name
+            * current_version (string) - currently installed version
+            * new_version (string) - version available in the repository
+
+            Example::
+
+                [
+                    {
+                        "name": "openssh-server",
+                        "current_version": "1:9.2p1-2+deb12u1",
+                        "new_version": "1:9.2p1-2+deb12u2",
+                    },
+                ]
+            """
+            ...
+
+        def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict:
+            """
+            Upgrades the given packages to the newest available version.
+
+            Only packages that are already installed may be upgraded; this method
+            does **not** install new packages.  Pass an empty list to upgrade
+            **all** packages that have pending updates.
+
+            :param packages: List of package names to upgrade.  Each name must
+                match ``^[a-zA-Z0-9_\\-\\+\\.]+$``; a :exc:`ValueError` is raised
+                for any name that does not conform.
+            :returns: A dict with:
+
+                * success (bool) – ``True`` if the package manager exited without error
+                * output (string) – combined stdout / stderr from the package manager
+                * error (string, optional) – short error message when *success* is ``False``
+
+            :raises ValueError: If any package name fails the safety check.
+
+            Example::
+
+                result = driver.apply_updates(["openssh-server", "curl"])
+                # → {"success": True, "output": "Reading package lists...\\n..."}
+
+                # Upgrade everything:
+                result = driver.apply_updates([])
+            """
+            ...

pyproject.toml

@@ -4,10 +4,10 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "0.5.0"
+version = "2.2.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
-requires-python = ">=3.9"
+requires-python = ">=3.10"
 license = { text = "Apache-2.0" }
 authors = [
     { name = "Christian Manivong", email = "christian@manivong.de" },
@@ -31,7 +31,6 @@ classifiers = [
     "License :: OSI Approved :: Apache Software License",
     "Operating System :: OS Independent",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
@@ -61,6 +60,6 @@ where = ["."]
 include = ["napalm_device_types*"]
 
 [tool.mypy]
-python_version = "3.9"
+python_version = "3.10"
 strict = true
 ignore_missing_imports = true

tests/test_dhcp_diff_apply.py

@@ -0,0 +1,208 @@
+"""Tests for DhcpServerMixin's generic reservation diff/apply mechanism.
+
+diff_dhcp_reservations/apply_dhcp_reservationset are concrete methods on the
+mixin (not overridden by concrete drivers) — they only depend on the three
+abstract methods (get_dhcp_reservations/apply_dhcp_reservation/
+commit_dhcp_reservations), so a fake in-memory driver is enough to exercise
+them fully; no real device or vendor driver needed. See README.md "Design
+principle: generic vs. device-specific logic" for why this logic lives here
+and not in a vendor driver.
+"""
+
+from typing import Any, Dict, List, Optional
+
+import pytest
+from napalm_device_types import DhcpServerMixin, FirewallDriver, ResidentialGatewayDriver
+from napalm_device_types.dhcp import normalize_mac
+from napalm_device_types.models import DhcpReservationDict
+
+
+def _reservation(**overrides: Any) -> DhcpReservationDict:
+    base: DhcpReservationDict = {
+        "uuid": "",
+        "mac": "aa:bb:cc:dd:ee:01",
+        "ip": "10.10.20.50",
+        "hostname": "nas",
+        "description": "Home NAS",
+        "subnet": "10.10.20.0/24",
+    }
+    base.update(overrides)  # type: ignore[typeddict-item]
+    return base
+
+
+class _FakeDhcpServer(DhcpServerMixin):
+    """In-memory fake — no network, no Kea/UCI specifics."""
+
+    def __init__(self, live: Optional[List[DhcpReservationDict]] = None) -> None:
+        self.live = live or []
+        self.applied: List[Dict[str, Any]] = []
+        self.commits = 0
+
+    def get_dhcp_reservations(self) -> List[DhcpReservationDict]:
+        return self.live
+
+    def apply_dhcp_reservation(
+        self, reservation: DhcpReservationDict, *, uuid: Optional[str] = None
+    ) -> Dict[str, Any]:
+        self.applied.append({"reservation": reservation, "uuid": uuid})
+        return {"success": True}
+
+    def commit_dhcp_reservations(self) -> Dict[str, Any]:
+        self.commits += 1
+        return {"success": True}
+
+
+class TestNormalizeMac:
+    def test_lowercases_and_colon_separates(self):
+        assert normalize_mac("AA-BB-CC-DD-EE-01") == "aa:bb:cc:dd:ee:01"
+
+    def test_accepts_bare_hex(self):
+        assert normalize_mac("aabbccddee01") == "aa:bb:cc:dd:ee:01"
+
+    def test_accepts_cisco_dotted(self):
+        assert normalize_mac("aabb.ccdd.ee01") == "aa:bb:cc:dd:ee:01"
+
+    def test_passes_through_unparseable_value_lowercased(self):
+        # Not 12 hex digits — return something stable rather than raising, so a
+        # malformed device response degrades to "never matches" instead of
+        # aborting the whole diff.
+        assert normalize_mac("not-a-mac") == "not-a-mac"
+
+    def test_handles_none(self):
+        assert normalize_mac(None) == ""
+
+
+class TestAbstractContract:
+    """The device hooks are declared for type checkers, not implemented.
+
+    A driver that never implemented them does not carry them at runtime, so
+    `hasattr` is a truthful capability probe and the declaration cannot shadow a
+    working implementation inherited from a sibling base.
+    """
+
+    @pytest.mark.parametrize(
+        "method",
+        [
+            "get_dhcp_reservations",
+            "apply_dhcp_reservation",
+            "commit_dhcp_reservations",
+            "get_dhcp_subnets",
+            "apply_dhcp_subnet",
+            "commit_dhcp_subnets",
+        ],
+    )
+    def test_hook_absent_until_a_driver_implements_it(self, method):
+        assert not hasattr(DhcpServerMixin, method)
+
+    def test_calling_a_missing_hook_fails_loudly(self):
+        with pytest.raises(AttributeError):
+            DhcpServerMixin().get_dhcp_reservations()
+
+
+class TestMixedIntoDriverTypes:
+    """Both firewalls and residential gateways run DHCP servers, so the mixin
+    must be reachable from either base class without re-declaring it."""
+
+    def test_firewall_driver_has_dhcp_reservation_methods(self):
+        assert issubclass(FirewallDriver, DhcpServerMixin)
+
+    def test_residential_gateway_driver_has_dhcp_reservation_methods(self):
+        assert issubclass(ResidentialGatewayDriver, DhcpServerMixin)
+
+
+class TestDiff:
+    def test_empty_device_yields_all_adds(self):
+        driver = _FakeDhcpServer([])
+        diff = driver.diff_dhcp_reservations(
+            [_reservation(), _reservation(mac="aa:bb:cc:dd:ee:02")]
+        )
+
+        assert len(diff["add"]) == 2
+        assert diff["update"] == []
+
+    def test_identical_reservation_produces_no_change(self):
+        live = _reservation(uuid="u1")
+        driver = _FakeDhcpServer([live])
+
+        diff = driver.diff_dhcp_reservations([_reservation()])
+
+        assert diff == {"add": [], "update": []}
+
+    def test_changed_ip_produces_update_with_changed_fields(self):
+        driver = _FakeDhcpServer([_reservation(uuid="u1")])
+
+        diff = driver.diff_dhcp_reservations([_reservation(ip="10.10.20.51")])
+
+        assert diff["add"] == []
+        assert len(diff["update"]) == 1
+        assert diff["update"][0]["uuid"] == "u1"
+        assert diff["update"][0]["changed_fields"] == ["ip"]
+
+    def test_changed_subnet_produces_update_not_add(self):
+        # A host moved to another VLAN keeps its MAC — that is an update of the
+        # existing reservation, not a second reservation for the same MAC.
+        driver = _FakeDhcpServer([_reservation(uuid="u1")])
+
+        diff = driver.diff_dhcp_reservations(
+            [_reservation(ip="10.30.20.50", subnet="10.30.20.0/24")]
+        )
+
+        assert diff["add"] == []
+        assert diff["update"][0]["changed_fields"] == ["ip", "subnet"]
+
+    def test_mac_formatting_differences_still_match(self):
+        driver = _FakeDhcpServer([_reservation(uuid="u1", mac="AA-BB-CC-DD-EE-01")])
+
+        diff = driver.diff_dhcp_reservations([_reservation(mac="aabb.ccdd.ee01")])
+
+        assert diff == {"add": [], "update": []}
+
+    def test_unmanaged_live_reservation_is_never_deleted(self):
+        # Hand-created reservations must survive — the desired set was never
+        # meant to describe them.
+        driver = _FakeDhcpServer([_reservation(uuid="u1", mac="aa:bb:cc:dd:ee:99")])
+
+        diff = driver.diff_dhcp_reservations([_reservation()])
+
+        assert len(diff["add"]) == 1
+        assert diff["update"] == []
+        assert "delete" not in diff
+
+
+class TestApplyReservationSet:
+    def test_applies_adds_and_updates_then_commits(self):
+        driver = _FakeDhcpServer([_reservation(uuid="u1", ip="10.10.20.9")])
+
+        lines = list(
+            driver.apply_dhcp_reservationset(
+                [_reservation(), _reservation(mac="aa:bb:cc:dd:ee:02", hostname="printer")]
+            )
+        )
+
+        assert driver.commits == 1
+        assert len(driver.applied) == 2
+        # The update targets the live uuid; the add does not.
+        by_uuid = {entry["uuid"] for entry in driver.applied}
+        assert by_uuid == {"u1", None}
+        assert any(line.startswith("[add]") for line in lines)
+        assert any(line.startswith("[update]") for line in lines)
+        assert lines[-1].startswith("[commit]")
+
+    def test_progress_lines_name_the_reservation(self):
+        driver = _FakeDhcpServer([])
+
+        lines = list(driver.apply_dhcp_reservationset([_reservation()]))
+
+        assert "aa:bb:cc:dd:ee:01" in lines[0]
+        assert "10.10.20.50" in lines[0]
+
+    def test_no_changes_skips_the_commit(self):
+        # Committing means reloading the DHCP daemon (Kea `service/reconfigure`),
+        # which drops in-flight requests. A no-op diff must not cause that.
+        driver = _FakeDhcpServer([_reservation(uuid="u1")])
+
+        lines = list(driver.apply_dhcp_reservationset([_reservation()]))
+
+        assert driver.commits == 0
+        assert driver.applied == []
+        assert lines == ["[commit] no changes"]

tests/test_dhcp_subnet_diff_apply.py

@@ -0,0 +1,228 @@
+"""Tests for DhcpServerMixin's generic subnet diff/apply mechanism.
+
+Same shape as test_dhcp_diff_apply.py: diff_dhcp_subnets/apply_dhcp_subnetset
+are concrete methods that only depend on the abstract trio
+(get_dhcp_subnets/apply_dhcp_subnet/commit_dhcp_subnets), so an in-memory
+fake exercises them fully. See README.md "Design principle: generic vs.
+device-specific logic".
+
+A subnet is a heavier object than a reservation: a wrong `pools` or
+`option_data` takes a whole VLAN offline rather than one host, so the tests
+below lean on the never-delete and preserve-unmanaged-options guarantees.
+"""
+
+from typing import Any, Dict, List, Optional
+
+import pytest
+from napalm_device_types import DhcpServerMixin
+from napalm_device_types.models import DhcpSubnetDict
+
+
+def _subnet(**overrides: Any) -> DhcpSubnetDict:
+    base: DhcpSubnetDict = {
+        "uuid": "",
+        "subnet": "10.10.20.0/24",
+        "description": "Home",
+        "pools": ["10.10.20.100-10.10.20.200"],
+        "option_data": {
+            "routers": ["10.10.20.1"],
+            "domain_name_servers": ["10.10.20.1"],
+        },
+        "match_client_id": False,
+    }
+    base.update(overrides)  # type: ignore[typeddict-item]
+    return base
+
+
+class _FakeDhcpServer(DhcpServerMixin):
+    def __init__(self, live: Optional[List[DhcpSubnetDict]] = None) -> None:
+        self.live = live or []
+        self.applied: List[Dict[str, Any]] = []
+        self.commits = 0
+
+    def get_dhcp_subnets(self) -> List[DhcpSubnetDict]:
+        return self.live
+
+    def apply_dhcp_subnet(
+        self, subnet: DhcpSubnetDict, *, uuid: Optional[str] = None
+    ) -> Dict[str, Any]:
+        self.applied.append({"subnet": subnet, "uuid": uuid})
+        return {"success": True}
+
+    def commit_dhcp_subnets(self) -> Dict[str, Any]:
+        self.commits += 1
+        return {"success": True}
+
+
+# ── diff: adds ────────────────────────────────────────────────────────────────
+
+
+def test_unknown_subnet_is_an_add() -> None:
+    diff = _FakeDhcpServer().diff_dhcp_subnets([_subnet()])
+    assert len(diff["add"]) == 1
+    assert diff["add"][0]["subnet"] == "10.10.20.0/24"
+    assert diff["update"] == []
+
+
+def test_matching_subnet_with_no_changes_is_neither() -> None:
+    live = _subnet(uuid="dev-1")
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([_subnet()])
+    assert diff == {"add": [], "update": []}
+
+
+# ── diff: identity is the CIDR ────────────────────────────────────────────────
+
+
+def test_subnets_are_matched_on_cidr() -> None:
+    live = _subnet(uuid="dev-1", description="renamed on the device")
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([_subnet()])
+    assert diff["add"] == []
+    assert diff["update"][0]["uuid"] == "dev-1"
+    assert diff["update"][0]["changed_fields"] == ["description"]
+
+
+def test_a_different_cidr_is_a_new_subnet_not_an_update() -> None:
+    live = _subnet(uuid="dev-1", subnet="10.10.20.0/24")
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([_subnet(subnet="10.10.30.0/24")])
+    assert len(diff["add"]) == 1
+    assert diff["update"] == []
+
+
+# ── diff: compared fields ─────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("description", "Office"),
+        ("pools", ["10.10.20.50-10.10.20.99"]),
+        ("match_client_id", True),
+    ],
+)
+def test_changed_field_is_reported(field: str, value: Any) -> None:
+    live = _subnet(uuid="dev-1")
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([_subnet(**{field: value})])
+    assert diff["update"][0]["changed_fields"] == [field]
+
+
+def test_changed_option_is_reported_as_option_data() -> None:
+    live = _subnet(uuid="dev-1")
+    desired = _subnet(
+        option_data={
+            "routers": ["10.10.20.1"],
+            "domain_name_servers": ["10.10.20.1"],
+            "domain_search": ["home.local", "office.local"],
+        }
+    )
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([desired])
+    assert diff["update"][0]["changed_fields"] == ["option_data"]
+
+
+def test_an_unmanaged_option_on_the_device_is_not_a_change() -> None:
+    """Absent key means "not managed" -- it must not provoke an update.
+
+    Kea autocollects routers/domain_name_servers/ntp_servers. A caller that
+    only wants to set domain_search would otherwise diff-fight the server
+    forever, reloading the DHCP daemon on every run.
+    """
+    live = _subnet(
+        uuid="dev-1",
+        option_data={
+            "routers": ["10.10.20.1"],
+            "domain_name_servers": ["10.10.20.1"],
+            "ntp_servers": ["10.10.20.1"],
+        },
+    )
+    desired = _subnet(option_data={"domain_search": ["home.local"]})
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([desired])
+    assert diff["update"][0]["changed_fields"] == ["option_data"]
+
+    # ...and once it matches, it stays quiet.
+    live2 = _subnet(
+        uuid="dev-1",
+        option_data={
+            "routers": ["10.10.20.1"],
+            "domain_name_servers": ["10.10.20.1"],
+            "ntp_servers": ["10.10.20.1"],
+            "domain_search": ["home.local"],
+        },
+    )
+    assert _FakeDhcpServer([live2]).diff_dhcp_subnets([desired]) == {"add": [], "update": []}
+
+
+def test_option_order_is_significant_for_domain_search() -> None:
+    """Search order decides which zone answers an unqualified name first."""
+    live = _subnet(uuid="dev-1", option_data={"domain_search": ["a.local", "b.local"]})
+    desired = _subnet(option_data={"domain_search": ["b.local", "a.local"]})
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([desired])
+    assert diff["update"][0]["changed_fields"] == ["option_data"]
+
+
+# ── diff: never delete ────────────────────────────────────────────────────────
+
+
+def test_live_subnet_absent_from_desired_is_never_deleted() -> None:
+    """Deleting a subnet takes a whole VLAN's DHCP down. Never implicit."""
+    live = _subnet(uuid="dev-1", subnet="10.10.99.0/24")
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([_subnet()])
+    assert len(diff["add"]) == 1
+    assert diff["update"] == []
+    assert "delete" not in diff
+
+
+# ── apply ─────────────────────────────────────────────────────────────────────
+
+
+def test_apply_creates_then_commits() -> None:
+    fake = _FakeDhcpServer()
+    lines = list(fake.apply_dhcp_subnetset([_subnet()]))
+    assert fake.applied[0]["uuid"] is None
+    assert fake.commits == 1
+    assert any(line.startswith("[add]") for line in lines)
+
+
+def test_apply_updates_in_place_with_the_device_uuid() -> None:
+    fake = _FakeDhcpServer([_subnet(uuid="dev-1")])
+    list(fake.apply_dhcp_subnetset([_subnet(description="Office")]))
+    assert fake.applied[0]["uuid"] == "dev-1"
+    assert fake.commits == 1
+
+
+def test_apply_does_not_commit_when_nothing_changed() -> None:
+    """A commit reconfigures the DHCP daemon -- too costly for a no-op run."""
+    fake = _FakeDhcpServer([_subnet(uuid="dev-1")])
+    lines = list(fake.apply_dhcp_subnetset([_subnet()]))
+    assert fake.commits == 0
+    assert lines == ["[commit] no changes"]
+
+
+def test_apply_names_the_subnet_in_its_progress_line() -> None:
+    fake = _FakeDhcpServer()
+    lines = list(fake.apply_dhcp_subnetset([_subnet()]))
+    assert "10.10.20.0/24" in lines[0]
+
+
+def test_apply_reports_changed_fields_on_update() -> None:
+    fake = _FakeDhcpServer([_subnet(uuid="dev-1")])
+    lines = list(fake.apply_dhcp_subnetset([_subnet(description="Office")]))
+    assert "description" in lines[0]
+
+
+# ── abstract surface ──────────────────────────────────────────────────────────
+
+
+class _BareDriver(DhcpServerMixin):
+    pass
+
+
+@pytest.mark.parametrize(
+    "method", ["get_dhcp_subnets", "apply_dhcp_subnet", "commit_dhcp_subnets"]
+)
+def test_device_specific_methods_are_absent_until_implemented(method: str) -> None:
+    """Declared under TYPE_CHECKING, so they do not exist until a driver adds them."""
+    assert not hasattr(_BareDriver, method)
+
+
+def test_calling_a_missing_device_method_fails_loudly() -> None:
+    with pytest.raises(AttributeError):
+        _BareDriver().get_dhcp_subnets()

tests/test_firewall_diff_apply.py

@@ -0,0 +1,171 @@
+"""Tests for FirewallDriver's generic diff/apply mechanism.
+
+diff_firewall_rules/apply_firewall_ruleset are concrete methods on the base
+class (not overridden by concrete drivers) — they only depend on the three
+abstract methods (get_firewall_rules/apply_firewall_rule/commit_firewall_rules),
+so a fake in-memory driver is enough to exercise them fully; no real device
+or vendor driver needed. See README.md "Design principle: generic vs.
+device-specific logic" for why this logic lives here and not in a vendor
+driver.
+"""
+
+from typing import Any, Dict, List, Optional
+
+import pytest
+from napalm_device_types import FirewallDriver
+from napalm_device_types.models import FirewallRuleDict
+
+
+def _rule(**overrides: Any) -> FirewallRuleDict:
+    base: FirewallRuleDict = {
+        "uuid": "",
+        "description": "allow_mgmt_to_fw_gui",
+        "action": "pass",
+        "interface": "lan",
+        "direction": "in",
+        "protocol": "tcp",
+        "source_net": "MGMT_NET",
+        "source_port": "",
+        "destination_net": "(self)",
+        "destination_port": "https",
+        "enabled": True,
+        "quick": True,
+        "log": False,
+    }
+    base.update(overrides)  # type: ignore[typeddict-item]
+    return base
+
+
+class _FakeFirewall(FirewallDriver):
+    """In-memory fake — no network, no OPNsense/vendor specifics."""
+
+    def __init__(self, live_rules: Optional[List[FirewallRuleDict]] = None) -> None:
+        self.live_rules = live_rules or []
+        self.applied: List[Dict[str, Any]] = []
+        self.committed = False
+
+    def get_firewall_rules(self) -> List[FirewallRuleDict]:
+        return self.live_rules
+
+    def apply_firewall_rule(
+        self, rule: FirewallRuleDict, *, uuid: Optional[str] = None
+    ) -> Dict[str, Any]:
+        self.applied.append({"rule": rule, "uuid": uuid})
+        return {"success": True}
+
+    def commit_firewall_rules(self) -> Dict[str, Any]:
+        self.committed = True
+        return {"success": True}
+
+
+class TestAbstractContract:
+    """The three device hooks are declared, never implemented, on the base.
+
+    They exist for type checkers only, so a driver that did not implement them
+    does not carry them at runtime -- which is what keeps `hasattr` truthful and
+    stops the declaration from overriding a sibling base's working version.
+    """
+
+    @pytest.mark.parametrize(
+        "method", ["get_firewall_rules", "apply_firewall_rule", "commit_firewall_rules"]
+    )
+    def test_hook_absent_until_a_driver_implements_it(self, method):
+        class _Bare(FirewallDriver):
+            def __init__(self) -> None:
+                pass
+
+        assert not hasattr(_Bare, method)
+
+    def test_calling_a_missing_hook_fails_loudly(self):
+        class _Bare(FirewallDriver):
+            def __init__(self) -> None:
+                pass
+
+        with pytest.raises(AttributeError):
+            _Bare().get_firewall_rules()
+
+
+class TestDiffFirewallRules:
+    def test_desired_rule_missing_live_is_an_add(self):
+        driver = _FakeFirewall(live_rules=[])
+        diff = driver.diff_firewall_rules([_rule()])
+
+        assert len(diff["add"]) == 1
+        assert diff["add"][0]["description"] == "allow_mgmt_to_fw_gui"
+        assert diff["update"] == []
+
+    def test_matching_rule_with_changed_field_is_an_update(self):
+        live = _rule(uuid="abc-123", action="block")
+        driver = _FakeFirewall(live_rules=[live])
+        diff = driver.diff_firewall_rules([_rule(action="pass")])
+
+        assert diff["add"] == []
+        assert len(diff["update"]) == 1
+        update = diff["update"][0]
+        assert update["uuid"] == "abc-123"
+        assert update["changed_fields"] == ["action"]
+        assert update["rule"]["action"] == "pass"
+
+    def test_identical_rule_produces_no_diff(self):
+        live = _rule(uuid="abc-123")
+        driver = _FakeFirewall(live_rules=[live])
+        diff = driver.diff_firewall_rules([_rule()])
+
+        assert diff["add"] == []
+        assert diff["update"] == []
+
+    def test_live_rule_not_in_desired_is_never_deleted(self):
+        """v1 never deletes -- rules present live but absent from `desired`
+        are simply ignored, not reported for removal."""
+        live = _rule(uuid="abc-123", description="some_unmanaged_rule")
+        driver = _FakeFirewall(live_rules=[live])
+        diff = driver.diff_firewall_rules([])
+
+        assert diff == {"add": [], "update": []}
+        assert "delete" not in diff
+
+    def test_multiple_changed_fields_all_reported(self):
+        live = _rule(uuid="abc-123", action="block", protocol="udp", log=True)
+        driver = _FakeFirewall(live_rules=[live])
+        diff = driver.diff_firewall_rules([_rule(action="pass", protocol="tcp", log=False)])
+
+        assert set(diff["update"][0]["changed_fields"]) == {"action", "protocol", "log"}
+
+
+class TestApplyFirewallRuleset:
+    def test_adds_are_applied_with_no_uuid(self):
+        driver = _FakeFirewall(live_rules=[])
+        list(driver.apply_firewall_ruleset([_rule()]))
+
+        assert len(driver.applied) == 1
+        assert driver.applied[0]["uuid"] is None
+        assert driver.applied[0]["rule"]["description"] == "allow_mgmt_to_fw_gui"
+
+    def test_updates_are_applied_with_existing_uuid(self):
+        live = _rule(uuid="abc-123", action="block")
+        driver = _FakeFirewall(live_rules=[live])
+        list(driver.apply_firewall_ruleset([_rule(action="pass")]))
+
+        assert len(driver.applied) == 1
+        assert driver.applied[0]["uuid"] == "abc-123"
+
+    def test_commits_after_applying(self):
+        driver = _FakeFirewall(live_rules=[])
+        list(driver.apply_firewall_ruleset([_rule()]))
+
+        assert driver.committed is True
+
+    def test_yields_a_progress_line_per_change(self):
+        driver = _FakeFirewall(live_rules=[])
+        lines = list(driver.apply_firewall_ruleset([_rule(), _rule(description="second_rule")]))
+
+        assert len(lines) >= 2
+        assert all(isinstance(line, str) for line in lines)
+
+    def test_no_changes_still_commits_but_applies_nothing(self):
+        live = _rule(uuid="abc-123")
+        driver = _FakeFirewall(live_rules=[live])
+        list(driver.apply_firewall_ruleset([_rule()]))
+
+        assert driver.applied == []
+        assert driver.committed is True

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_kernel_facts.py

@@ -0,0 +1,151 @@
+"""get_kernel_facts: what the running kernel has built and loaded.
+
+A kernel CVE's preconditions ask whether a module is loaded or a build option
+set. Reading that is the same on every Linux host -- one read-only command and
+its parse -- so both live here once, and a driver only carries the command
+across (#268 in netOrk).
+"""
+
+from __future__ import annotations
+
+import base64
+import gzip
+import os
+import subprocess
+
+import pytest
+
+from napalm_device_types import OSDriver
+from napalm_device_types.kernel import (
+    KERNEL_FACTS_COMMAND,
+    KernelFactsMixin,
+    module_name,
+    parse_kernel_facts,
+)
+
+REPORT = """[release]
+6.1.0-25-amd64
+[loaded]
+tipc
+nf_tables
+[builtin]
+kernel/net/ipv4/tcp_cubic.ko
+kernel/drivers/char/tpm/tpm-tis.ko
+[available]
+kernel/net/tipc/tipc.ko.xz
+kernel/net/can/can-raw.ko.zst
+kernel/net/netfilter/nf_tables.ko
+[config]
+CONFIG_TIPC=m
+CONFIG_BPF_JIT=y
+CONFIG_DEFAULT_HOSTNAME="(none)"
+CONFIG_HZ=250
+"""
+
+
+def _wire(report: str, *, noise: str = "") -> str:
+    """The report as the command prints it: framed, gzipped, base64 in lines."""
+    payload = base64.encodebytes(gzip.compress(report.encode())).decode()
+    return f"{noise}KFACTS_BEGIN\n{payload}KFACTS_END\n"
+
+
+class TestParsing:
+    def test_every_section_is_read(self):
+        facts = parse_kernel_facts(_wire(REPORT))
+
+        assert facts["release"] == "6.1.0-25-amd64"
+        assert facts["loaded"] == ["nf_tables", "tipc"]
+        assert facts["builtin"] == ["tcp_cubic", "tpm_tis"]
+        assert facts["available"] == ["can_raw", "nf_tables", "tipc"]
+        assert facts["config"] == {
+            "CONFIG_TIPC": "m",
+            "CONFIG_BPF_JIT": "y",
+            "CONFIG_DEFAULT_HOSTNAME": "(none)",
+            "CONFIG_HZ": "250",
+        }
+
+    def test_a_section_never_printed_is_none_not_empty(self):
+        """``None`` is "could not read"; an empty list would claim "read it,
+        and there is nothing" -- and that is what turns a module into
+        ``not_met`` downstream."""
+        facts = parse_kernel_facts(_wire("[release]\n6.1.0\n[loaded]\n"))
+
+        assert facts["loaded"] == []
+        assert facts["builtin"] is None
+        assert facts["available"] is None
+        assert facts["config"] is None
+
+    def test_whatever_surrounds_the_frame_is_ignored(self):
+        """A screen-scraping transport may echo the command or a banner."""
+        noise = "Last login: today\nprintf '%s%s\\n' KFACTS_ BEGIN; ...\n"
+
+        assert parse_kernel_facts(_wire(REPORT, noise=noise))["release"] == "6.1.0-25-amd64"
+
+    def test_output_without_the_frame_raises(self):
+        with pytest.raises(ValueError):
+            parse_kernel_facts("sh: gzip: not found\n")
+
+    def test_a_damaged_payload_raises(self):
+        with pytest.raises(ValueError):
+            parse_kernel_facts("KFACTS_BEGIN\nnot base64 at all!\nKFACTS_END\n")
+
+
+class TestModuleNames:
+    @pytest.mark.parametrize(
+        "raw, name",
+        [
+            ("tipc", "tipc"),
+            ("kernel/net/tipc/tipc.ko", "tipc"),
+            ("kernel/net/tipc/tipc.ko.zst", "tipc"),
+            ("kernel/net/can/can-raw.ko.xz", "can_raw"),
+            ("CAN-RAW", "can_raw"),
+            ("  nf_tables  ", "nf_tables"),
+        ],
+    )
+    def test_dash_and_underscore_are_one_name(self, raw, name):
+        """The kernel treats ``-`` and ``_`` in module names as the same."""
+        assert module_name(raw) == name
+
+
+class TestTheCommand:
+    def test_the_frame_is_not_in_the_command_itself(self):
+        """An echoing transport prints the command back; the markers must only
+        appear once the command has run."""
+        assert "KFACTS_BEGIN" not in KERNEL_FACTS_COMMAND
+        assert "KFACTS_END" not in KERNEL_FACTS_COMMAND
+
+    def test_it_writes_nothing(self):
+        for verb in ("modprobe", "insmod", "rmmod", "sudo", " > ", ">>"):
+            assert verb not in KERNEL_FACTS_COMMAND
+
+    @pytest.mark.skipif(os.uname().sysname != "Linux", reason="reads a Linux kernel")
+    def test_it_runs_and_parses_on_this_host(self):
+        out = subprocess.run(
+            ["sh", "-c", KERNEL_FACTS_COMMAND], capture_output=True, text=True, timeout=60
+        ).stdout
+
+        facts = parse_kernel_facts(out)
+
+        assert facts["release"] == os.uname().release
+        assert facts["loaded"] is None or all(isinstance(m, str) for m in facts["loaded"])
+
+
+class TestTheTemplate:
+    def test_a_driver_supplies_only_the_transport(self):
+        class Driver(KernelFactsMixin):
+            def _run_kernel_facts_command(self, command: str) -> str:
+                self.sent = command
+                return _wire(REPORT)
+
+        driver = Driver()
+        facts = driver.get_kernel_facts()
+
+        assert driver.sent == KERNEL_FACTS_COMMAND
+        assert facts["release"] == "6.1.0-25-amd64"
+
+    def test_not_every_os_driver_has_it(self):
+        """A Windows host is an OSDriver too, and has no Linux kernel to read:
+        ``hasattr`` has to stay a truthful answer, so the drivers that can mix
+        this in themselves."""
+        assert not issubclass(OSDriver, KernelFactsMixin)
+        assert not hasattr(OSDriver, "get_kernel_facts")

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_ping_sweep.py

@@ -0,0 +1,221 @@
+"""Tests for the generic ping sweep (PingSweepMixin + driver_supports_ping)."""
+
+import pytest
+from napalm.base import NetworkDriver
+
+from napalm_device_types import DeviceTypeDriver, PingSweepMixin, driver_supports_ping
+
+
+# ── Fakes ─────────────────────────────────────────────────────────────────────
+
+
+def _napalm_ok(rtt: float, probes: int = 1) -> dict:
+    """A NAPALM-format ping() reply for a reachable destination."""
+    return {
+        "success": {
+            "probes_sent": probes,
+            "packet_loss": 0,
+            "rtt_min": rtt,
+            "rtt_avg": rtt,
+            "rtt_max": rtt,
+            "rtt_stddev": 0.0,
+            "results": [{"ip_address": "10.0.0.1", "rtt": rtt}],
+        }
+    }
+
+
+def _napalm_lost(probes: int = 1) -> dict:
+    """A NAPALM-format ping() reply where every probe was lost."""
+    return {
+        "success": {
+            "probes_sent": probes,
+            "packet_loss": probes,
+            "rtt_min": 0.0,
+            "rtt_avg": 0.0,
+            "rtt_max": 0.0,
+            "rtt_stddev": 0.0,
+            "results": [],
+        }
+    }
+
+
+class FakePingDriver(PingSweepMixin):
+    """Minimal driver exposing ping() — stands in for a real vendor driver."""
+
+    def __init__(self, replies=None, raises=None):
+        self.replies = replies or {}
+        self.raises = raises or {}
+        self.calls = []
+
+    def ping(self, destination, source="", ttl=255, timeout=2, size=100, count=5, vrf=""):
+        self.calls.append({"destination": destination, "timeout": timeout, "count": count})
+        if destination in self.raises:
+            raise self.raises[destination]
+        return self.replies.get(destination, _napalm_lost())
+
+
+class NoPingDriver(PingSweepMixin):
+    """Driver without its own ping() — inherits NAPALM's NotImplementedError stub."""
+
+    ping = NetworkDriver.ping
+
+
+class OptedOutDriver(FakePingDriver):
+    """Driver that implements ping() but declares it unusable for sweeps."""
+
+    SUPPORTS_PING = False
+
+
+# ── driver_supports_ping ──────────────────────────────────────────────────────
+
+
+def test_driver_supports_ping_false_for_unoverridden_ping():
+    assert driver_supports_ping(NoPingDriver) is False
+
+
+def test_driver_supports_ping_false_for_base_network_driver():
+    assert driver_supports_ping(NetworkDriver) is False
+
+
+def test_driver_supports_ping_true_when_overridden():
+    assert driver_supports_ping(FakePingDriver) is True
+
+
+def test_driver_supports_ping_honours_explicit_opt_out():
+    assert driver_supports_ping(OptedOutDriver) is False
+
+
+def test_driver_supports_ping_false_for_class_without_ping():
+    class Bare:
+        pass
+
+    assert driver_supports_ping(Bare) is False
+
+
+def test_device_type_driver_exposes_supports_ping_classmethod():
+    assert DeviceTypeDriver.supports_ping() is False
+    assert FakePingDriver.supports_ping() is True
+
+
+# ── ping_sweep ────────────────────────────────────────────────────────────────
+
+
+def test_ping_sweep_marks_reachable_and_unreachable_hosts():
+    driver = FakePingDriver(replies={"10.0.0.1": _napalm_ok(1.5)})
+
+    result = driver.ping_sweep(["10.0.0.1", "10.0.0.2"])
+
+    assert result["scanned"] == 2
+    assert result["alive_count"] == 1
+    assert result["truncated"] is False
+    assert result["entries"] == [
+        {"ip": "10.0.0.1", "alive": True, "rtt_ms": 1.5},
+        {"ip": "10.0.0.2", "alive": False, "rtt_ms": None},
+    ]
+
+
+def test_ping_sweep_uses_single_fast_probe_by_default():
+    driver = FakePingDriver()
+
+    driver.ping_sweep(["10.0.0.1"])
+
+    assert driver.calls == [{"destination": "10.0.0.1", "timeout": 1, "count": 1}]
+
+
+def test_ping_sweep_forwards_count_and_timeout():
+    driver = FakePingDriver()
+
+    driver.ping_sweep(["10.0.0.1"], count=3, timeout=5)
+
+    assert driver.calls == [{"destination": "10.0.0.1", "timeout": 5, "count": 3}]
+
+
+def test_ping_sweep_treats_error_reply_as_unreachable():
+    driver = FakePingDriver(replies={"10.0.0.9": {"error": "unknown host"}})
+
+    result = driver.ping_sweep(["10.0.0.9"])
+
+    assert result["entries"][0]["alive"] is False
+    assert result["entries"][0]["error"] == "unknown host"
+    assert result["alive_count"] == 0
+
+
+def test_ping_sweep_records_exception_and_continues():
+    driver = FakePingDriver(
+        replies={"10.0.0.2": _napalm_ok(2.0)},
+        raises={"10.0.0.1": RuntimeError("session closed")},
+    )
+
+    result = driver.ping_sweep(["10.0.0.1", "10.0.0.2"])
+
+    assert result["entries"][0] == {
+        "ip": "10.0.0.1",
+        "alive": False,
+        "rtt_ms": None,
+        "error": "session closed",
+    }
+    assert result["entries"][1]["alive"] is True
+    assert result["scanned"] == 2
+
+
+def test_ping_sweep_truncates_at_max_targets():
+    driver = FakePingDriver()
+
+    result = driver.ping_sweep([f"10.0.0.{i}" for i in range(1, 11)], max_targets=4)
+
+    assert result["scanned"] == 4
+    assert result["truncated"] is True
+    assert len(driver.calls) == 4
+
+
+def test_ping_sweep_respects_class_level_max_targets():
+    class SmallSweepDriver(FakePingDriver):
+        PING_SWEEP_MAX_TARGETS = 2
+
+    driver = SmallSweepDriver()
+
+    result = driver.ping_sweep(["10.0.0.1", "10.0.0.2", "10.0.0.3"])
+
+    assert result["scanned"] == 2
+    assert result["truncated"] is True
+
+
+def test_ping_sweep_reports_progress_per_destination():
+    driver = FakePingDriver(replies={"10.0.0.1": _napalm_ok(1.0)})
+    seen = []
+
+    driver.ping_sweep(
+        ["10.0.0.1", "10.0.0.2"],
+        on_progress=lambda done, total: seen.append((done, total)),
+    )
+
+    assert seen == [(1, 2), (2, 2)]
+
+
+def test_ping_sweep_on_empty_destination_list():
+    driver = FakePingDriver()
+
+    result = driver.ping_sweep([])
+
+    assert result == {"entries": [], "scanned": 0, "alive_count": 0, "truncated": False}
+
+
+def test_ping_sweep_raises_when_driver_cannot_ping():
+    driver = NoPingDriver()
+
+    with pytest.raises(NotImplementedError):
+        driver.ping_sweep(["10.0.0.1"])
+
+
+def test_ping_sweep_stops_when_stop_requested():
+    driver = FakePingDriver()
+    calls = {"n": 0}
+
+    def _should_stop():
+        calls["n"] += 1
+        return calls["n"] > 1
+
+    result = driver.ping_sweep(["10.0.0.1", "10.0.0.2", "10.0.0.3"], should_stop=_should_stop)
+
+    assert result["scanned"] == 1
+    assert len(driver.calls) == 1

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_role_contracts.py

@@ -0,0 +1,141 @@
+# -*- coding: utf-8 -*-
+"""The contract that makes multi-role drivers safe.
+
+A role base declares what a device of that kind can be asked for. It must not
+*implement* anything -- not even a placeholder. A ``NotImplementedError`` stub
+on a base class is not neutral under multiple inheritance: it wins the MRO
+against a sibling base's working implementation and silently replaces it. That
+failure has hit this codebase three times (OpenWrt's seven forwarding methods,
+QNAP's ``get_services``, OpenMediaVault avoiding ``StorageDriver`` altogether).
+
+These tests pin the property that prevents a fourth.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+import pytest
+
+from napalm_device_types import (
+    AccessPointDriver,
+    DeviceTypeDriver,
+    FirewallDriver,
+    HypervisorDriver,
+    MediaDriver,
+    OSDriver,
+    PhoneDriver,
+    ResidentialGatewayDriver,
+    StorageDriver,
+    SwitchDriver,
+)
+from napalm_device_types.roles import primary_role_of, roles_of
+
+ROLE_BASES = [
+    AccessPointDriver,
+    FirewallDriver,
+    HypervisorDriver,
+    MediaDriver,
+    OSDriver,
+    PhoneDriver,
+    ResidentialGatewayDriver,
+    StorageDriver,
+    SwitchDriver,
+]
+
+
+@pytest.mark.parametrize("base", ROLE_BASES, ids=lambda b: b.__name__)
+class TestRoleBasesAreContractsOnly:
+    def test_defines_no_methods_at_runtime(self, base):
+        """A role base is a declaration. Anything callable it owns can shadow a
+        sibling base, so it must own nothing callable at all."""
+        own = [
+            name
+            for name, val in vars(base).items()
+            if not name.startswith("__")
+            and (inspect.isfunction(val) or isinstance(val, (classmethod, staticmethod)))
+        ]
+        assert own == [], f"{base.__name__} implements {own}; move it to a function class"
+
+    def test_declares_a_role_key(self, base):
+        assert isinstance(vars(base).get("ROLE"), str) and vars(base)["ROLE"]
+
+    def test_has_a_real_docstring(self, base):
+        """TYPE_LABEL used to be assigned above the triple-quoted string, which
+        made it a bare expression rather than a docstring -- __doc__ was None on
+        all seven bases, killing help() and IDE hovers."""
+        assert base.__doc__ and base.__doc__.strip()
+
+
+class TestDeclaredMethodsDoNotExistAtRuntime:
+    """`hasattr` is netOrk's capability probe. It only tells the truth when a
+    declared-but-unimplemented method is genuinely absent."""
+
+    @pytest.mark.parametrize(
+        ("base", "method"),
+        [
+            (StorageDriver, "get_disks"),
+            (StorageDriver, "get_volumes"),
+            (HypervisorDriver, "get_vms"),
+            (FirewallDriver, "send_wake_on_lan"),
+            (SwitchDriver, "set_vlan"),
+            (AccessPointDriver, "get_wireless_config"),
+            (OSDriver, "get_processes"),
+            (ResidentialGatewayDriver, "get_wan_status"),
+            (PhoneDriver, "get_sip_accounts"),
+            (MediaDriver, "get_playback_state"),
+        ],
+    )
+    def test_absent_until_a_driver_implements_it(self, base, method):
+        assert not hasattr(base, method)
+
+
+class TestNoShadowingAcrossRoles:
+    """The regression test for the bug class."""
+
+    def test_role_base_listed_first_does_not_shadow_sibling(self):
+        """StorageDriver precedes the working mixin -- the order that broke QNAP."""
+
+        class WorkingPackages:
+            def get_packages(self):
+                return [{"name": "vim", "version": "9.0"}]
+
+            def get_services(self):
+                return [{"name": "sshd", "state": "running"}]
+
+        class Combined(StorageDriver, WorkingPackages):
+            pass
+
+        assert Combined.get_packages is WorkingPackages.get_packages
+        assert Combined.get_services is WorkingPackages.get_services
+
+    def test_two_role_bases_can_be_combined(self):
+        """A QNAP is NAS, hypervisor and Linux host. It must be able to say so."""
+
+        class Nas(StorageDriver, HypervisorDriver, OSDriver):
+            pass
+
+        assert [r.__name__ for r in roles_of(Nas)] == [
+            "StorageDriver",
+            "HypervisorDriver",
+            "OSDriver",
+        ]
+
+
+class TestRoleIntrospection:
+    def test_primary_role_follows_base_order(self):
+        class Nas(StorageDriver, OSDriver):
+            pass
+
+        class Host(OSDriver, StorageDriver):
+            pass
+
+        assert primary_role_of(Nas) == "storage"
+        assert primary_role_of(Host) == "linux"
+
+    def test_driver_without_a_role_has_none(self):
+        class Bare(DeviceTypeDriver):
+            pass
+
+        assert roles_of(Bare) == []
+        assert primary_role_of(Bare) is None

tests/test_send_wake_on_lan.py

@@ -0,0 +1,41 @@
+"""Tests for the FirewallDriver.send_wake_on_lan contract.
+
+``send_wake_on_lan`` is declared on ``FirewallDriver`` but implemented by very
+few firewalls. It used to exist as a ``NotImplementedError`` stub; it is now a
+``TYPE_CHECKING`` declaration, so a driver that never implemented it simply does
+not have the attribute. That is what lets ``hasattr`` answer honestly, and what
+stops the declaration from shadowing a working implementation inherited from a
+sibling base.
+"""
+
+import pytest
+from napalm_device_types import FirewallDriver
+
+
+class _BareFirewall(FirewallDriver):
+    """FirewallDriver.__init__ is NetworkDriver's, which itself raises
+    NotImplementedError — override with a no-op so tests exercise
+    send_wake_on_lan itself, not construction."""
+
+    def __init__(self):
+        pass
+
+
+def test_absent_on_a_driver_that_never_implemented_it():
+    assert not hasattr(_BareFirewall, "send_wake_on_lan")
+
+
+def test_calling_it_anyway_fails_loudly():
+    """A caller that skips the hasattr check must not get silence."""
+    with pytest.raises(AttributeError):
+        _BareFirewall().send_wake_on_lan("AA:BB:CC:DD:EE:FF")
+
+
+def test_subclass_can_implement_send_wake_on_lan():
+    class MyFirewall(_BareFirewall):
+        def send_wake_on_lan(self, mac_address: str, interface: str = ""):
+            return {"success": True, "output": f"woke {mac_address} via {interface}"}
+
+    assert hasattr(MyFirewall, "send_wake_on_lan")
+    result = MyFirewall().send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
+    assert result == {"success": True, "output": "woke AA:BB:CC:DD:EE:FF via lan"}

tests/test_systemd_services.py

@@ -0,0 +1,412 @@
+"""systemd services: listing them in one round trip, and starting and stopping them.
+
+What systemd reports, and how a unit is started or stopped, is the same on
+every host that runs it -- so the command, its parse, the name check and the
+reading of the exit status live here once, and a driver only carries a command
+across (napalm-linux#7, napalm-proxmox#6).
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+
+import pytest
+
+from napalm_device_types import OSDriver
+from napalm_device_types.systemd import (
+    ACTION_TIMEOUT,
+    SERVICE_ACTIONS,
+    SYSTEMD_SERVICES_COMMAND,
+    SystemdServicesMixin,
+    SystemdUnavailable,
+    parse_action_result,
+    parse_systemd_services,
+    service_action_command,
+    unit_name,
+)
+
+FILES = """\
+apparmor.service                             enabled         enabled
+ssh.service                                  enabled         enabled
+sshd.service                                 alias           -
+getty@.service                               enabled         enabled
+rsync.service                                disabled        enabled
+cups.service                                 indirect        enabled
+plymouth-quit.service                        static          -
+systemd-networkd-wait-online.service         enabled-runtime enabled
+nfs-server.service                           masked          enabled
+"""
+
+UNITS = """\
+MainPID=812
+Id=ssh.service
+Names=ssh.service sshd.service
+LoadState=loaded
+ActiveState=active
+SubState=running
+UnitFileState=enabled
+
+MainPID=0
+Id=apparmor.service
+Names=apparmor.service
+LoadState=loaded
+ActiveState=active
+SubState=exited
+UnitFileState=enabled
+
+MainPID=900
+Id=getty@tty1.service
+Names=getty@tty1.service
+LoadState=loaded
+ActiveState=active
+SubState=running
+UnitFileState=enabled
+
+MainPID=0
+Id=systemd-fsck@dev-disk-by\\x2dlabel-BOOT.service
+Names=systemd-fsck@dev-disk-by\\x2dlabel-BOOT.service
+LoadState=loaded
+ActiveState=inactive
+SubState=dead
+UnitFileState=static
+
+MainPID=0
+Id=display-manager.service
+Names=display-manager.service
+LoadState=not-found
+ActiveState=inactive
+SubState=dead
+UnitFileState=
+
+MainPID=0
+Id=nfs-server.service
+Names=nfs-server.service
+LoadState=masked
+ActiveState=inactive
+SubState=dead
+UnitFileState=masked
+
+MainPID=0
+Id=systemd-networkd-wait-online.service
+Names=systemd-networkd-wait-online.service
+LoadState=loaded
+ActiveState=active
+SubState=exited
+UnitFileState=enabled-runtime
+"""
+
+
+GENERATED_UNIT = """
+MainPID=0
+Id=rrdcached.service
+Names=rrdcached.service
+LoadState=loaded
+ActiveState=active
+SubState=running
+UnitFileState=generated
+"""
+
+
+def _wire(
+    files: str = FILES,
+    units: str = UNITS,
+    *,
+    generated: str = "",
+    noise: str = "",
+    end: bool = True,
+) -> str:
+    """The report as the command prints it, framed."""
+    tail = "SVC_END\n" if end else ""
+    return f"{noise}SVC_BEGIN\n[files]\n{files}[units]\n{units}[generated]\n{generated}{tail}"
+
+
+def _by_name(services):
+    return {s["name"]: s for s in services}
+
+
+class TestParseSystemdServices:
+    def test_a_loaded_unit_is_read_with_its_state(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert services["ssh"] == {"name": "ssh", "running": True, "enabled": True, "pid": 812}
+        assert services["apparmor"] == {
+            "name": "apparmor",
+            "running": False,
+            "enabled": True,
+            "pid": 0,
+        }
+
+    def test_enabled_means_enabled_now_not_merely_installed(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert services["systemd-networkd-wait-online"]["enabled"] is True
+        assert services[r"systemd-fsck@dev-disk-by\x2dlabel-BOOT"]["enabled"] is False
+        assert services["nfs-server"]["enabled"] is False
+
+    def test_a_generated_unit_takes_its_boot_state_from_is_enabled(self):
+        """A SysV script's unit is generated; only is-enabled knows its rc links."""
+        raw = _wire(units=UNITS + GENERATED_UNIT, generated="rrdcached.service enabled\n")
+
+        assert _by_name(parse_systemd_services(raw))["rrdcached"]["enabled"] is True
+
+    def test_a_generated_unit_is_not_enabled_unless_is_enabled_says_so(self):
+        raw = _wire(units=UNITS + GENERATED_UNIT, generated="rrdcached.service disabled\n")
+
+        assert _by_name(parse_systemd_services(raw))["rrdcached"]["enabled"] is False
+
+    def test_a_unit_that_is_not_there_is_left_out(self):
+        assert "display-manager" not in _by_name(parse_systemd_services(_wire()))
+
+    def test_an_installed_unit_that_is_not_loaded_is_listed(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert services["rsync"] == {"name": "rsync", "running": False, "enabled": False, "pid": 0}
+        assert services["cups"]["enabled"] is False
+
+    def test_templates_and_static_files_that_are_not_loaded_are_not(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert "getty@" not in services
+        assert "plymouth-quit" not in services
+        assert services["getty@tty1"]["running"] is True
+
+    def test_an_alias_never_appears_beside_its_unit(self):
+        assert "sshd" not in _by_name(parse_systemd_services(_wire()))
+
+    def test_an_alias_that_older_systemd_calls_enabled_does_not_either(self):
+        files = (
+            "\n".join(
+                "sshd.service enabled enabled" if line.startswith("sshd.service") else line
+                for line in FILES.splitlines()
+            )
+            + "\n"
+        )
+
+        assert "sshd" not in _by_name(parse_systemd_services(_wire(files=files)))
+
+    def test_an_escaped_name_survives(self):
+        assert r"systemd-fsck@dev-disk-by\x2dlabel-BOOT" in _by_name(
+            parse_systemd_services(_wire())
+        )
+
+    def test_blocks_run_together_are_still_told_apart(self):
+        """xargs may split the unit list across two systemctl runs."""
+        units = UNITS.replace(
+            "UnitFileState=enabled\n\nMainPID=0\nId=apparmor",
+            "UnitFileState=enabled\nMainPID=0\nId=apparmor",
+        )
+
+        services = _by_name(parse_systemd_services(_wire(units=units)))
+
+        assert services["ssh"]["pid"] == 812
+        assert services["apparmor"]["running"] is False
+
+    def test_the_list_is_sorted_by_name(self):
+        names = [s["name"] for s in parse_systemd_services(_wire())]
+
+        assert names == sorted(names)
+
+    def test_terminal_colours_in_the_report_are_dropped(self):
+        files = FILES.replace(
+            "rsync.service                                disabled",
+            "rsync.service \x1b[0;1;31mdisabled\x1b[0m",
+        )
+
+        assert "rsync" in _by_name(parse_systemd_services(_wire(files=files)))
+
+    def test_whatever_surrounds_the_frame_is_ignored(self):
+        noisy = _wire(noise="user@host:~$ systemctl ...\n") + "user@host:~$ "
+
+        assert "ssh" in _by_name(parse_systemd_services(noisy))
+
+    def test_a_cut_short_report_raises(self):
+        """A missing tail must not read as services that went away."""
+        with pytest.raises(ValueError):
+            parse_systemd_services(_wire(end=False))
+
+    def test_output_without_the_frame_raises(self):
+        with pytest.raises(ValueError):
+            parse_systemd_services("bash: systemctl: command not found\n")
+
+    def test_a_host_without_systemd_says_so(self):
+        raw = "SVC_BEGIN\n[no-systemd]\n[files]\n[units]\nSVC_END\n"
+
+        with pytest.raises(SystemdUnavailable):
+            parse_systemd_services(raw)
+
+    def test_no_systemd_is_a_not_implemented_error(self):
+        assert issubclass(SystemdUnavailable, NotImplementedError)
+
+
+class TestTheCommand:
+    def test_the_frame_is_not_in_the_command_itself(self):
+        """An echoing transport must not show the end marker early."""
+        assert "SVC_END" not in SYSTEMD_SERVICES_COMMAND
+        assert "SVC_BEGIN" not in SYSTEMD_SERVICES_COMMAND
+
+    def test_it_changes_nothing(self):
+        for verb in ("start", "stop", "restart", "enable", "disable", "mask"):
+            assert f"systemctl {verb}" not in SYSTEMD_SERVICES_COMMAND
+
+    @pytest.mark.skipif(not os.path.isdir("/run/systemd/system"), reason="needs systemd")
+    def test_it_runs_and_parses_on_this_host(self):
+        out = subprocess.run(
+            ["sh", "-c", SYSTEMD_SERVICES_COMMAND], capture_output=True, text=True, timeout=60
+        ).stdout
+
+        services = _by_name(parse_systemd_services(out))
+
+        assert "systemd-journald" in services
+        assert services["systemd-journald"]["running"] is True
+
+
+class TestUnitName:
+    @pytest.mark.parametrize(
+        ("raw", "name"),
+        [
+            ("ssh", "ssh"),
+            ("ssh.service", "ssh"),
+            ("getty@tty1", "getty@tty1"),
+            ("wg-quick@wg0", "wg-quick@wg0"),
+            ("snapd.apparmor", "snapd.apparmor"),
+            ("systemd-backlight@backlight:acpi_video0", "systemd-backlight@backlight:acpi_video0"),
+            (r"systemd-fsck@dev-disk-by\x2dlabel-BOOT", r"systemd-fsck@dev-disk-by\x2dlabel-BOOT"),
+        ],
+    )
+    def test_a_unit_name_is_accepted(self, raw, name):
+        assert unit_name(raw) == name
+
+    @pytest.mark.parametrize(
+        "raw",
+        [
+            "",
+            "-x",
+            "foo@",
+            "foo@.service",
+            "a b",
+            "a;b",
+            "$(id)",
+            "a/b",
+            r"bad\x2",
+            "ssh\n",
+            "x" * 256,
+        ],
+    )
+    def test_anything_else_is_refused(self, raw):
+        with pytest.raises(ValueError):
+            unit_name(raw)
+
+
+class TestServiceActionCommand:
+    def test_the_command_is_bounded_and_never_asks(self):
+        cmd = service_action_command("getty@tty1", "restart")
+
+        assert cmd.startswith(f"timeout {ACTION_TIMEOUT} systemctl --no-ask-password restart -- ")
+        assert "getty@tty1.service" in cmd
+
+    def test_an_escaped_name_is_quoted_for_the_shell(self):
+        cmd = service_action_command(r"systemd-fsck@dev-disk-by\x2dlabel-BOOT", "stop")
+
+        assert r"'systemd-fsck@dev-disk-by\x2dlabel-BOOT.service'" in cmd
+
+    def test_its_exit_status_is_printed_after_it(self):
+        assert service_action_command("ssh", "start").endswith("; echo __SVC_RC=$?")
+
+    def test_the_actions(self):
+        assert SERVICE_ACTIONS == ("start", "stop", "restart", "enable", "disable")
+
+    def test_an_unknown_action_is_refused(self):
+        with pytest.raises(ValueError):
+            service_action_command("ssh", "mask")
+
+    def test_an_invalid_name_is_refused(self):
+        with pytest.raises(ValueError):
+            service_action_command("ssh; reboot", "stop")
+
+
+class TestParseActionResult:
+    def test_exit_status_zero_is_success(self):
+        assert parse_action_result("__SVC_RC=0\n") == {"success": True, "output": ""}
+
+    def test_what_systemctl_printed_comes_back_without_the_marker(self):
+        raw = (
+            "Created symlink /etc/systemd/system/multi-user.target.wants/cron.service.\n__SVC_RC=0"
+        )
+
+        result = parse_action_result(raw)
+
+        assert result["success"] is True
+        assert result["output"].startswith("Created symlink")
+        assert "__SVC_RC" not in result["output"]
+
+    def test_terminal_colours_are_dropped(self):
+        """systemctl colours its errors when a transport gives it a terminal."""
+        raw = (
+            "\x1b[0;1;31mFailed to restart x.service: Unit x.service not found.\x1b[0m\n"
+            "__SVC_RC=5\n"
+        )
+
+        assert parse_action_result(raw)["output"] == (
+            "Failed to restart x.service: Unit x.service not found."
+        )
+
+    def test_a_failure_keeps_its_message(self):
+        raw = "Failed to start foo.service: Unit foo.service not found.\n__SVC_RC=5\n"
+
+        assert parse_action_result(raw) == {
+            "success": False,
+            "output": "Failed to start foo.service: Unit foo.service not found.",
+        }
+
+    def test_a_job_still_running_at_the_timeout_is_not_called_done(self):
+        result = parse_action_result("__SVC_RC=124\n")
+
+        assert result["success"] is False
+        assert str(ACTION_TIMEOUT) in result["output"]
+
+    def test_no_exit_status_is_no_success(self):
+        assert parse_action_result("Connection reset\n")["success"] is False
+
+    def test_the_echoed_command_is_not_taken_for_the_status(self):
+        raw = "timeout 45 systemctl restart -- cron.service 2>&1; echo __SVC_RC=$?\n__SVC_RC=1\n"
+
+        assert parse_action_result(raw)["success"] is False
+
+
+class _Driver(SystemdServicesMixin):
+    def __init__(self, reply: str) -> None:
+        self.reply = reply
+        self.calls: list = []
+
+    def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str:
+        self.calls.append((command, privileged, timeout))
+        return self.reply
+
+
+class TestSystemdServicesMixin:
+    def test_listing_runs_the_command_unprivileged(self):
+        driver = _Driver(_wire())
+
+        assert "ssh" in _by_name(driver.get_services())
+        assert driver.calls == [(SYSTEMD_SERVICES_COMMAND, False, ACTION_TIMEOUT + 15)]
+
+    def test_an_action_runs_privileged_and_reports_its_outcome(self):
+        driver = _Driver("__SVC_RC=0\n")
+
+        assert driver.manage_service("cron", "restart") == {"success": True, "output": ""}
+        command, privileged, timeout = driver.calls[0]
+        assert command == service_action_command("cron", "restart")
+        assert privileged is True
+        assert timeout > ACTION_TIMEOUT
+
+    def test_an_invalid_request_is_refused_before_anything_is_sent(self):
+        driver = _Driver("__SVC_RC=0\n")
+
+        with pytest.raises(ValueError):
+            driver.manage_service("cron;reboot", "stop")
+        assert driver.calls == []
+
+    def test_not_every_os_driver_has_it(self):
+        assert not hasattr(OSDriver, "manage_service")
+        assert callable(getattr(SystemdServicesMixin, "manage_service"))

tests/test_vm_cpu_types.py

@@ -0,0 +1,58 @@
+"""A new VM's virtual CPU model can be chosen, from a list the hypervisor offers.
+
+Proxmox gives a VM created without a ``cpu`` argument the ``kvm64`` model,
+which has no AVX -- and MongoDB 5.0 and later will not start without it. Which
+model is right depends on the cluster (``host`` cannot live-migrate between
+different CPUs, ``x86-64-v3`` does not start on a CPU older than Haswell), so
+the caller chooses, from entries that say what each model provides and whether
+the node at hand can run it.
+
+The declarations sit under ``TYPE_CHECKING`` (see test_role_contracts), so the
+signature is read from the source rather than from the class.
+"""
+
+from __future__ import annotations
+
+import ast
+import inspect
+from typing import List, get_type_hints
+
+import napalm_device_types.hypervisor as hypervisor_module
+from napalm_device_types import HypervisorDriver
+from napalm_device_types.models import VMCpuTypeDict
+
+
+def _declared(name: str) -> ast.FunctionDef:
+    tree = ast.parse(inspect.getsource(hypervisor_module))
+    for node in ast.walk(tree):
+        if isinstance(node, ast.FunctionDef) and node.name == name:
+            return node
+    raise AssertionError(f"HypervisorDriver does not declare {name}()")
+
+
+class TestCreateVmTakesACpuType:
+    def test_cpu_type_is_an_optional_keyword(self):
+        fn = _declared("create_vm_from_cloud_init")
+        kwonly = {arg.arg: default for arg, default in zip(fn.args.kwonlyargs, fn.args.kw_defaults)}
+        assert "cpu_type" in kwonly
+        default = kwonly["cpu_type"]
+        assert isinstance(default, ast.Constant) and default.value is None
+
+
+class TestCpuTypeListing:
+    def test_is_declared(self):
+        assert _declared("get_vm_cpu_types").returns is not None
+
+    def test_absent_until_a_driver_implements_it(self):
+        """netOrk probes capabilities with hasattr; a hypervisor without a
+        choice of CPU model must not seem to offer one."""
+        assert not hasattr(HypervisorDriver, "get_vm_cpu_types")
+
+    def test_entry_shape(self):
+        assert get_type_hints(VMCpuTypeDict) == {
+            "name": str,
+            "description": str,
+            "features": List[str],
+            "available": bool,
+            "default": bool,
+        }

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