napalm_opnsense/opnsense.py

@@ -39,11 +39,16 @@ import difflib
 import json
 import logging
 import socket
+import time
 from ipaddress import ip_address, ip_network
 from typing import Any
 
 logger = logging.getLogger(__name__)
 
+#: How long refresh_available_updates waits for the firewall's update check.
+_REFRESH_POLLS = 20
+_REFRESH_INTERVAL = 3
+
 import requests
 from requests.exceptions import RequestException
 
@@ -1998,45 +2003,62 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
         return tunnels
 
     def get_available_updates(self) -> list[dict[str, Any]]:
-        """Return available firmware and package updates.
+        """Return the pending firmware and package updates the firewall last found.
 
-        Triggers an async update-check on OPNsense via
+        Reads the cached ``GET /api/core/firmware/status``; it triggers no check
-        ``POST /api/core/firmware/check``, then polls
+        (that is :meth:`refresh_available_updates`). An empty list means the last
-        ``GET /api/core/firmware/status`` for up to 15 seconds.
+        check found nothing.
-        Returns a list of ``{name, current_version, new_version}`` dicts,
+
-        or an empty list when everything is up to date or the check has
+        :raises RuntimeError: when the firewall never checked or cannot reach its
-        not yet finished.
+            mirror -- never an empty list for "don't know".
         """
-        import time
+        status = self._checked_firmware_status()
-        try:
+        if status.get("status") not in ("update", "upgrade"):
-            self._post("/api/core/firmware/check")
+            return []
-        except Exception as exc:
+        return sorted(
-            logger.debug("Firmware update check trigger failed: %s", exc)
+            (
+                {
+                    "name": u.get("name", ""),
+                    "current_version": u.get("current_version", u.get("version", "")),
+                    "new_version": u.get("new_version", u.get("version", "")),
+                }
+                for u in status.get("upgrade_packages") or status.get("updates") or []
+            ),
+            key=lambda u: u["name"],
+        )
 
-        for _ in range(5):
+    def _checked_firmware_status(self) -> dict[str, Any]:
-            time.sleep(3)
+        status = self._get("/api/core/firmware/status")
-            try:
+        if not status.get("last_check"):
-                status = self._get("/api/core/firmware/status")
+            raise RuntimeError("The firewall has not checked for updates yet")
-                state = status.get("status", "none")
+        for field in ("connection", "repository"):
-                if state in ("update", "upgrade"):
+            if status.get(field, "ok") != "ok":
-                    updates = (
+                raise RuntimeError(f"The firmware {field} is {status.get(field)!r}")
-                        status.get("upgrade_packages")
+        return status
-                        or status.get("updates")
+
-                        or []
+    def refresh_available_updates(self) -> dict[str, Any]:
-                    )
+        """Run the firewall's update check (``firmware/check``) and wait for it."""
-                    return [
+        before = self._get("/api/core/firmware/status").get("last_check")
-                        {
+        self._post("/api/core/firmware/check")
-                            "name": u.get("name", ""),
+        for _ in range(_REFRESH_POLLS):
-                            "current_version": u.get("current_version", u.get("version", "")),
+            time.sleep(_REFRESH_INTERVAL)
-                            "new_version": u.get("new_version", u.get("version", "")),
+            status = self._get("/api/core/firmware/status")
-                        }
+            if status.get("last_check") and status.get("last_check") != before:
-                        for u in updates
+                return {"success": True, "output": status.get("status_msg", "")}
-                    ]
+        return {
-                if state == "latest":
+            "success": False,
-                    return []
+            "output": f"The update check did not finish within {_REFRESH_POLLS * _REFRESH_INTERVAL} s",
-            except Exception as exc:
+        }
-                logger.debug("Firmware status poll failed: %s", exc)
+
-        return []
+    def get_host_status(self) -> dict[str, Any]:
+        """Whether the firewall needs a reboot to finish an update; it does not
+        patch itself as far as netOrk can tell."""
+        pending = self._get("/api/core/firmware/status").get("needs_reboot") == "1"
+        return {
+            "reboot_required": pending,
+            "reboot_reason": "the firmware status reports a pending reboot" if pending else None,
+            "auto_updates": None,
+        }
 
     def get_device_warnings(self) -> list[dict[str, Any]]:
         """Return a list of warning dicts for issues detected on this device.

tests/unit/test_updates.py

@@ -0,0 +1,125 @@
+"""Pending updates on OPNsense, read from the firmware status the firewall caches.
+
+``get_available_updates`` used to trigger ``firmware/check`` and sleep up to
+15 s; a poll could not afford that, and when the check was not done in time it
+reported "no updates". Reading now comes from the cached ``firmware/status``,
+which carries when the firewall last checked. A firewall that never checked, or
+cannot reach its mirror, raises: netOrk keeps "pending since" per package, and
+an empty list would mean nothing is pending. ``refresh_available_updates`` runs
+the check and waits for it (netOrk MVP 5).
+
+The status fields are the real ones of an OPNsense 26.7.5 firewall.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from napalm_opnsense.opnsense import OPNsenseDriver
+
+
+def _status(**over) -> dict:
+    status = {
+        "connection": "ok",
+        "repository": "ok",
+        "last_check": "Mon Oct  5 07:37:38 CEST 2026",
+        "needs_reboot": "0",
+        "upgrade_needs_reboot": "0",
+        "status": "none",
+        "status_msg": "There are no updates available on the selected mirror.",
+        "upgrade_packages": [],
+    }
+    status.update(over)
+    return status
+
+
+PENDING = [
+    {"name": "opnsense", "current_version": "26.7.5", "new_version": "26.7.6", "reason": "upgrade"},
+    {"name": "openssl", "current_version": "3.0.16", "new_version": "3.0.17", "reason": "upgrade"},
+]
+
+
+@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 TestAvailableUpdates:
+    def test_nothing_pending_after_a_check_is_an_empty_list(self, driver):
+        with (
+            patch.object(driver, "_get", return_value=_status()) as get,
+            patch.object(driver, "_post") as post,
+        ):
+            assert driver.get_available_updates() == []
+
+        get.assert_called_once_with("/api/core/firmware/status")
+        post.assert_not_called()
+
+    def test_pending_packages_from_the_cached_status(self, driver):
+        status = _status(status="update", upgrade_packages=PENDING)
+        with patch.object(driver, "_get", return_value=status):
+            updates = driver.get_available_updates()
+
+        assert updates == [
+            {"name": "openssl", "current_version": "3.0.16", "new_version": "3.0.17"},
+            {"name": "opnsense", "current_version": "26.7.5", "new_version": "26.7.6"},
+        ]
+
+    def test_a_firewall_that_never_checked_raises(self, driver):
+        with patch.object(driver, "_get", return_value=_status(last_check="")):
+            with pytest.raises(RuntimeError, match="checked"):
+                driver.get_available_updates()
+
+    @pytest.mark.parametrize("field", ["connection", "repository"])
+    def test_a_mirror_it_cannot_reach_raises(self, driver, field):
+        with patch.object(driver, "_get", return_value=_status(**{field: "error"})):
+            with pytest.raises(RuntimeError):
+                driver.get_available_updates()
+
+
+class TestRefresh:
+    def test_the_check_runs_and_is_waited_for(self, driver):
+        before = _status(last_check="Mon Oct  5 07:37:38 CEST 2026")
+        after = _status(last_check="Tue Oct  6 09:00:01 CEST 2026")
+        with (
+            patch.object(driver, "_get", side_effect=[before, before, after]),
+            patch.object(driver, "_post") as post,
+            patch("napalm_opnsense.opnsense.time.sleep"),
+        ):
+            result = driver.refresh_available_updates()
+
+        post.assert_called_once_with("/api/core/firmware/check")
+        assert result["success"] is True
+
+    def test_a_check_that_does_not_finish_in_time_says_so(self, driver):
+        with (
+            patch.object(driver, "_get", return_value=_status()),
+            patch.object(driver, "_post"),
+            patch("napalm_opnsense.opnsense.time.sleep"),
+        ):
+            result = driver.refresh_available_updates()
+
+        assert result["success"] is False
+
+
+class TestHostStatus:
+    def test_needs_reboot_comes_from_the_firmware_status(self, driver):
+        with patch.object(driver, "_get", return_value=_status(needs_reboot="1")):
+            status = driver.get_host_status()
+
+        assert status == {
+            "reboot_required": True,
+            "reboot_reason": "the firmware status reports a pending reboot",
+            "auto_updates": None,
+        }
+
+    def test_no_pending_reboot(self, driver):
+        with patch.object(driver, "_get", return_value=_status()):
+            assert driver.get_host_status()["reboot_required"] is False