feat: fix routines as templates with the caller's values; deprecate run_device_action (4.13.0)

run_device_action was a string dispatch drivers filled with an
application's remediations, values hard-coded inside (community
"public", the caller's address, a /24 firewall rule).

napalm_device_types.fixes declares them as templates instead. A
template runs declared steps in order and stops at the first failure
(run_fix_steps); the driver implements the steps, or overrides the
whole method where its device takes the change in one call; every
value comes from the caller.

- SnmpAgentMixin.configure_snmp_agent(settings): install, configure,
  firewall, start. The caller verifies with its own SNMP client.
  get_management_peer() (optional) says the address the device sees the
  management connection come from, for choosing allowed_sources.
- EnsurePackageMixin.ensure_package(name): install only what is missing.
- PackageProxyMixin.configure_package_proxy(url): declared.
- ContainerEngineMixin.grant_container_engine_access(user): declared.

None is mixed into a role base, so a capability check stays truthful.

Closes #33

README.md

@@ -91,6 +91,32 @@ is a thin bundle over them — `PackageManagementMixin`, `HealthMetricsMixin`,
 `HostRebootMixin` (`reboot_host`) is mixed into `DeviceTypeDriver` itself, since any
 device may be restartable; like the others it only declares.
 
+**Fixes are templates** (`napalm_device_types.fixes`, since 4.13). A change the
+application wants made on a device runs as fixed steps the driver implements, with values
+the caller supplies: `SnmpAgentMixin.configure_snmp_agent(settings)` (install, configure,
+firewall, start), `EnsurePackageMixin.ensure_package(name)`,
+`PackageProxyMixin.configure_package_proxy(url)` and
+`ContainerEngineMixin.grant_container_engine_access(user)`. A driver whose device takes the
+change in one call overrides the template method. A driver mixes in the fixes its device
+can take; none is on a role base. They replace `run_device_action`, which is deprecated.
+
+**SNMP is the application's** (`docs/STRUCTURE.md`). `HealthMetricsMixin` is deprecated
+since 4.12 and goes in 5.0. A driver does not collect over SNMP; a driver whose device
+publishes its health values under OIDs of its own declares them as data:
+
+```python
+class MySwitchDriver(SwitchDriver):
+    SNMP_HEALTH_PROFILE: ClassVar[SnmpHealthProfile] = {
+        "uptime": {"oids": ["1.3.6.1.2.1.1.3.0"]},
+        "cpu": {"oids": ["<vendor cpu oid>", "<older firmware's>"]},
+        "memory": {"total": {"oids": ["<oid>"]}, "free": {"oids": ["<oid>"]}, "unit": "KiB"},
+    }
+```
+
+A value is `oids` (the first that answers wins) or a `table` folded by `first`, `average`
+or `max`, with an optional `pattern` for an agent that answers with text. What the profile
+does not name, the application reads from the standard UCD-MIB and IF-MIB.
+
 `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
@@ -128,6 +154,17 @@ A mechanism it could not read completely is `None` ("did not look"), a host with
 `[]`. The command is about 4 kB, so a driver sends it on an exec channel
 (`_run_virtual_ips_command(command, privileged=)`), not typed into a shell.
 
+`get_dhcp_ha` (declared on `DhcpServerMixin`, optional) is what a DHCP server's own HA
+configuration says: whether HA is on, which peer this server is, each peer's name, role --
+as the server words it, not a shared vocabulary -- and, where the server has one, the URL
+its HA listener answers on. It is configuration, not state: whether the pair is in hot
+standby right now is not in it. `ConfigSyncMixin` (`get_config_sync`) is where a host
+pushes its configuration (`peer`, None for no one) and for which services, by the names
+`get_services` reports; sync items the driver has no service name for stay visible under
+`unmapped`. A driver mixes it in -- no role promises it. Both getters raise when they could
+not read; neither answers "off" for "could not ask". What the facts mean for a pair (hot
+standby, which side to write to) is the caller's to work out.
+
 A firewall's CARP addresses are a third mechanism, `carp`: a firewall driver reads them from
 its own API and reports them under the reading's `carp` key (`vip_manager` and `keepalived`
 then None). What a CARP state means is protocol knowledge and stays in that driver. The

napalm_device_types/__init__.py

@@ -37,6 +37,7 @@ instead of being restated on every role that happens to need it:
 
 * :class:`~napalm_device_types.channel.CommandChannelMixin`
 * :class:`~napalm_device_types.config_lifecycle.ConfigLifecycleMixin`
+* :class:`~napalm_device_types.config_sync.ConfigSyncMixin`
 * :class:`~napalm_device_types.container_engine.ContainerEngineMixin`
 * :class:`~napalm_device_types.dhcp.DhcpServerMixin`
 * :class:`~napalm_device_types.firewall_rules.FirewallRuleMixin`
@@ -71,6 +72,7 @@ from napalm_device_types.channel import (
     run_on_transport,
 )
 from napalm_device_types.config_lifecycle import ConfigLifecycleMixin
+from napalm_device_types.config_sync import ConfigSyncMixin
 from napalm_device_types.container_engine import (
     ContainerEngineConnection,
     ContainerEngineMixin,
@@ -81,6 +83,12 @@ 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.fixes import (
+    EnsurePackageMixin,
+    PackageProxyMixin,
+    SnmpAgentMixin,
+    run_fix_steps,
+)
 from napalm_device_types.health_metrics import HealthMetricsMixin
 from napalm_device_types.host_reboot import HostRebootMixin
 from napalm_device_types.interface_filter import InterfaceFilterMixin
@@ -97,12 +105,23 @@ from napalm_device_types.management_controller import ManagementControllerDriver
 from napalm_device_types.media import MediaDriver
 from napalm_device_types.models import (
     POWER_ACTIONS,
+    ConfigSyncDict,
+    DhcpHaDict,
+    DhcpHaPeerDict,
     EventLogEntryDict,
     FirmwareDict,
     HardwareInventoryDict,
     ManagementProtocolsDict,
     PowerAction,
     PowerStateDict,
+    FixResultDict,
+    FixStepDict,
+    FixStepOutcomeDict,
+    SnmpAgentSettingsDict,
+    SnmpHealthProfile,
+    SnmpInterfacesSpec,
+    SnmpMemorySpec,
+    SnmpValueSpec,
 )
 from napalm_device_types.nat_vpn import NatVpnMixin
 from napalm_device_types.packages import PackageManagementMixin
@@ -148,6 +167,18 @@ from napalm_device_types.storage import StorageDriver
 from napalm_device_types.switch import SwitchDriver
 
 __all__ = [
+    "EnsurePackageMixin",
+    "FixResultDict",
+    "FixStepDict",
+    "FixStepOutcomeDict",
+    "PackageProxyMixin",
+    "SnmpAgentMixin",
+    "SnmpAgentSettingsDict",
+    "run_fix_steps",
+    "SnmpHealthProfile",
+    "SnmpInterfacesSpec",
+    "SnmpMemorySpec",
+    "SnmpValueSpec",
     "AccessPointDriver",
     "ByteStream",
     "CommandChannelMixin",
@@ -159,6 +190,10 @@ __all__ = [
     "open_stream_on_transport",
     "run_on_transport",
     "ConfigLifecycleMixin",
+    "ConfigSyncDict",
+    "ConfigSyncMixin",
+    "DhcpHaDict",
+    "DhcpHaPeerDict",
     "DeviceTypeDriver",
     "DhcpServerMixin",
     "FingerprintRule",

napalm_device_types/config_sync.py

@@ -0,0 +1,35 @@
+# -*- coding: utf-8 -*-
+"""Configuration sync between the two hosts of an HA pair.
+
+A host that pushes its configuration to a peer (OPNsense's XMLRPC sync, say)
+makes the peer's configuration of the synced services a copy of its own: a
+change written on the receiving side lasts only until the next push. Who pushes
+what is therefore a fact a caller needs before it writes to either host.
+
+Declared under ``if TYPE_CHECKING``: a contract, not a placeholder. A driver
+mixes it in; no role promises it.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from napalm_device_types.models import ConfigSyncDict
+
+
+class ConfigSyncMixin:
+    """Where a host pushes its configuration, and for which of its services."""
+
+    if TYPE_CHECKING:
+
+        def get_config_sync(self) -> ConfigSyncDict:
+            """
+            Returns the peer this host pushes its configuration to (None when it
+            pushes to no one) and the services whose configuration it pushes, by
+            the name ``get_services`` reports. Sync items the driver has no
+            service name for are listed under ``unmapped`` as the host names them.
+
+            :raises Exception: When the setting could not be read -- never
+                ``peer=None`` for "could not ask".
+            """
+            ...

napalm_device_types/container_engine.py

@@ -32,7 +32,7 @@ import shlex
 from typing import Any, Dict, List, Optional, Sequence, TYPE_CHECKING, cast
 
 from napalm_device_types.channel import ByteStream, CommandResult
-from napalm_device_types.models import ContainerEngineDict
+from napalm_device_types.models import ContainerEngineDict, FixResultDict
 
 #: Engines this package knows how to reach, and the API each speaks.
 ENGINES: Dict[str, str] = {"docker": "docker-engine"}
@@ -111,6 +111,12 @@ class ContainerEngineMixin:
 
         def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream: ...
 
+        def grant_container_engine_access(self, user: str) -> FixResultDict:
+            """Let *user* reach the container engine without privilege (the
+            ``docker`` group on most hosts). A fix template: see
+            :mod:`napalm_device_types.fixes`. Declared only; a driver implements it."""
+            ...
+
     def _container_engine_binary(self, engine: str) -> str:
         """The host command that is *engine*'s CLI. Override where it is not on PATH."""
         return engine

napalm_device_types/dhcp.py

@@ -28,6 +28,7 @@ from typing import Any, Dict, Iterator, List, Optional, TYPE_CHECKING
 
 from napalm_device_types._matching import check_desired, index_live
 from napalm_device_types.models import (
+    DhcpHaDict,
     DhcpReservationDiffDict,
     DhcpReservationDict,
     DhcpReservationUpdateDict,
@@ -219,6 +220,20 @@ class DhcpServerMixin:
             """
             ...
 
+        def get_dhcp_ha(self) -> DhcpHaDict:
+            """
+            Returns what the server's own HA configuration declares: whether HA
+            is switched on, which peer this server is, and each peer with its
+            role as the server words it.
+
+            Optional: a driver without it has no ``get_dhcp_ha`` at all, and a
+            caller asks ``callable(getattr(driver, "get_dhcp_ha", None))``.
+
+            :raises Exception: When the configuration could not be read -- never
+                ``enabled=False`` for "could not ask".
+            """
+            ...
+
     # ------------------------------------------------------------------
     # Generic, vendor-neutral algorithms.
     # ------------------------------------------------------------------

napalm_device_types/fixes.py

@@ -0,0 +1,151 @@
+# -*- coding: utf-8 -*-
+"""Fix routines: changes an application wants made on a device, as templates.
+
+A template fixes the order of the steps; the driver knows how each step is done on
+its device; the caller supplies every value -- the community, the allowed sources,
+the proxy URL, the user. A driver whose device takes the whole change in one call
+(a REST API) overrides the template method instead of its steps.
+
+None of these is mixed into a role base. A driver mixes in the fixes its device can
+take, so ``callable(getattr(driver, "configure_snmp_agent", None))`` stays a truthful
+answer; a driver that inherits one it cannot run sets it to ``None``.
+
+They replace ``run_device_action`` (docs/STRUCTURE.md):
+
+==========================  ===================================================
+``fix_snmp``                :meth:`SnmpAgentMixin.configure_snmp_agent`
+``fix_apt_proxy``           :meth:`PackageProxyMixin.configure_package_proxy`
+``fix_docker_permissions``  ``ContainerEngineMixin.grant_container_engine_access``
+``install_lldpd`` and co.   :meth:`EnsurePackageMixin.ensure_package`
+==========================  ===================================================
+"""
+
+from __future__ import annotations
+
+from typing import Any, Callable, List, Optional, Sequence, Tuple, TYPE_CHECKING
+
+from napalm_device_types.models import (
+    FixResultDict,
+    FixStepDict,
+    FixStepOutcomeDict,
+    PackageDict,
+    SnmpAgentSettingsDict,
+)
+
+#: One step of a fix: its name and the call that does it.
+FixStep = Tuple[str, Callable[[], FixStepOutcomeDict]]
+
+
+def run_fix_steps(steps: Sequence[FixStep]) -> FixResultDict:
+    """Run *steps* in order and stop at the first that fails.
+
+    A step that raises has failed, with the exception's message as its output.
+    """
+    done: List[FixStepDict] = []
+    for name, step in steps:
+        try:
+            outcome = step()
+            ok, output = bool(outcome["ok"]), str(outcome.get("output", ""))
+        except Exception as exc:  # noqa: BLE001 - any failure of a step ends the fix
+            ok, output = False, str(exc)
+        done.append({"step": name, "ok": ok, "output": output})
+        if not ok:
+            break
+    return {
+        "success": bool(done) and all(s["ok"] for s in done) and len(done) == len(steps),
+        "steps": done,
+        "output": "\n".join(f"[{s['step']}] {s['output']}".rstrip() for s in done),
+    }
+
+
+class SnmpAgentMixin:
+    """Run an SNMP agent on the device that answers the caller.
+
+    The steps: install the agent, write its configuration, let the allowed sources
+    through the host's firewall, start it. Whether it answers is the caller's to ask,
+    with its own SNMP client.
+    """
+
+    def configure_snmp_agent(self, settings: SnmpAgentSettingsDict) -> FixResultDict:
+        """Install, configure, open and start the agent with *settings*."""
+        return run_fix_steps(
+            [
+                ("install", self._install_snmp_agent),
+                ("configure", lambda: self._write_snmp_agent_config(settings)),
+                ("firewall", lambda: self._allow_snmp_sources(settings)),
+                ("start", self._start_snmp_agent),
+            ]
+        )
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def get_management_peer(self) -> Optional[str]:
+            """The address the device sees this management connection come from
+            (for SSH, the client side of ``SSH_CONNECTION``), or None when it cannot
+            tell. A fact for the caller choosing ``allowed_sources``; optional --
+            a driver that cannot tell does not implement it."""
+            ...
+
+        # The steps a driver implements:
+
+        def _install_snmp_agent(self) -> FixStepOutcomeDict: ...
+
+        def _write_snmp_agent_config(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict: ...
+
+        def _allow_snmp_sources(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict:
+            """Let ``settings["allowed_sources"]`` reach the agent; ``ok`` with nothing
+            to do where the host has no firewall of its own."""
+            ...
+
+        def _start_snmp_agent(self) -> FixStepOutcomeDict:
+            """(Re)start the agent so it reads the new configuration, and enable it at boot."""
+            ...
+
+
+class EnsurePackageMixin:
+    """Make sure a package is installed, installing it only where it is missing.
+
+    Works over :class:`~napalm_device_types.packages.PackageManagementMixin`'s
+    ``get_packages`` and ``install_package``, which the driver implements.
+    """
+
+    def ensure_package(self, name: str) -> FixResultDict:
+        """Install *name* unless it is installed already."""
+
+        def install() -> FixStepOutcomeDict:
+            if any(p.get("name") == name and p.get("installed") for p in self.get_packages()):
+                return {"ok": True, "output": f"{name} is already installed"}
+            result: Any = self.install_package(name)  # type: ignore[func-returns-value]
+            # The contract says None (and raise on failure); some drivers still
+            # return {"success", "output"}.
+            if isinstance(result, dict):
+                return {"ok": bool(result.get("success")), "output": str(result.get("output", ""))}
+            return {"ok": True, "output": f"installed {name}"}
+
+        return run_fix_steps([("install", install)])
+
+    if TYPE_CHECKING:  # pragma: no cover
+
+        def get_packages(self) -> List[PackageDict]: ...
+
+        def install_package(self, name: str, version: str = "") -> None: ...
+
+
+class PackageProxyMixin:
+    """Point the device's package manager at a caching proxy."""
+
+    if TYPE_CHECKING:  # pragma: no cover
+
+        def configure_package_proxy(self, url: str) -> FixResultDict:
+            """Have the package manager fetch through *url*; an empty *url* removes it."""
+            ...
+
+
+__all__ = [
+    "EnsurePackageMixin",
+    "FixStep",
+    "PackageProxyMixin",
+    "SnmpAgentMixin",
+    "run_fix_steps",
+]
+

napalm_device_types/health_metrics.py

@@ -1,6 +1,11 @@
 # -*- coding: utf-8 -*-
 """SNMP health collection, shared by every device type that speaks UCD-MIB.
 
+**Deprecated** (docs/STRUCTURE.md): drivers do not collect over SNMP. The
+application polls; a driver whose device publishes values under its own OIDs
+declares them as ``SNMP_HEALTH_PROFILE`` (:class:`~napalm_device_types.models.SnmpHealthProfile`).
+This mixin goes in 5.0.
+
 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
@@ -15,6 +20,7 @@ implement ``get_health_metrics`` themselves.
 from __future__ import annotations
 
 import re
+import warnings
 from typing import Any, Awaitable, Callable, ClassVar, Dict, cast
 
 from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
@@ -39,7 +45,17 @@ class HealthMetricsMixin:
         snmp_get: Callable[..., Awaitable[Any]],
         snmp_walk: Callable[..., Awaitable[Dict[str, Any]]],
     ) -> HealthMetricsDict:
-        """Collect SNMP health metrics (CPU, memory, uptime, load, interfaces)."""
+        """Collect SNMP health metrics (CPU, memory, uptime, load, interfaces).
+
+        Deprecated: the application polls; see the module docstring.
+        """
+        warnings.warn(
+            "HealthMetricsMixin.get_health_metrics is deprecated and goes in 5.0: "
+            "the application polls over SNMP; a driver declares only its OIDs, "
+            "as SNMP_HEALTH_PROFILE",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         # collect_ucd_metrics is untyped shared plumbing; the shape it returns is
         # the contract HealthMetricsDict describes.
         return cast(

napalm_device_types/models.py

@@ -600,6 +600,48 @@ class DhcpSubnetDiffDict(TypedDict):
     update: List[DhcpSubnetUpdateDict]
 
 
+class DhcpHaPeerDict(TypedDict):
+    """One server of a DHCP HA relationship, as the server's configuration names it."""
+
+    name: str
+    #: The peer's role as the server words it ("primary", "standby", ...) -- data,
+    #: not a vocabulary every server shares.
+    role: str
+    #: Where the peer's HA listener answers, when the server addresses it by URL.
+    url: NotRequired[str]
+
+
+class DhcpHaDict(TypedDict):
+    """What a DHCP server's own HA configuration declares.
+
+    Configured, not observed: whether the pair is in hot standby right now, or
+    one side has taken over, is the server's state and not in here.
+    """
+
+    #: The name of the service this configuration belongs to, as ``get_services``
+    #: reports it -- the caller matches it to that service.
+    service: NotRequired[str]
+    enabled: bool
+    #: Which of ``peers`` is this server.
+    this_server: NotRequired[str]
+    peers: List[DhcpHaPeerDict]
+
+
+class ConfigSyncDict(TypedDict):
+    """Where a host pushes its configuration, and whose.
+
+    ``peer`` None: the host pushes to no one (the receiving side of a pair, or a
+    host without the feature switched on).
+    """
+
+    peer: Optional[str]
+    #: The services whose configuration it pushes, by the name ``get_services``
+    #: reports.
+    services: List[str]
+    #: Sync items the driver has no service name for, as the host names them.
+    unmapped: List[str]
+
+
 class VPNTunnelDict(TypedDict):
     type: str
     local_endpoint: str
@@ -951,6 +993,86 @@ class HealthMetricsDict(TypedDict):
     interfaces: NotRequired[Dict[str, HealthMetricsIfaceDict]]
 
 
+class FixStepOutcomeDict(TypedDict):
+    """What one step of a fix routine reports (see ``napalm_device_types.fixes``)."""
+
+    ok: bool
+    output: str
+
+
+class FixStepDict(TypedDict):
+    """One step of a fix routine, as ``FixResultDict`` lists it."""
+
+    step: str
+    ok: bool
+    output: str
+
+
+class FixResultDict(TypedDict):
+    """The result of a fix routine: every step that ran, up to the first failure."""
+
+    success: bool
+    steps: List[FixStepDict]
+    output: str  # the steps' output, one "[step] output" line each
+
+
+class SnmpAgentSettingsDict(TypedDict):
+    """What the caller wants an SNMP agent to be (``SnmpAgentMixin``)."""
+
+    community: str  # the read-only v2c community
+    allowed_sources: List[str]  # addresses or CIDRs let through to the agent; [] changes no firewall
+    contact: NotRequired[str]  # sysContact
+    location: NotRequired[str]  # sysLocation
+    listen: NotRequired[str]  # the address the agent binds to; absent: every address
+
+
+class SnmpValueSpec(TypedDict, total=False):
+    """Where an SNMP agent publishes one value, inside a ``SnmpHealthProfile``.
+
+    Either ``oids``, scalars asked in order (the first non-empty answer wins), or
+    ``table``, a column walked and folded to one value by ``fold``. ``pattern`` is a
+    regular expression whose first group is the number, for an agent that answers
+    with text.
+    """
+
+    oids: List[str]
+    table: str
+    fold: Literal["first", "average", "max"]
+    pattern: str
+
+
+class SnmpMemorySpec(TypedDict, total=False):
+    """Memory: ``total`` with ``used`` or ``free``, or ``percent`` alone."""
+
+    total: SnmpValueSpec
+    used: SnmpValueSpec
+    free: SnmpValueSpec
+    percent: SnmpValueSpec
+    unit: Literal["KiB", "bytes"]  # of total, used and free; KiB, as UCD-MIB counts, by default
+
+
+class SnmpInterfacesSpec(TypedDict, total=False):
+    """How the IF-MIB counters of a device are to be read."""
+
+    skip: str  # a pattern of interface names whose counters are noise, not signal
+    tx_errors_are_drops: bool  # the device counts dropped packets as TX errors (OPNsense)
+
+
+class SnmpHealthProfile(TypedDict, total=False):
+    """Where a device's SNMP agent publishes its health values.
+
+    A driver declares it as ``SNMP_HEALTH_PROFILE`` when its device does not answer
+    the standard UCD-MIB and IF-MIB; the application polls and computes. What a
+    profile does not name is read the standard way. Uptime is in TimeTicks, CPU in
+    percent busy.
+    """
+
+    uptime: SnmpValueSpec
+    cpu: SnmpValueSpec
+    memory: SnmpMemorySpec
+    interfaces: SnmpInterfacesSpec
+
+
 class NICConfigDict(TypedDict):
     """Network interface configuration for VM provisioning."""
 

napalm_device_types/os.py

@@ -201,6 +201,11 @@ class OSDriver(UpdateMixin, ServiceControlMixin, PackageManagementMixin, HealthM
             """
             Executes a named administrative action on the device.
 
+            **Deprecated** (docs/STRUCTURE.md): the actions become fix templates
+            in :mod:`napalm_device_types.fixes`, whose values the caller supplies --
+            ``configure_snmp_agent``, ``configure_package_proxy``,
+            ``grant_container_engine_access``, ``ensure_package``. Goes in 5.0.
+
             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.

pyproject.toml

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

tests/test_dhcp_ha_config_sync.py

@@ -0,0 +1,70 @@
+# -*- coding: utf-8 -*-
+"""Two facts an HA pair of servers reports about itself.
+
+`get_dhcp_ha` is what a DHCP server's own HA configuration says: the peers it
+pairs with and each peer's role, as the server words it. `get_config_sync` is
+where a host pushes its configuration and for which of its services. Both are
+declarations; what they mean for the pair -- hot standby, who to write to --
+is the caller's to work out.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+import napalm_device_types as package
+from napalm_device_types import ConfigSyncMixin, DhcpServerMixin, FirewallDriver
+
+
+class TestDeclaredOnly:
+    def test_get_dhcp_ha_is_absent_until_a_driver_implements_it(self):
+        assert not hasattr(DhcpServerMixin, "get_dhcp_ha")
+
+    def test_get_config_sync_is_absent_until_a_driver_implements_it(self):
+        assert not hasattr(ConfigSyncMixin, "get_config_sync")
+
+    def test_config_sync_mixin_owns_nothing_callable(self):
+        own = [
+            name
+            for name, value in vars(ConfigSyncMixin).items()
+            if not name.startswith("__") and inspect.isfunction(value)
+        ]
+        assert own == []
+
+    def test_the_firewall_role_does_not_carry_config_sync(self):
+        """A driver mixes the capability in; the role does not promise it."""
+        assert not issubclass(FirewallDriver, ConfigSyncMixin)
+
+    def test_a_driver_that_implements_them_is_found_by_hasattr(self):
+        class Pair(ConfigSyncMixin, DhcpServerMixin):
+            def get_dhcp_ha(self):
+                return {"enabled": False, "peers": []}
+
+            def get_config_sync(self):
+                return {"peer": None, "services": [], "unmapped": []}
+
+        assert callable(getattr(Pair(), "get_dhcp_ha", None))
+        assert callable(getattr(Pair(), "get_config_sync", None))
+
+
+class TestTypes:
+    def test_are_exported(self):
+        for name in ("ConfigSyncMixin", "ConfigSyncDict", "DhcpHaDict", "DhcpHaPeerDict"):
+            assert name in package.__all__ and hasattr(package, name), name
+
+    def test_a_peer_needs_only_its_name_and_role(self):
+        """A future server may not have a URL to report; Kea does."""
+        from napalm_device_types import DhcpHaPeerDict
+
+        assert DhcpHaPeerDict.__required_keys__ == frozenset({"name", "role"})
+
+    def test_ha_needs_only_whether_it_is_on_and_the_peers(self):
+        from napalm_device_types import DhcpHaDict
+
+        assert DhcpHaDict.__required_keys__ == frozenset({"enabled", "peers"})
+
+    def test_config_sync_reports_what_it_could_not_name(self):
+        """A sync item the driver has no service name for is still a fact."""
+        from napalm_device_types import ConfigSyncDict
+
+        assert ConfigSyncDict.__required_keys__ == frozenset({"peer", "services", "unmapped"})

tests/test_fix_templates.py

@@ -0,0 +1,174 @@
+"""Fix routines are templates (docs/STRUCTURE.md, NetOrk/netork#1105).
+
+The template fixes the order of the steps, the driver knows how each step is done
+on its device, the caller supplies every value. None of them is mixed into a role
+base, so ``callable(getattr(driver, name, None))`` stays a truthful answer.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from napalm_device_types import (
+    AccessPointDriver,
+    EnsurePackageMixin,
+    FirewallDriver,
+    HypervisorDriver,
+    ManagementControllerDriver,
+    MediaDriver,
+    OSDriver,
+    PhoneDriver,
+    ResidentialGatewayDriver,
+    StorageDriver,
+    SwitchDriver,
+    PackageProxyMixin,
+    SnmpAgentMixin,
+    run_fix_steps,
+)
+
+ROLE_BASES = [
+    AccessPointDriver, FirewallDriver, HypervisorDriver, ManagementControllerDriver, MediaDriver,
+    OSDriver, PhoneDriver, ResidentialGatewayDriver, StorageDriver, SwitchDriver,
+]  # fmt: skip
+
+
+def _ok(output: str = ""):
+    return lambda: {"ok": True, "output": output}
+
+
+def _fail(output: str):
+    return lambda: {"ok": False, "output": output}
+
+
+class TestRunFixSteps:
+    def test_every_step_runs_in_order_and_is_reported(self):
+        result = run_fix_steps([("install", _ok("installed")), ("start", _ok("started"))])
+
+        assert result["success"] is True
+        assert result["steps"] == [
+            {"step": "install", "ok": True, "output": "installed"},
+            {"step": "start", "ok": True, "output": "started"},
+        ]
+        assert result["output"] == "[install] installed\n[start] started"
+
+    def test_the_first_failure_stops_the_rest(self):
+        ran = []
+
+        def later():
+            ran.append("later")
+            return {"ok": True, "output": ""}
+
+        result = run_fix_steps([("install", _fail("no package")), ("start", later)])
+
+        assert result["success"] is False
+        assert [s["step"] for s in result["steps"]] == ["install"]
+        assert ran == []
+
+    def test_a_step_that_raises_is_a_failed_step(self):
+        def broken():
+            raise RuntimeError("connection lost")
+
+        result = run_fix_steps([("install", broken)])
+
+        assert result["success"] is False
+        assert result["steps"] == [{"step": "install", "ok": False, "output": "connection lost"}]
+
+
+class _Agent(SnmpAgentMixin):
+    def __init__(self, fail_at: str = ""):
+        self.calls: list = []
+        self.fail_at = fail_at
+
+    def _step(self, name, *args):
+        self.calls.append((name, *args))
+        return {"ok": name != self.fail_at, "output": name}
+
+    def _install_snmp_agent(self):
+        return self._step("install")
+
+    def _write_snmp_agent_config(self, settings):
+        return self._step("configure", settings)
+
+    def _allow_snmp_sources(self, settings):
+        return self._step("firewall", settings)
+
+    def _start_snmp_agent(self):
+        return self._step("start")
+
+
+class TestSnmpAgent:
+    SETTINGS = {"community": "c0mm", "allowed_sources": ["10.0.0.5"], "contact": "ops", "location": "rack 1"}
+
+    def test_install_configure_open_and_start_with_the_caller_s_values(self):
+        agent = _Agent()
+
+        result = agent.configure_snmp_agent(self.SETTINGS)
+
+        assert result["success"] is True
+        assert agent.calls == [
+            ("install",),
+            ("configure", self.SETTINGS),
+            ("firewall", self.SETTINGS),
+            ("start",),
+        ]
+
+    def test_a_failed_install_configures_nothing(self):
+        agent = _Agent(fail_at="install")
+
+        assert agent.configure_snmp_agent(self.SETTINGS)["success"] is False
+        assert agent.calls == [("install",)]
+
+
+class _Packages(EnsurePackageMixin):
+    def __init__(self, installed):
+        self.installed = installed
+        self.asked: list = []
+
+    def get_packages(self):
+        return [{"name": n, "installed": True} for n in self.installed] + [
+            {"name": "lldpd", "installed": False}
+        ]
+
+    def install_package(self, name, version=""):
+        self.asked.append(name)
+
+
+class TestEnsurePackage:
+    def test_installs_what_is_missing(self):
+        driver = _Packages(installed=["curl"])
+
+        result = driver.ensure_package("lldpd")
+
+        assert result["success"] is True
+        assert driver.asked == ["lldpd"]
+
+    def test_leaves_alone_what_is_there(self):
+        driver = _Packages(installed=["lldpd"])
+
+        result = driver.ensure_package("lldpd")
+
+        assert result["success"] is True
+        assert driver.asked == []
+        assert "already installed" in result["output"]
+
+    def test_an_install_that_reports_failure_fails(self):
+        driver = _Packages(installed=[])
+        driver.install_package = lambda name, version="": {"success": False, "output": "E: no candidate"}
+
+        result = driver.ensure_package("lldpd")
+
+        assert result["success"] is False
+        assert "no candidate" in result["output"]
+
+
+@pytest.mark.parametrize("base", ROLE_BASES, ids=lambda b: b.__name__)
+@pytest.mark.parametrize(
+    "method",
+    ["configure_snmp_agent", "ensure_package", "configure_package_proxy", "grant_container_engine_access"],
+)
+def test_no_role_base_offers_a_fix(base, method):
+    assert getattr(base, method, None) is None
+
+
+def test_the_package_proxy_is_declared_only():
+    assert not hasattr(PackageProxyMixin, "configure_package_proxy")

tests/test_snmp_health_profile.py

@@ -0,0 +1,53 @@
+"""SNMP health: drivers declare data, the application polls (docs/STRUCTURE.md).
+
+``SnmpHealthProfile`` is the shape a driver states its device's OIDs in.
+``HealthMetricsMixin``, which collected in the base package, is deprecated.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import ClassVar, get_type_hints
+
+import pytest
+
+from napalm_device_types import (
+    HealthMetricsMixin,
+    SnmpHealthProfile,
+    SnmpInterfacesSpec,
+    SnmpMemorySpec,
+    SnmpValueSpec,
+)
+
+
+def test_a_profile_says_where_each_value_lives():
+    assert set(get_type_hints(SnmpHealthProfile)) == {"uptime", "cpu", "memory", "interfaces"}
+    assert set(get_type_hints(SnmpValueSpec)) == {"oids", "table", "fold", "pattern"}
+    assert set(get_type_hints(SnmpMemorySpec)) == {"total", "used", "free", "percent", "unit"}
+    assert set(get_type_hints(SnmpInterfacesSpec)) == {"skip", "tx_errors_are_drops"}
+
+
+def test_every_key_is_optional():
+    for spec in (SnmpHealthProfile, SnmpValueSpec, SnmpMemorySpec, SnmpInterfacesSpec):
+        assert spec.__required_keys__ == frozenset(), spec.__name__
+
+
+def test_a_driver_declares_its_profile_as_a_class_attribute():
+    class Driver:
+        SNMP_HEALTH_PROFILE: ClassVar[SnmpHealthProfile] = {
+            "cpu": {"oids": ["1.3.6.1.4.1.9.1", "1.3.6.1.4.1.9.2"]},
+            "memory": {"percent": {"table": "1.3.6.1.4.1.9.3", "fold": "first"}},
+        }
+
+    assert Driver.SNMP_HEALTH_PROFILE["cpu"]["oids"][0] == "1.3.6.1.4.1.9.1"
+
+
+def test_the_collector_in_the_base_package_warns_that_it_is_deprecated():
+    async def get(oid):
+        return None
+
+    async def walk(oid):
+        return {}
+
+    with pytest.warns(DeprecationWarning, match="SNMP_HEALTH_PROFILE"):
+        asyncio.run(HealthMetricsMixin.get_health_metrics(get, walk))