diff --git a/README.md b/README.md index d784b09..e6fb8e3 100644 --- a/README.md +++ b/README.md @@ -91,6 +91,15 @@ 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: diff --git a/napalm_device_types/__init__.py b/napalm_device_types/__init__.py index 60fcabc..7f942a6 100644 --- a/napalm_device_types/__init__.py +++ b/napalm_device_types/__init__.py @@ -83,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 @@ -108,6 +114,10 @@ from napalm_device_types.models import ( ManagementProtocolsDict, PowerAction, PowerStateDict, + FixResultDict, + FixStepDict, + FixStepOutcomeDict, + SnmpAgentSettingsDict, SnmpHealthProfile, SnmpInterfacesSpec, SnmpMemorySpec, @@ -157,6 +167,14 @@ 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", diff --git a/napalm_device_types/container_engine.py b/napalm_device_types/container_engine.py index 265940a..743ab1d 100644 --- a/napalm_device_types/container_engine.py +++ b/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 diff --git a/napalm_device_types/fixes.py b/napalm_device_types/fixes.py new file mode 100644 index 0000000..33c2ea0 --- /dev/null +++ b/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", +] + diff --git a/napalm_device_types/models.py b/napalm_device_types/models.py index 6be7d50..756d806 100644 --- a/napalm_device_types/models.py +++ b/napalm_device_types/models.py @@ -993,6 +993,39 @@ 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``. diff --git a/napalm_device_types/os.py b/napalm_device_types/os.py index 41785d0..27f7287 100644 --- a/napalm_device_types/os.py +++ b/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. diff --git a/pyproject.toml b/pyproject.toml index 2923ea5..ada551f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-device-types" -version = "4.12.0" +version = "4.13.0" description = "Abstract device-type base classes for NAPALM drivers" readme = "README.md" requires-python = ">=3.10" diff --git a/tests/test_fix_templates.py b/tests/test_fix_templates.py new file mode 100644 index 0000000..886a975 --- /dev/null +++ b/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")