napalm-opnsense/tests/unit/test_updates.py

"""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


class TestRebootHost:
    """``reboot_host`` restarts the firewall through its firmware API (netOrk #637)."""

    def test_the_firmware_api_is_asked(self, driver):
        with patch.object(driver, "_post", return_value={"status": "ok"}) as post:
            driver.reboot_host()

        post.assert_called_once_with("/api/core/firmware/reboot")

    def test_a_refusal_is_raised(self, driver):
        with patch.object(driver, "_post", side_effect=ConnectionError("403 Forbidden")):
            with pytest.raises(RuntimeError, match="403"):
                driver.reboot_host()