From fdda745388617b29804727ea2e0281c2931060d2 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Tue, 6 Oct 2026 00:20:10 +0200 Subject: [PATCH] feat: read pending updates from the cached firmware status, and run the check on request get_available_updates triggered firmware/check and slept up to 15 s, too long for a poll, and returned [] when the check had not finished. For netOrk MVP 5 it now reads the cached GET /api/core/firmware/status: - [] only when the last check found nothing; it raises when the firewall never checked (no last_check) or its connection/repository is not "ok". - refresh_available_updates(): POST firmware/check, then waits up to 60 s for last_check to change. - get_host_status(): reboot_required from the status' needs_reboot. --- napalm_opnsense/opnsense.py | 94 ++++++++++++++++----------- tests/unit/test_updates.py | 125 ++++++++++++++++++++++++++++++++++++ 2 files changed, 183 insertions(+), 36 deletions(-) create mode 100644 tests/unit/test_updates.py diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index d1200be..c3276cc 100644 --- a/napalm_opnsense/opnsense.py +++ b/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 - ``POST /api/core/firmware/check``, then polls - ``GET /api/core/firmware/status`` for up to 15 seconds. - Returns a list of ``{name, current_version, new_version}`` dicts, - or an empty list when everything is up to date or the check has - not yet finished. + Reads the cached ``GET /api/core/firmware/status``; it triggers no check + (that is :meth:`refresh_available_updates`). An empty list means the last + check found nothing. + + :raises RuntimeError: when the firewall never checked or cannot reach its + mirror -- never an empty list for "don't know". """ - import time - try: - self._post("/api/core/firmware/check") - except Exception as exc: - logger.debug("Firmware update check trigger failed: %s", exc) + status = self._checked_firmware_status() + if status.get("status") not in ("update", "upgrade"): + return [] + return sorted( + ( + { + "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): - time.sleep(3) - try: - status = self._get("/api/core/firmware/status") - state = status.get("status", "none") - if state in ("update", "upgrade"): - updates = ( - status.get("upgrade_packages") - or status.get("updates") - or [] - ) - return [ - { - "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 updates - ] - if state == "latest": - return [] - except Exception as exc: - logger.debug("Firmware status poll failed: %s", exc) - return [] + def _checked_firmware_status(self) -> dict[str, Any]: + status = self._get("/api/core/firmware/status") + if not status.get("last_check"): + raise RuntimeError("The firewall has not checked for updates yet") + for field in ("connection", "repository"): + if status.get(field, "ok") != "ok": + raise RuntimeError(f"The firmware {field} is {status.get(field)!r}") + return status + + def refresh_available_updates(self) -> dict[str, Any]: + """Run the firewall's update check (``firmware/check``) and wait for it.""" + before = self._get("/api/core/firmware/status").get("last_check") + self._post("/api/core/firmware/check") + for _ in range(_REFRESH_POLLS): + time.sleep(_REFRESH_INTERVAL) + status = self._get("/api/core/firmware/status") + if status.get("last_check") and status.get("last_check") != before: + return {"success": True, "output": status.get("status_msg", "")} + return { + "success": False, + "output": f"The update check did not finish within {_REFRESH_POLLS * _REFRESH_INTERVAL} s", + } + + 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. diff --git a/tests/unit/test_updates.py b/tests/unit/test_updates.py new file mode 100644 index 0000000..9d82732 --- /dev/null +++ b/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