feat: public methods for the upgrade, Unbound forwards and OSPF (0.4.0)

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.

napalm_opnsense/opnsense.py

@@ -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:
+            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.
 

napalm_opnsense/ospf_mixin.py

@@ -0,0 +1,41 @@
+"""FRR's OSPF state in neutral terms (napalm-device-types ``OspfMixin``).
+
+Reads the os-frr plugin's diagnostics (``/api/quagga/diagnostics/...``); the
+field names are FRR's. Whether an interface without neighbours or without
+authentication is a problem is the caller's to decide.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from napalm_device_types import OspfMixin
+
+_DIAG = "/api/quagga/diagnostics"
+
+
+def _interface(ifc: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "address": ifc.get("ipAddress") or "",
+        "prefix_length": int(ifc.get("ipAddressPrefixlen") or 0),
+        "area": ifc.get("area") or "",
+        "network_type": ifc.get("networkType") or "",
+        "authentication": ifc.get("authentication") or "",
+        "neighbors": int(ifc.get("nbrCount") or 0),
+        "neighbors_adjacent": int(ifc.get("nbrAdjacentCount") or 0),
+        "cost": int(ifc.get("cost") or 0),
+    }
+
+
+class OPNsenseOspfMixin(OspfMixin):
+    """Needs the driver's ``_get``."""
+
+    def get_ospf(self) -> dict[str, Any]:
+        """Router ID, areas and OSPF per interface; empty when OSPF does not run."""
+        overview = self._get(f"{_DIAG}/ospfoverview").get("response") or {}  # type: ignore[attr-defined]
+        interfaces = (self._get(f"{_DIAG}/ospfinterface").get("response") or {}).get("interfaces") or {}  # type: ignore[attr-defined]
+        return {
+            "router_id": overview.get("routerId") or None,
+            "areas": list((overview.get("areas") or {}).keys()),
+            "interfaces": {name: _interface(ifc) for name, ifc in interfaces.items()},
+        }

napalm_opnsense/resolver_mixin.py

@@ -0,0 +1,65 @@
+"""Unbound's domain forwards (napalm-device-types ``DnsForwardMixin``).
+
+OPNsense keeps forwards and DNS-over-TLS servers in one model (``dot``); a
+forward is a row of type ``forward``. Every change is applied with a
+reconfigure before the method returns.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from napalm_device_types import DnsForwardMixin
+
+_SETTINGS = "/api/unbound/settings"
+_RECONFIGURE = "/api/unbound/service/reconfigure"
+
+
+def _domain(name: str) -> str:
+    return name.strip().rstrip(".")
+
+
+class OPNsenseResolverMixin(DnsForwardMixin):
+    """Needs the driver's ``_post``."""
+
+    def _forward_rows(self) -> list[dict[str, Any]]:
+        rows = self._post(f"{_SETTINGS}/searchForward", {}).get("rows") or []  # type: ignore[attr-defined]
+        return [r for r in rows if r.get("type", "forward") == "forward"]
+
+    def _saved(self, result: dict[str, Any], what: str) -> None:
+        if result.get("result") not in ("saved", "deleted", None):
+            raise RuntimeError(f"Unbound refused {what}: {result.get('validations') or result}")
+
+    def get_dns_forwards(self) -> list[dict[str, Any]]:
+        """Every forward Unbound has, enabled or not; DNS-over-TLS servers are not forwards."""
+        return [
+            {
+                "domain": _domain(r.get("domain", "")),
+                "server": r.get("server", ""),
+                "port": int(r.get("port") or 53),
+                "enabled": str(r.get("enabled", "1")) == "1",
+            }
+            for r in self._forward_rows()
+        ]
+
+    def set_dns_forward(self, domain: str, server: str, port: int = 53) -> None:
+        """Forward *domain* to *server*; an existing forward of the domain is changed."""
+        name = _domain(domain)
+        row = {"enabled": "1", "type": "forward", "domain": name, "server": server, "port": str(port)}
+        known = [r for r in self._forward_rows() if _domain(r.get("domain", "")) == name]
+        if known:
+            result = self._post(f"{_SETTINGS}/setForward/{known[0]['uuid']}", {"dot": row})  # type: ignore[attr-defined]
+        else:
+            result = self._post(f"{_SETTINGS}/addForward", {"dot": row})  # type: ignore[attr-defined]
+        self._saved(result or {}, f"the forward of {name}")
+        self._post(_RECONFIGURE)  # type: ignore[attr-defined]
+
+    def remove_dns_forward(self, domain: str) -> None:
+        """Delete every forward of *domain*; nothing to do when there is none."""
+        name = _domain(domain)
+        uuids = [r["uuid"] for r in self._forward_rows() if _domain(r.get("domain", "")) == name and r.get("uuid")]
+        if not uuids:
+            return
+        for uuid in uuids:
+            self._post(f"{_SETTINGS}/delForward/{uuid}")  # type: ignore[attr-defined]
+        self._post(_RECONFIGURE)  # type: ignore[attr-defined]

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-opnsense"
-version = "0.3.0"
+version = "0.4.0"
 description = "NAPALM driver for OPNsense (read-only via REST API)."
 readme = "README.md"
 license = { text = "Apache-2.0" }
@@ -24,7 +24,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0.0",
-    "napalm_device_types>=4.13.0",
+    "napalm_device_types>=4.17.0",
     "requests>=2.28.0",
 ]
 

tests/unit/test_public_contracts.py

@@ -0,0 +1,173 @@
+"""What netOrk reached through ``_get``/``_post``, as public methods (netOrk #1113).
+
+netOrk followed the firmware upgrade, edited Unbound's forwards and read FRR's
+OSPF state through this driver's private REST helpers. The contracts are
+napalm-device-types 4.17's; the OPNsense API calls and FRR's field names stay
+here.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from napalm_opnsense.opnsense import OPNsenseDriver
+
+
+@pytest.fixture
+def driver():
+    with patch("napalm_opnsense.opnsense.requests.Session"):
+        drv = OPNsenseDriver(hostname="fw", username="k", password="s", optional_args={"verify": False})
+        drv.session = MagicMock()
+        yield drv
+
+
+class TestUpdateProgress:
+    @pytest.mark.parametrize(
+        "status, state",
+        [("running", "running"), ("done", "done"), ("error", "failed"), ("", "idle"), (None, "idle")],
+    )
+    def test_the_upgrade_status_as_a_state(self, driver, status, state):
+        st = {"status": status, "log": "***GOT REQUEST TO UPGRADE***\nfetching\n"}
+        with patch.object(driver, "_get", return_value=st) as get:
+            progress = driver.get_update_progress()
+
+        get.assert_called_once_with("/api/core/firmware/upgradestatus")
+        assert progress == {"state": state, "log": "***GOT REQUEST TO UPGRADE***\nfetching\n"}
+
+    def test_the_packages_go_with_the_upgrade(self, driver):
+        """netOrk named the packages it upgrades; ``apply_updates`` dropped them."""
+        with patch.object(driver, "_post", return_value={"status": "ok"}) as post:
+            result = driver.apply_updates(["opnsense", "openssl"])
+
+        post.assert_called_once_with(
+            "/api/core/firmware/upgrade", {"upgrade_packages": ["opnsense", "openssl"]}
+        )
+        assert result["success"] is True
+
+    def test_no_packages_is_the_whole_upgrade(self, driver):
+        with patch.object(driver, "_post", return_value={"status": "ok"}) as post:
+            driver.apply_updates([])
+
+        post.assert_called_once_with("/api/core/firmware/upgrade")
+
+
+ROWS = [
+    {"uuid": "u1", "enabled": "1", "type": "forward", "domain": "lab.example", "server": "10.0.0.53", "port": "53"},
+    {"uuid": "u2", "enabled": "0", "type": "forward", "domain": "old.example", "server": "10.0.0.9", "port": ""},
+    {"uuid": "u3", "enabled": "1", "type": "dot", "domain": "", "server": "9.9.9.9", "port": "853"},
+]
+
+
+class TestDnsForwards:
+    def _post(self, calls):
+        def post(path, data=None):
+            calls.append((path, data))
+            if path.endswith("searchForward"):
+                return {"rows": ROWS}
+            return {"result": "saved", "status": "ok"}
+
+        return post
+
+    def test_the_forwards_without_dns_over_tls(self, driver):
+        calls = []
+        with patch.object(driver, "_post", side_effect=self._post(calls)):
+            forwards = driver.get_dns_forwards()
+
+        assert forwards == [
+            {"domain": "lab.example", "server": "10.0.0.53", "port": 53, "enabled": True},
+            {"domain": "old.example", "server": "10.0.0.9", "port": 53, "enabled": False},
+        ]
+
+    def test_a_new_domain_is_added_and_applied(self, driver):
+        calls = []
+        with patch.object(driver, "_post", side_effect=self._post(calls)):
+            driver.set_dns_forward("new.example.", "10.0.0.53")
+
+        paths = [p for p, _ in calls]
+        assert paths[-2:] == ["/api/unbound/settings/addForward", "/api/unbound/service/reconfigure"]
+        added = calls[-2][1]["dot"]
+        assert added == {"enabled": "1", "type": "forward", "domain": "new.example", "server": "10.0.0.53", "port": "53"}
+
+    def test_a_known_domain_is_changed_not_doubled(self, driver):
+        calls = []
+        with patch.object(driver, "_post", side_effect=self._post(calls)):
+            driver.set_dns_forward("lab.example", "10.0.0.54", port=5353)
+
+        paths = [p for p, _ in calls]
+        assert "/api/unbound/settings/setForward/u1" in paths
+        assert "/api/unbound/settings/addForward" not in paths
+
+    def test_a_refusal_is_raised(self, driver):
+        def post(path, data=None):
+            if path.endswith("searchForward"):
+                return {"rows": []}
+            return {"result": "failed", "validations": {"dot.server": "invalid"}}
+
+        with patch.object(driver, "_post", side_effect=post), pytest.raises(RuntimeError, match="invalid"):
+            driver.set_dns_forward("x.example", "not-an-ip")
+
+    def test_removing_deletes_every_forward_of_the_domain(self, driver):
+        calls = []
+        with patch.object(driver, "_post", side_effect=self._post(calls)):
+            driver.remove_dns_forward("lab.example.")
+
+        paths = [p for p, _ in calls]
+        assert paths[-2:] == ["/api/unbound/settings/delForward/u1", "/api/unbound/service/reconfigure"]
+
+    def test_removing_an_unknown_domain_changes_nothing(self, driver):
+        calls = []
+        with patch.object(driver, "_post", side_effect=self._post(calls)):
+            driver.remove_dns_forward("nothing.example")
+
+        assert [p for p, _ in calls] == ["/api/unbound/settings/searchForward"]
+
+
+OVERVIEW = {"response": {"routerId": "10.255.0.1", "areas": {"0.0.0.0": {}, "0.0.0.1": {}}}}
+INTERFACES = {
+    "response": {
+        "interfaces": {
+            "wg0": {
+                "ipAddress": "10.255.0.1",
+                "ipAddressPrefixlen": 24,
+                "area": "0.0.0.0",
+                "networkType": "POINTOPOINT",
+                "authentication": "MessageDigest",
+                "nbrCount": 2,
+                "nbrAdjacentCount": 1,
+                "cost": 10,
+            }
+        }
+    }
+}
+
+
+class TestOspf:
+    def test_frr_in_neutral_terms(self, driver):
+        def get(path):
+            return OVERVIEW if path.endswith("ospfoverview") else INTERFACES
+
+        with patch.object(driver, "_get", side_effect=get):
+            ospf = driver.get_ospf()
+
+        assert ospf == {
+            "router_id": "10.255.0.1",
+            "areas": ["0.0.0.0", "0.0.0.1"],
+            "interfaces": {
+                "wg0": {
+                    "address": "10.255.0.1",
+                    "prefix_length": 24,
+                    "area": "0.0.0.0",
+                    "network_type": "POINTOPOINT",
+                    "authentication": "MessageDigest",
+                    "neighbors": 2,
+                    "neighbors_adjacent": 1,
+                    "cost": 10,
+                }
+            },
+        }
+
+    def test_no_ospf_running(self, driver):
+        with patch.object(driver, "_get", return_value={"response": {}}):
+            assert driver.get_ospf() == {"router_id": None, "areas": [], "interfaces": {}}