feat: declare get_management_peer for choosing an SNMP agent's sources

The caller has to name the sources an SNMP agent answers, and cannot
reliably tell its own address as the device sees it (NAT, relays).
SnmpAgentMixin now declares get_management_peer(): the address the
device sees the management connection come from, or None. Optional:
a driver that cannot tell does not implement it.

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.12). 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.11 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,13 @@ 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.
 
+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
+contract only carries the result: unlike the other two, a CARP entry says who holds the
+address (`holds`), because a firewall's interface list need not show it. `holds` None
+leaves the holder to the interfaces.
+
 `SystemIdentityMixin` (`get_system_identity`) reads a host's SMBIOS identity -- system UUID,
 product serial and board serial from `/sys/class/dmi/id` -- which is what ties a server to
 its BMC (`ManagementControllerDriver.get_hardware_inventory` reports the same values from

napalm_device_types/__init__.py

@@ -81,6 +81,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
@@ -103,6 +109,14 @@ from napalm_device_types.models import (
     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 +162,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",

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/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

@@ -404,29 +404,42 @@ class VirtualIpDict(TypedDict):
     """An address the host's HA configuration lets float between machines.
 
     Declared, not observed: whether the host holds the address right now is in
-    its interface addresses, not here. Only these fields leave the host -- never
-    a password, an etcd endpoint or a notify script.
+    its interface addresses, not here -- unless the mechanism says so itself.
+    CARP does (``holds``): a firewall's interface list need not show the
+    address at all, but CARP's state names the MASTER. Only these fields leave
+    the host -- never a password, an etcd endpoint or a notify script.
     """
 
-    mechanism: str  # "vip-manager" | "keepalived"
-    instance: str  # the vrrp_instance name, or the vip-manager unit without ".service"
+    mechanism: str  # "vip-manager" | "keepalived" | "carp"
+    #: the vrrp_instance name, the vip-manager unit without ".service", or the
+    #: CARP address's interface description
+    instance: str
     address: str  # normalised: "10.7.224.10", "fd00::10"
     prefix_length: Optional[int]  # None where the configuration gives none
     interface: Optional[str]
-    group_key: Optional[str]  # vip-manager's trigger-key, or "vrid:<n>" for keepalived
-    running: Optional[bool]  # the unit is active
+    #: vip-manager's trigger-key, "vrid:<n>" for keepalived, "vhid:<n>" for CARP
+    group_key: Optional[str]
+    running: Optional[bool]  # the unit is active; for CARP: in MASTER or BACKUP
+    #: The host holds the address now, by the mechanism's own state (CARP's
+    #: MASTER). Absent or None: the interface addresses tell.
+    holds: NotRequired[Optional[bool]]
 
 
 class VirtualIpsDict(TypedDict):
-    """What ``VirtualIpsMixin.get_virtual_ips`` read, per mechanism.
+    """What a driver's ``get_virtual_ips`` read, per mechanism.
 
     A list is what was looked at -- ``[]`` a host without that mechanism. None is
     "did not look": a configuration it could not read, an include it could not
-    follow, a host without systemd. A consumer keeps what it knew for a None.
+    follow, a host without systemd, a mechanism the driver does not read (a
+    firewall's reading has ``vip_manager`` and ``keepalived`` None). A consumer
+    keeps what it knew for a None, and for a missing ``carp``.
     """
 
     vip_manager: Optional[List[VirtualIpDict]]
     keepalived: Optional[List[VirtualIpDict]]
+    #: A firewall's CARP addresses (not read by ``VirtualIpsMixin``, which reads Linux
+    #: hosts): every address bound to a VHID, the CARP address and IP aliases on it.
+    carp: NotRequired[Optional[List[VirtualIpDict]]]
 
 
 class PortForwardDict(TypedDict):
@@ -938,6 +951,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.9.0"
+version = "4.12.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.10"

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