Compare commits

..
Author SHA1 Message Date
christianmanivong c7ea15dbb4 docs: what the drivers need from the SNMP profile, opt-outs, and how getters behave
CI / test (3.10) (push) Successful in 26s
CI / test (3.11) (push) Successful in 24s
CI / test (3.12) (push) Successful in 26s
CI / test (3.13) (push) Successful in 25s
CI / test (3.10) (pull_request) Successful in 25s
CI / test (3.11) (pull_request) Successful in 24s
CI / test (3.12) (pull_request) Successful in 26s
CI / test (3.13) (pull_request) Successful in 25s
The SNMP profile has to express fallback OIDs, tables folded by first,
average or maximum, numbers inside text values, memory as used, free or
percent, and the uptime source. TX errors counting drops becomes
OPNsense's entry instead of every firewall's.

A driver opts out of an inherited template with None, so callers ask
callable(). USES_SSH, FULL_UPGRADE and platform stay as driver facts.
Getters do not write, a failed read raises, a write reports its real
result, and the application uses public driver methods only.
2026-10-14 10:56:17 +02:00
christianmanivong 01fe680456 docs: a structure that keeps device-types vendor- and product-agnostic
CI / test (3.10) (push) Successful in 28s
CI / test (3.11) (push) Successful in 26s
CI / test (3.12) (push) Successful in 27s
CI / test (3.13) (push) Successful in 25s
CI / test (3.10) (pull_request) Successful in 25s
CI / test (3.11) (pull_request) Successful in 24s
CI / test (3.12) (pull_request) Successful in 25s
CI / test (3.13) (pull_request) Successful in 24s
Roles and capabilities declare, templates run declared steps, reconcilers
stay free of I/O. POSIX host readers, the SSH exec transport and the
privilege layer move to a new package, napalm-unix. SNMP polling, ping
sweeps, discovery scoring, guest profiles, display labels and warnings
belong to the application; drivers declare SNMP OIDs as data and report
facts. Fixes become templates whose values come from the caller.

Says in which order this happens and where the open proposals #8-#14 and
#22 go instead.
2026-10-14 10:20:55 +02:00
11 changed files with 252 additions and 568 deletions

No files matched your search

-33
View File
@@ -91,32 +91,6 @@ 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
@@ -154,13 +128,6 @@ 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
+241
View File
@@ -0,0 +1,241 @@
# Structure
Status: agreed plan. Nothing here is implemented yet; the migration at the end says in
which order it will be.
## Principle
**napalm-device-types is vendor-, OS-, protocol- and product-agnostic.** It holds the
contracts every driver implements and the generic logic that works unchanged for every
driver of a role. It knows no shell, no paramiko, no package manager, no container CLI,
no SNMP OID, no cloud-init and no application that uses the drivers.
That rules out three kinds of code that are here today:
- **How a kind of host is read.** Shell commands and their parsers for POSIX hosts move to
a new package, **napalm-unix**, which napalm-linux and napalm-bsd depend on.
- **What an application does with a device.** Polling a device over SNMP, sweeping a range
with pings, scoring discovery evidence, choosing what a provisioned guest installs, and
deciding what counts as a problem belong to the application, not to the drivers.
- **Product naming.** Files, markers and messages a driver leaves on a device carry the
driver's name or none, never the name of the application that called it.
```
napalm-device-types roles, capabilities, templates, models, reconcilers <- every driver
^
napalm-unix POSIX host readers, SSH exec transport, privilege <- napalm-linux, napalm-bsd
^
napalm-linux, napalm-bsd connection, platform getters, platform steps
application SNMP polling, ping sweeps, discovery scoring, guest
profiles, warnings, and the values every fix is given
```
## What stays in napalm-device-types
| Kind | Rule | Modules |
|---|---|---|
| **Role** | Declares only, under `TYPE_CHECKING`. No runtime method. | `access_point`, `switch`, `firewall`, `hypervisor`, `os`, `storage`, `phone`, `media`, `residential_gateway`, `management_controller` |
| **Capability** | Declares only. | `packages`, `services`, `updates`, `nat_vpn`, `mac_acl`, `config_lifecycle`, `interface_filter`, `host_reboot` |
| **Template** | A concrete method that runs a fixed sequence of abstract steps. A driver implements the steps, or overrides the whole method when its device does it in one go. All values come from the caller. | the fix routines below |
| **Contract types** | Interfaces and data shapes other packages implement or fill. | `CommandChannelMixin`, `CommandResult`, `ByteStream`; the declaration of `ContainerEngineMixin`; the reader mixins' public method declarations; the SNMP profile type; `models` |
| **Reconciler** | Concrete and vendor-neutral, without I/O of its own. | `firewall_rules` and `dhcp` diff/apply, `_matching`, `lag` |
| **Introspection** | Concrete, without I/O. | `roles`; `base` (`FingerprintRule`, `PortSpec`, the fingerprint attributes) |
| **Driver facts** | Class attributes that describe the device or how it is reached, not what an application does with it. | `USES_SSH`, `FULL_UPGRADE`, `platform`, `SNMP_HEALTH_PROFILE` |
### Fix routines are templates
A change the application wants made on a device, for example "run an SNMP agent that
answers us", is one template per purpose. The template fixes the order of the steps; the
driver knows how each step is done on its device; the caller supplies every value.
```python
class SnmpAgentMixin:
def configure_snmp_agent(self, settings: SnmpAgentSettingsDict) -> FixResultDict:
steps = [self._install_snmp_agent, self._write_snmp_agent_config,
self._enable_snmp_agent, self._verify_snmp_agent]
... # runs them in order, stops at the first failure
# declared only; napalm-unix implements them for net-snmp
def _install_snmp_agent(self) -> FixStepDict: ...
def _write_snmp_agent_config(self, settings: SnmpAgentSettingsDict) -> FixStepDict: ...
def _enable_snmp_agent(self) -> FixStepDict: ...
def _verify_snmp_agent(self, settings: SnmpAgentSettingsDict) -> FixStepDict: ...
```
napalm-unix implements the four steps for net-snmp, for Linux and the BSDs alike. A REST
driver such as OPNsense overrides `configure_snmp_agent` as a whole, because its device
takes the whole configuration in one call. The community, the allowed sources, contact
and location are the caller's.
The same shape replaces every action in today's `run_device_action`:
| Today | Template |
|---|---|
| `fix_snmp` | `configure_snmp_agent(settings)` |
| `fix_apt_proxy` | `configure_package_proxy(url)` |
| `fix_docker_permissions` | `grant_container_engine_access(user)` |
| `install_lldpd`, `install_auc` | `ensure_package(name)` |
The mixin is mixed in by the driver, or by the platform package, not by a role base. A
driver that inherits a template it cannot run on its device, as a NAS driver built on the
Linux driver inherits the package proxy, opts out by setting the method to `None`. So a
caller asks `callable(getattr(driver, "configure_snmp_agent", None))`, never `hasattr`.
The caller verifies the result from its own side where it can; the SNMP agent is asked
by the application's SNMP client, not by a probe the driver hand-builds.
### SNMP: drivers declare data, the application polls
The drivers do not collect over SNMP. A driver whose device answers the standard UCD-MIB
and IF-MIB declares nothing. A driver whose device publishes the same values under
proprietary OIDs declares them as data, in a class attribute of a type defined here:
```python
class ProcurveDriver(SwitchDriver):
SNMP_HEALTH_PROFILE: ClassVar[SnmpHealthProfile] = {
"uptime": {"oids": ["1.3.6.1.4.1.11.2.14.11.1.2.4.1.5.1"], "format": "ticks"},
"cpu": {"oids": ["1.3.6.1.4.1.11.2.14.11.5.1.9.6.1.0",
"1.3.6.1.4.1.11.2.3.7.8.6.1.0"]},
"memory": {"total": "...", "free": "...", "unit": "bytes"},
}
```
One OID per value is not enough for the drivers there are. The profile type has to say:
- **Fallbacks:** an ordered list of OIDs; the first that answers wins (ProCurve's CPU).
- **Tables:** an OID to walk and how to fold its rows: first, average or maximum (TP-Link
reports CPU and memory per stack unit).
- **Text values:** a pattern that takes the number out of a string (Netgear reports CPU
as `"5 Secs ( 3.2%)"`).
- **Memory:** total with used or free, or a percentage only, and the unit.
- **Uptime:** which OID, standard ones included (`sysUpTime` where a device has no
`hrSystemUptime`), and its format.
- **Interfaces:** a pattern of names to skip, and whether the TX-error counter counts
drops. Today the second is `True` on every firewall, because OPNsense counts drops
there; it becomes OPNsense's profile entry.
Polling, the standard OIDs, the counter arithmetic (64-bit counters first, 32-bit as the
fallback) and the defaults are the application's.
### Warnings are facts
A driver reports facts: the proxy a host is configured with, a DSL line's SNR, the
firmware version it runs and the one it is offered. It does not compare them with a
setting of the application or with a threshold, and it does not name a warning.
The same goes for lookups a device does not make itself: what firmware a vendor
currently publishes is the application's catalogue, not a getter.
### Getters read, and say when they cannot
- A getter changes nothing on the device. It does not start a daemon, rewrite its
configuration or install a package on the way to an answer.
- A read that fails raises. An empty list means "there is none", never "could not ask";
a sync that deletes and re-adds on top of an empty answer would otherwise duplicate
everything.
- A write reports failure from the device's answer or exit status, never `success: True`
by default.
### The application uses public methods only
The application never calls a driver's private methods (`_get`, `_post`,
`_send_command`, `_action_*`). What it needs from a device becomes a declared method of
the driver.
## What moves to napalm-unix
| Module | Knows |
|---|---|
| `host_status` | apt, dnf, dpkg, systemd timers, `df` |
| `kernel` | `/proc`, `/lib/modules`, gzip, base64 |
| `listening` | `ss`, `netstat`, `/proc/<pid>/cgroup` |
| `systemd` | `systemctl` |
| `virtual_ips` | vip-manager, keepalived, awk |
| `system_identity` | SMBIOS via sysfs and `dmidecode` |
| `package_updates` | apt and dnf output |
| `ssh_server` | OpenSSH (`sshd -T`, `sshd -t`, `sshd_config`); which algorithms to drop stays the caller's |
| `terminal` | ANSI escape codes in shell output |
| `channel`: `run_on_transport`, `open_stream_on_transport`, `ParamikoExecStream` | paramiko exec channels |
| `container_engine`: `ContainerEngineConnection`, engine probing | the Docker CLI, `system dial-stdio` |
It also takes what napalm-linux and napalm-bsd each carry today:
- **Privilege:** one `sudo -S` / `sudo -n` / root layer, including the rule that a
privileged command never gets stdin while a sudo password is set, and one way to write a
file as root (stage with `mktemp`, then `install`). Two copies of this have already
drifted apart.
- **Parsers:** `/etc/passwd` with `/etc/group`, crontab lines, `ps` with `lstart`.
- **Fix steps:** net-snmp, the package proxy, container engine group membership.
The FreeBSD part of `host_status` (`freebsd-version -k` against `-r`) moves to napalm-bsd.
In napalm-unix every reader follows one pattern:
- **Shape:** `FOO_COMMAND`, `parse_foo(output)`, and `FooMixin.get_foo()`.
- **Reaching the host:** the mixin uses `run_command(command, privileged=...)` from
`CommandChannelMixin`. The six per-reader `_run_*` hooks, with four different signatures,
go away.
- **Framing:** one frame helper writes and parses the `BEGIN`/`END` markers for every reader.
- **Errors:** one exception for "cannot read" and one for "not on this host". A read that
fails or is cut short raises; it never reads as "nothing there".
- **Names on the host:** temporary files, markers and drop-ins carry `napalm` or the
driver's name. A drop-in written under an older name is still recognised and replaced.
Layout: `napalm_unix/{posix,linux,freebsd,openssh,transport}/`.
## What moves to the application
| Today | Why it is not a driver's |
|---|---|
| `health_metrics`, `_ucd_metrics` | Polling over SNMP is something the application does to a device; the driver is not even connected while it runs. |
| `ping_sweep` | Looping over addresses, capping the targets and reporting progress is a scan. A device with a faster batch keeps it as a capability of its own. |
| `rules_max_weight` | Scoring is how one application ranks drivers. The rules themselves stay with the driver. |
| `provisioning` (`QEMU_GUEST_AGENTS`, cloud-init network-config, decompressors) | Which agent a guest gets and how its network is configured is a guest profile. The hypervisor driver is handed the packages and commands. |
| `TYPE_LABEL`, `REBOOT_SETTLE_SECONDS` | A display name and a polling delay. |
| firmware catalogues (napalm-netgear's `netgear_plus`) | What a vendor publishes is not something the device reports. |
| `run_device_action` | Replaced by the templates above. |
`OSDriver.ROLE` stops defaulting to `"linux"`. Each OS driver states its own role.
## Rules, enforced by tests in this package
1. Roles and capabilities own no runtime method; a template runs only declared steps.
2. No module imports `paramiko`, `socket`, `subprocess` or `asyncio`, and none defines a
`*_COMMAND`.
3. No module, string or default names an application.
4. Role methods return TypedDicts, not `Any`.
Each rule is added with the step that makes it pass.
## Migration
This package cannot keep re-exports pointing at napalm-unix: the contract package would
then depend on its implementation. So a moved name is first deprecated here and then
removed, and nothing breaks in between.
| Step | Change |
|---|---|
| 1 | The application moves to 4.9 and the drivers that need it; napalm-bsd stops passing stdin to a privileged command while a sudo password is set. |
| 2 | This document; the application's architecture docs and this README follow it. |
| 3 | `SnmpHealthProfile`. The application polls on its own; the four switch and phone drivers declare their OIDs as data. `HealthMetricsMixin` is deprecated. |
| 4 | The fix templates. napalm-linux and napalm-bsd implement the steps; the application calls the templates with its values. `run_device_action` is deprecated. The other drivers follow, one issue each. |
| 5 | Ping sweep, scoring, provisioning, `TYPE_LABEL` and `REBOOT_SETTLE_SECONDS` move to the application and are deprecated here. Drivers report facts instead of warnings. |
| 6 | napalm-unix, with the moved code and its tests, the unified hooks and the frame helper. napalm-linux and napalm-bsd import from it; the moved names are deprecated here. |
| 7 | 5.0 removes everything deprecated. It also splits `models.py` by area, types the remaining `Dict[str, Any]` role returns, declares what drivers already offer undeclared (`get_networks`, `search_packages`, `first_boot_pending`, the base-system update entry), and makes `install_package` declare the result drivers return. |
| 8 | The README shrinks to the concepts; the reader details move with the readers. |
## Proposals this replaces
Several open issues proposed adding shared driver code here. Under this structure:
| Issue | Goes to |
|---|---|
| #8 netmiko session | not here: a CLI-driver helper package, or left in the drivers |
| #9 config mode for CLI switches | not here: a CLI-driver helper package, or left in the drivers |
| #10 LLDP rows | the projection of rows into NAPALM's two shapes stays here; the `lldpctl` parser goes to napalm-unix |
| #11 CLI text helpers | not here; the natural port sort, which `lag` already has, may become public |
| #12 ping | the reply builder stays here; the BusyBox and iputils parser goes to napalm-unix |
| #13 Linux userland | napalm-unix; the snmpd part becomes the SNMP template's steps |
| #14 `run_device_action` | replaced by the fix templates |
| #22 hooks onto `run_command` | step 6 |
-26
View File
@@ -81,12 +81,6 @@ 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
@@ -109,14 +103,6 @@ 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
@@ -162,18 +148,6 @@ 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",
+1 -7
View File
@@ -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, FixResultDict
from napalm_device_types.models import ContainerEngineDict
#: Engines this package knows how to reach, and the API each speaks.
ENGINES: Dict[str, str] = {"docker": "docker-engine"}
@@ -111,12 +111,6 @@ 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
-151
View File
@@ -1,151 +0,0 @@
# -*- 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",
]
+1 -17
View File
@@ -1,11 +1,6 @@
# -*- 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
@@ -20,7 +15,6 @@ 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
@@ -45,17 +39,7 @@ class HealthMetricsMixin:
snmp_get: Callable[..., Awaitable[Any]],
snmp_walk: Callable[..., Awaitable[Dict[str, Any]]],
) -> HealthMetricsDict:
"""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 SNMP health metrics (CPU, memory, uptime, load, interfaces)."""
# collect_ucd_metrics is untyped shared plumbing; the shape it returns is
# the contract HealthMetricsDict describes.
return cast(
+8 -101
View File
@@ -404,42 +404,29 @@ 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 -- 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.
its interface addresses, not here. Only these fields leave the host -- never
a password, an etcd endpoint or a notify script.
"""
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
mechanism: str # "vip-manager" | "keepalived"
instance: str # the vrrp_instance name, or the vip-manager unit without ".service"
address: str # normalised: "10.7.224.10", "fd00::10"
prefix_length: Optional[int] # None where the configuration gives none
interface: Optional[str]
#: 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]]
group_key: Optional[str] # vip-manager's trigger-key, or "vrid:<n>" for keepalived
running: Optional[bool] # the unit is active
class VirtualIpsDict(TypedDict):
"""What a driver's ``get_virtual_ips`` read, per mechanism.
"""What ``VirtualIpsMixin.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 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``.
follow, a host without systemd. A consumer keeps what it knew for a None.
"""
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):
@@ -951,86 +938,6 @@ 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."""
-5
View File
@@ -201,11 +201,6 @@ 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.
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "napalm-device-types"
version = "4.12.0"
version = "4.9.0"
description = "Abstract device-type base classes for NAPALM drivers"
readme = "README.md"
requires-python = ">=3.10"
-174
View File
@@ -1,174 +0,0 @@
"""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")
-53
View File
@@ -1,53 +0,0 @@
"""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))