feat: public contracts for update progress, DNS forwards and OSPF (4.17.0) #46

Merged
christianmanivong merged 1 commits from feat/opnsense-public-contracts into main 2026-10-18 13:06:08 +00:00
7 changed files with 181 additions and 2 deletions

No files matched your search

+12
View File
@@ -90,7 +90,9 @@ from napalm_device_types.fixes import (
run_fix_steps,
)
from napalm_device_types.health_metrics import HealthMetricsMixin
from napalm_device_types.dns_forward import DnsForwardMixin
from napalm_device_types.firmware import FirmwareOfferMixin
from napalm_device_types.ospf import OspfMixin
from napalm_device_types.licenses import LicenseMixin
from napalm_device_types.host_reboot import HostRebootMixin
from napalm_device_types.interface_filter import InterfaceFilterMixin
@@ -117,7 +119,11 @@ from napalm_device_types.models import (
PowerAction,
PowerStateDict,
HypervisorHostDict,
DnsForwardDict,
FirmwareOfferDict,
OspfDict,
OspfInterfaceDict,
UpdateProgressDict,
LicenseDict,
FixResultDict,
FixStepDict,
@@ -173,7 +179,13 @@ from napalm_device_types.switch import SwitchDriver
__all__ = [
"HypervisorHostDict",
"DnsForwardDict",
"DnsForwardMixin",
"FirmwareOfferDict",
"OspfDict",
"OspfInterfaceDict",
"OspfMixin",
"UpdateProgressDict",
"FirmwareOfferMixin",
"LicenseDict",
"LicenseMixin",
+35
View File
@@ -0,0 +1,35 @@
# -*- coding: utf-8 -*-
"""Domains a device's DNS resolver forwards to another server.
An application that makes the device's resolver ask another server for a zone
-- the authority for an internal domain -- edits these. The driver writes them
in its device's terms and applies them; the forwards are the caller's.
Declared only; a driver whose device resolves DNS implements it.
"""
from __future__ import annotations
from typing import List, TYPE_CHECKING
from napalm_device_types.models import DnsForwardDict
class DnsForwardMixin:
if TYPE_CHECKING: # pragma: no cover - declared for type checkers only
def get_dns_forwards(self) -> List[DnsForwardDict]:
"""Every domain the resolver forwards, enabled or not."""
...
def set_dns_forward(self, domain: str, server: str, port: int = 53) -> None:
"""Forward *domain* to *server*: added, or the existing forward of
*domain* changed. Applied before it returns.
:raises RuntimeError: when the device refuses it.
"""
...
def remove_dns_forward(self, domain: str) -> None:
"""Stop forwarding *domain*; nothing to do when it is not forwarded.
Applied before it returns."""
...
+42
View File
@@ -1015,6 +1015,48 @@ class LicenseDict(TypedDict):
withholds: NotRequired[List[str]]
UpdateState = Literal["idle", "running", "done", "failed"]
class UpdateProgressDict(TypedDict):
"""Where an update the device runs by itself stands
(``UpdateMixin.get_update_progress``)."""
state: UpdateState # "idle": none has run since the device started, or none is known
log: str # the device's own log of it, whole; a caller follows it by line count
class DnsForwardDict(TypedDict):
"""A domain the device's resolver forwards to another server
(``DnsForwardMixin.get_dns_forwards``)."""
domain: str # without the trailing dot
server: str
port: int
enabled: bool
class OspfInterfaceDict(TypedDict):
"""OSPF on one interface (``OspfMixin.get_ospf``)."""
address: str
prefix_length: int
area: str
network_type: str # as the routing daemon names it, e.g. "POINTOPOINT"
authentication: str # as the routing daemon names it, e.g. "MessageDigest"; "" for none
neighbors: int
neighbors_adjacent: int # neighbours with a full adjacency
cost: int
class OspfDict(TypedDict):
"""The device's OSPF state (``OspfMixin.get_ospf``)."""
router_id: Optional[str]
areas: List[str]
interfaces: Dict[str, OspfInterfaceDict] # by interface name
class FirmwareOfferDict(TypedDict):
"""The firmware a device runs and the newest its vendor offers
(``FirmwareOfferMixin.get_firmware_offer``). A fact, not a judgement."""
+31
View File
@@ -0,0 +1,31 @@
# -*- coding: utf-8 -*-
"""A device's OSPF state, in neutral terms.
The routing daemon's own output (FRR, Quagga, a vendor's CLI) is the driver's
to translate; whether an interface without neighbours or without
authentication is a problem is the caller's to decide.
Declared only; a driver whose device runs OSPF implements it.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from napalm_device_types.models import OspfDict
class OspfMixin:
if TYPE_CHECKING: # pragma: no cover - declared for type checkers only
def get_ospf(self) -> OspfDict:
"""Router ID, areas and OSPF per interface.
Example::
{"router_id": "10.255.0.1", "areas": ["0.0.0.0"],
"interfaces": {"wg0": {"address": "10.255.0.1", "prefix_length": 24,
"area": "0.0.0.0", "network_type": "POINTOPOINT",
"authentication": "MessageDigest", "neighbors": 1,
"neighbors_adjacent": 1, "cost": 10}}}
"""
...
+13 -1
View File
@@ -15,7 +15,7 @@ from __future__ import annotations
from typing import Any, Dict, List, TYPE_CHECKING
from napalm_device_types.models import ApplyUpdatesResultDict, UpdateDict
from napalm_device_types.models import ApplyUpdatesResultDict, UpdateDict, UpdateProgressDict
class UpdateMixin:
@@ -61,6 +61,18 @@ class UpdateMixin:
"""
...
def get_update_progress(self) -> UpdateProgressDict:
"""
Where an update the device runs by itself stands, for a caller that
follows it: a firewall's firmware upgrade, which :meth:`apply_updates`
only starts. ``state`` is ``"running"`` until it ends in ``"done"``
or ``"failed"``; ``log`` is the device's whole log of it.
Declared only: a driver whose :meth:`apply_updates` returns when the
update is over has nothing to report here.
"""
...
def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict:
"""
Upgrades the given packages to the newest available version.
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "napalm-device-types"
version = "4.16.0"
version = "4.17.0"
description = "Abstract device-type base classes for NAPALM drivers"
readme = "README.md"
requires-python = ">=3.10"
+47
View File
@@ -0,0 +1,47 @@
"""Public contracts for what an application reached through driver internals.
netOrk followed a firewall's upgrade, edited its resolver's forwards and read
its OSPF state through the driver's private REST helpers (NetOrk/netork#1113).
Each becomes a declared method with a neutral shape; the vendor's own format
is translated by the driver.
"""
from __future__ import annotations
from typing import get_type_hints
from napalm_device_types import (
DnsForwardDict,
DnsForwardMixin,
OspfDict,
OspfInterfaceDict,
OspfMixin,
UpdateMixin,
UpdateProgressDict,
)
def test_update_progress_is_a_state_and_its_log():
assert set(get_type_hints(UpdateProgressDict)) == {"state", "log"}
assert not hasattr(UpdateMixin, "get_update_progress")
def test_a_dns_forward_names_the_domain_and_where_it_goes():
assert set(get_type_hints(DnsForwardDict)) == {"domain", "server", "port", "enabled"}
for name in ("get_dns_forwards", "set_dns_forward", "remove_dns_forward"):
assert not hasattr(DnsForwardMixin, name), name
def test_ospf_in_neutral_terms():
assert set(get_type_hints(OspfDict)) == {"router_id", "areas", "interfaces"}
assert set(get_type_hints(OspfInterfaceDict)) == {
"address",
"prefix_length",
"area",
"network_type",
"authentication",
"neighbors",
"neighbors_adjacent",
"cost",
}
assert not hasattr(OspfMixin, "get_ospf")