feat: read pending updates from the cached firmware status, and run the check on request
CI / test (3.10) (push) Successful in 30s
CI / test (3.11) (push) Successful in 29s
CI / test (3.12) (push) Successful in 30s
CI / test (3.10) (pull_request) Successful in 28s
CI / test (3.11) (pull_request) Successful in 27s
CI / test (3.12) (pull_request) Successful in 29s
CI / test (3.10) (push) Successful in 30s
CI / test (3.11) (push) Successful in 29s
CI / test (3.12) (push) Successful in 30s
CI / test (3.10) (pull_request) Successful in 28s
CI / test (3.11) (pull_request) Successful in 27s
CI / test (3.12) (pull_request) Successful in 29s
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.
This commit is contained in:
@@ -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
|
||||
Reference in New Issue
Block a user