feat: fix routines as templates with the caller's values; deprecate run_device_action (4.12.0)
CI / test (3.10) (push) Successful in 29s
CI / test (3.11) (push) Successful in 26s
CI / test (3.12) (push) Successful in 30s
CI / test (3.13) (push) Successful in 30s
CI / test (3.10) (pull_request) Successful in 35s
CI / test (3.11) (pull_request) Successful in 33s
CI / test (3.12) (pull_request) Successful in 35s
CI / test (3.13) (pull_request) Successful in 34s

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.
- 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
This commit is contained in:
christianmanivong committed 2026-10-16 12:06:19 +02:00
1 parent 64cc20ba00
commit 95222377f0
8 files changed
+389 -2

No files matched your search

+174
View File
@@ -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")