docs(firewall): document the identifier convention Wake-on-LAN depends on

A driver whose send_wake_on_lan() interface is not the name get_interfaces()
is keyed by leaves callers with no way to offer a valid choice. OPNsense keys
by the physical device ("em0") but wakes by the assigned name ("lan"), and
rejects the former — so the assigned name has to travel with the interface
data as an "identifier" key.

README.md

@@ -22,6 +22,75 @@ 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.
 
+## 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 +109,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
 

napalm_device_types/__init__.py

@@ -28,14 +28,19 @@ Also provided:
 * :class:`~napalm_device_types.config_lifecycle.ConfigLifecycleMixin` --
   stand-alone mixin to reduce duplication of config lifecycle methods across
   drivers.
+* :class:`~napalm_device_types.dhcp.DhcpServerMixin` -- static DHCP
+  reservation read/diff/apply, mixed into the firewall and gateway base
+  classes.
 """
 
 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.ping_sweep import PingSweepMixin, driver_supports_ping
 from napalm_device_types.residential_gateway import ResidentialGatewayDriver
 from napalm_device_types.storage import StorageDriver
 from napalm_device_types.switch import SwitchDriver
@@ -44,12 +49,17 @@ __all__ = [
     "AccessPointDriver",
     "ConfigLifecycleMixin",
     "DeviceTypeDriver",
+    "DhcpServerMixin",
     "FingerprintRule",
     "FirewallDriver",
     "HypervisorDriver",
     "OSDriver",
+    "PingSweepMixin",
     "PortSpec",
     "ResidentialGatewayDriver",
     "StorageDriver",
     "SwitchDriver",
+    "driver_supports_ping",
+    "normalize_cidr",
+    "normalize_mac",
 ]

napalm_device_types/access_point.py

@@ -280,6 +280,25 @@ class AccessPointDriver(DeviceTypeDriver):
         """
         raise NotImplementedError
 
+    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", [])
+        """
+        raise NotImplementedError
+
     def get_dot1x_config(self) -> Dict[str, Dot1XConfigDict]:
         """
         Returns the 802.1X / WPA-Enterprise (RADIUS) configuration per SSID.

napalm_device_types/base.py

@@ -12,6 +12,8 @@ from typing import NamedTuple
 
 from napalm.base import NetworkDriver
 
+from napalm_device_types.ping_sweep import PingSweepMixin
+
 
 class FingerprintRule(NamedTuple):
     """Single pattern-matching rule for device fingerprinting.
@@ -45,13 +47,15 @@ class PortSpec(NamedTuple):
     mandatory: bool = False
 
 
-class DeviceTypeDriver(NetworkDriver):
+class DeviceTypeDriver(PingSweepMixin, 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:
 

napalm_device_types/dhcp.py

@@ -0,0 +1,398 @@
+# -*- 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
+
+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.
+    # ------------------------------------------------------------------
+
+    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.
+        """
+        raise NotImplementedError
+
+    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}``.
+        """
+        raise NotImplementedError
+
+    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.
+        """
+        raise NotImplementedError
+
+    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}``.
+        """
+        raise NotImplementedError
+
+    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}``.
+        """
+        raise NotImplementedError
+
+    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}``.
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # 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,10 +10,14 @@ Usage::
             ...
 """
 
-from typing import Any, Dict, List
+from typing import Any, Dict, Iterator, List, Optional
 from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types.dhcp import DhcpServerMixin
 from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
 from napalm_device_types.models import (
+    FirewallRuleDict,
+    FirewallRuleDiffDict,
+    FirewallRuleUpdateDict,
     HealthMetricsDict,
     NATTranslationDict,
     PackageDict,
@@ -22,8 +26,22 @@ from napalm_device_types.models import (
     VPNTunnelDict,
 )
 
+_FIREWALL_RULE_COMPARE_FIELDS = (
+    "action",
+    "interface",
+    "direction",
+    "protocol",
+    "source_net",
+    "source_port",
+    "destination_net",
+    "destination_port",
+    "log",
+    "quick",
+    "enabled",
+)
 
-class FirewallDriver(DeviceTypeDriver):
+
+class FirewallDriver(DhcpServerMixin, DeviceTypeDriver):
     TYPE_LABEL: str = "Firewall"
     """
     Abstract intermediate driver for firewall/security devices.
@@ -160,6 +178,44 @@ class FirewallDriver(DeviceTypeDriver):
         """
         raise NotImplementedError
 
+    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.
+
+        :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.
+
+        :returns: A dict with:
+
+            * success (bool) - ``True`` if the magic packet was sent without error
+            * output (string) - human-readable status message
+
+        Example::
+
+            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"}
+
+        .. note::
+
+            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.
+        """
+        raise NotImplementedError
+
     def get_packages(self) -> List[PackageDict]:
         """
         Returns all packages / plugins currently known to the firewall's
@@ -298,3 +354,131 @@ class FirewallDriver(DeviceTypeDriver):
             )
         """
         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 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.
+        """
+        raise NotImplementedError
+
+    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}``.
+        """
+        raise NotImplementedError
+
+    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}``.
+        """
+        raise NotImplementedError
+
+    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/hypervisor.py

@@ -19,6 +19,7 @@ from napalm_device_types.models import (
     NetworkTargetDict,
     PackageDict,
     SnapshotDict,
+    StorageTargetDict,
     StorageVolumeDict,
     VMConfigDict,
     VMDict,
@@ -548,47 +549,68 @@ class HypervisorDriver(DeviceTypeDriver):
         self,
         name: str,
         *,
-        template: str,
+        image_url: str,
         cpu: int,
         memory: int,
         nics: List[NICConfigDict],
         cloud_init_config: Dict[str, Any],
+        image_checksum: str | None = None,
         ssh_public_keys: List[str] | None = None,
         disk_resize_gb: int | None = None,
+        storage: str | None = None,
+        download_timeout: int = 300,
         timeout: int = 180,
     ) -> VMProvisionResultDict:
         """
-        Create a new virtual machine from a Cloud-Init template.
+        Create a new virtual machine from a cloud image via Cloud-Init.
 
-        Clones a pre-existing VM template, configures virtual network interfaces,
-        and injects Cloud-Init configuration via a storage snippet or similar
-        mechanism.  The resulting VM is left in a running state.
+        Downloads the cloud image (qcow2/raw) directly on the hypervisor if not
+        already cached there, creates a new VM shell, imports the image as its
+        root disk, configures virtual network interfaces, and injects Cloud-Init
+        configuration via a storage snippet or similar mechanism. The resulting
+        VM is left in a running state.
+
+        Implementations should cache downloaded images by URL/filename on the
+        hypervisor so repeated provisioning from the same image does not
+        re-download it every time.
 
         Args:
             name (string) - new VM display name
-            template (string) - hypervisor-internal ID/name of the template VM to clone
+            image_url (string) - URL of the cloud image to download and use as
+              the VM's root disk (e.g. an official Debian/Ubuntu cloud image).
             cpu (int) - number of virtual CPUs to assign
             memory (int) - RAM to assign in megabytes
             nics (list[NICConfigDict]) - list of network interface configurations.
               First NIC is primary (DHCP by default); subsequent NICs are optional.
               Each entry specifies bridge, optional vlan_tag (access) or trunk_vlan_tags,
-              and dhcp flag.
+              dhcp flag, and an optional explicit mac address (omit to let the
+              hypervisor auto-generate one; needed when a caller must know the
+              MAC ahead of time, e.g. to create a matching DHCP reservation).
             cloud_init_config (dict) - user-data dict (will be rendered to YAML).
               Should include hostname, bootstrap_token, runcmd, and any custom config.
+            image_checksum (string | None) - expected checksum of the downloaded
+              image (e.g. "sha256:<hex>"). If given, verified after download;
+              mismatch raises RuntimeError. If None, no verification is performed.
             ssh_public_keys (list[str] | None) - SSH public keys to inject into guest.
               If None or empty, no SSH key injection is performed.
             disk_resize_gb (int | None) - resize root disk to this size in GB.
-              If None, disk remains template size. Default None.
-            timeout (int) - maximum seconds to wait for provisioning completion
-              (clone, config, start). Default 180.
+              If None, disk remains the downloaded image's native size. Default None.
+            storage (string | None) - name of the storage pool to place the root
+              disk on (a name returned by ``get_image_storages()``). If None,
+              the driver auto-detects the first enabled, node-available storage
+              whose content includes "images".
+            download_timeout (int) - maximum seconds to wait for the image download
+              (skipped entirely if already cached on the hypervisor). Default 300.
+            timeout (int) - maximum seconds to wait for the remaining provisioning
+              steps (VM creation, disk import, config, start). Default 180.
 
         Returns:
             VMProvisionResultDict - ``{"vmid": str, "name": str, "node": str}``
               vmid is the hypervisor-internal VM ID as a string (numeric for Proxmox).
 
         Raises:
-            RuntimeError - if provisioning fails (storage unavailable, invalid
-              config, timeout, etc.)
+            RuntimeError - if provisioning fails (download failure, checksum
+              mismatch, storage unavailable, invalid config, timeout, etc.)
         """
         raise NotImplementedError
 
@@ -674,3 +696,21 @@ class HypervisorDriver(DeviceTypeDriver):
               per-kind rules).
         """
         raise NotImplementedError
+
+    def get_image_storages(self) -> List[StorageTargetDict]:
+        """
+        List the storage pools a new VM's root disk may be placed on, scoped to
+        the specific node the VM will be created on.
+
+        Returns only storages that are actually usable for this purpose right
+        now: content includes "images", administratively enabled, and — for
+        hypervisors where storage can be restricted to a subset of cluster
+        nodes — available on this node specifically. A storage configured
+        cluster-wide but restricted to other nodes must not appear here, since
+        passing its name to ``create_vm_from_cloud_init(storage=...)`` would fail.
+
+        Returns:
+            List[StorageTargetDict] - each entry's ``name`` is directly usable
+              as ``create_vm_from_cloud_init``'s ``storage`` argument.
+        """
+        raise NotImplementedError

napalm_device_types/models.py

@@ -314,6 +314,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
@@ -713,6 +840,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):
@@ -741,9 +869,46 @@ class NetworkTargetDict(TypedDict):
     - 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``.
+      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 or vnet name, usable directly as NICConfigDict.bridge
     kind: str  # "bridge" or "vnet"
     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
+
+
+# ---------------------------------------------------------------------------
+# 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/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

@@ -20,6 +20,7 @@ Usage::
 
 from typing import Dict, List
 from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types.dhcp import DhcpServerMixin
 from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
 from napalm_device_types.models import (
     HealthMetricsDict,
@@ -34,7 +35,7 @@ from napalm_device_types.models import (
 )
 
 
-class ResidentialGatewayDriver(DeviceTypeDriver):
+class ResidentialGatewayDriver(DhcpServerMixin, DeviceTypeDriver):
     TYPE_LABEL: str = "Gateway"
     """
     Abstract intermediate driver for residential gateways (router + firewall + AP).

tests/test_dhcp_diff_apply.py

@@ -0,0 +1,195 @@
+"""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:
+    def test_get_dhcp_reservations_raises_not_implemented_by_default(self):
+        with pytest.raises(NotImplementedError):
+            DhcpServerMixin().get_dhcp_reservations()
+
+    def test_apply_dhcp_reservation_raises_not_implemented_by_default(self):
+        with pytest.raises(NotImplementedError):
+            DhcpServerMixin().apply_dhcp_reservation(_reservation())
+
+    def test_commit_dhcp_reservations_raises_not_implemented_by_default(self):
+        with pytest.raises(NotImplementedError):
+            DhcpServerMixin().commit_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(
+    "call",
+    [
+        lambda d: d.get_dhcp_subnets(),
+        lambda d: d.apply_dhcp_subnet({}),  # type: ignore[typeddict-item]
+        lambda d: d.commit_dhcp_subnets(),
+    ],
+)
+def test_device_specific_methods_raise_not_implemented(call: Any) -> None:
+    with pytest.raises(NotImplementedError):
+        call(_BareDriver())

tests/test_firewall_diff_apply.py

@@ -0,0 +1,170 @@
+"""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:
+    def test_get_firewall_rules_raises_not_implemented_by_default(self):
+        class _Bare(FirewallDriver):
+            def __init__(self) -> None:
+                pass
+
+        with pytest.raises(NotImplementedError):
+            _Bare().get_firewall_rules()
+
+    def test_apply_firewall_rule_raises_not_implemented_by_default(self):
+        class _Bare(FirewallDriver):
+            def __init__(self) -> None:
+                pass
+
+        with pytest.raises(NotImplementedError):
+            _Bare().apply_firewall_rule(_rule())
+
+    def test_commit_firewall_rules_raises_not_implemented_by_default(self):
+        class _Bare(FirewallDriver):
+            def __init__(self) -> None:
+                pass
+
+        with pytest.raises(NotImplementedError):
+            _Bare().commit_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_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_send_wake_on_lan.py

@@ -0,0 +1,32 @@
+"""Tests for FirewallDriver.send_wake_on_lan default contract."""
+
+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_send_wake_on_lan_raises_not_implemented_by_default():
+    with pytest.raises(NotImplementedError):
+        _BareFirewall().send_wake_on_lan("AA:BB:CC:DD:EE:FF")
+
+
+def test_send_wake_on_lan_accepts_optional_interface():
+    with pytest.raises(NotImplementedError):
+        _BareFirewall().send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
+
+
+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}"}
+
+    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"}