feat: public methods for the upgrade, Unbound forwards and OSPF (0.4.0)
CI / test (3.10) (push) Successful in 31s
CI / test (3.11) (push) Successful in 30s
CI / test (3.12) (push) Successful in 32s
CI / test (3.10) (pull_request) Successful in 30s
CI / test (3.11) (pull_request) Successful in 28s
CI / test (3.12) (pull_request) Successful in 39s

netOrk reached these through the private _get/_post (netOrk #1113); they
are napalm-device-types 4.17's contracts now:

- apply_updates passes the packages it is given (it dropped them) and
  get_update_progress reports the upgrade's state and log.
- get_dns_forwards, set_dns_forward (an existing forward of the domain
  is changed, not doubled) and remove_dns_forward, each applied with a
  reconfigure; a refusal is raised.
- get_ospf translates FRR's diagnostics into neutral terms.

Requires napalm-device-types 4.17.0.
This commit is contained in:
Christian Manivong committed 2026-10-18 15:04:03 +02:00
1 parent 7183da24fb
commit 7dccd0d1f8
5 files changed
+307 -11

No files matched your search

+26 -9
View File
@@ -68,6 +68,8 @@ from napalm.base.exceptions import ConnectionException, ConnectionClosedExceptio
from napalm_opnsense.ha import config_sync, dhcp_ha
from napalm_opnsense.listening import parse_socket_statistics, wireguard_listen_ports
from napalm_opnsense.ping_mixin import OPNsensePingMixin
from napalm_opnsense.ospf_mixin import OPNsenseOspfMixin
from napalm_opnsense.resolver_mixin import OPNsenseResolverMixin
from napalm_opnsense.port_forwards import alias_index, port_forwards, wan_interfaces
from napalm_opnsense.virtual_ips import carp_addresses
@@ -89,7 +91,14 @@ def _carp_master(entry: dict[str, Any]) -> bool:
return str(entry.get("status") or "").strip().upper() == "MASTER"
class OPNsenseDriver(OPNsensePingMixin, ConfigSyncMixin, SnmpAgentMixin, FirewallDriver):
class OPNsenseDriver(
OPNsensePingMixin,
ConfigSyncMixin,
OPNsenseResolverMixin,
OPNsenseOspfMixin,
SnmpAgentMixin,
FirewallDriver,
):
"""NAPALM driver for OPNsense (read-only, REST API)."""
VENDOR = "OPNsense"
@@ -2197,22 +2206,30 @@ class OPNsenseDriver(OPNsensePingMixin, ConfigSyncMixin, SnmpAgentMixin, Firewal
}
def apply_updates(self, packages: list[str]) -> dict[str, Any]:
"""Trigger a full firmware upgrade on OPNsense.
"""Start the firmware upgrade of *packages*, or of everything for none.
Note: OPNsense upgrades the entire system at once rather than
individual packages. The ``packages`` parameter is accepted for
API compatibility but is ignored — the full upgrade is always
applied.
Calls ``POST /api/core/firmware/upgrade``.
Returns once the firewall has taken the request; the upgrade runs on
and :meth:`get_update_progress` follows it. Calls
``POST /api/core/firmware/upgrade``.
"""
try:
result = self._post("/api/core/firmware/upgrade")
if packages:
result = self._post("/api/core/firmware/upgrade", {"upgrade_packages": list(packages)})
else:
result = self._post("/api/core/firmware/upgrade")
return {"success": True, "output": str(result)}
except Exception as exc:
logger.warning("Firmware upgrade failed: %s", exc)
return {"success": False, "output": str(exc)}
def get_update_progress(self) -> dict[str, Any]:
"""Where the firmware upgrade stands, and its whole log
(``GET /api/core/firmware/upgradestatus``)."""
st = self._get("/api/core/firmware/upgradestatus")
status = st.get("status") or ""
state = {"running": "running", "done": "done", "": "idle"}.get(status, "failed")
return {"state": state, "log": st.get("log") or ""}
def search_packages(self, query: str) -> list[dict[str, Any]]:
"""Search available OPNsense plugins by name or description.