Christian Manivong
02a441ff09
For netOrk MVP 5, on napalm-device-types 2.3.0:
- get_available_updates reads `apt list --upgradable` over the exec path
(APT_UPGRADABLE_COMMAND), so each update carries its suite and security
status; the APT API, which names only "Debian"/"Proxmox", is the fallback
with security unknown. It raises when neither answers instead of returning
[] -- it used to swallow every error.
- refresh_available_updates(): POST nodes/{n}/apt/update.
- HostStatusMixin over the exec path (reboot required, self-patching).
121 lines
4.5 KiB
Python
121 lines
4.5 KiB
Python
"""Pending updates on a Proxmox node: from where, whether they are security fixes,
|
|
and whether the node needs a reboot.
|
|
|
|
The node's APT API names only an Origin ("Debian", "Proxmox"), the same for the
|
|
main and the security archive. ``apt list --upgradable`` over the exec path
|
|
names the suite (``trixie-security``), so that is read first; the API remains
|
|
the fallback, with the security status left unknown. A reader that cannot read
|
|
raises: an empty list would tell netOrk that nothing is pending.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from napalm_device_types import HostStatusMixin
|
|
from napalm_device_types.host_status import HOST_STATUS_COMMAND
|
|
from napalm_device_types.package_updates import APT_UPGRADABLE_COMMAND
|
|
|
|
from napalm_proxmox.driver import ProxmoxDriver
|
|
|
|
APT = (
|
|
"libssl3t64/stable-security 3.5.1-1+deb13u2 amd64 [upgradable from: 3.5.1-1+deb13u1]\n"
|
|
"ceph-common/stable 20.2.4-pve5 amd64 [upgradable from: 20.2.4-pve4]\n"
|
|
)
|
|
API_ENTRY = {
|
|
"Package": "librados2",
|
|
"OldVersion": "20.2.4-pve4",
|
|
"Version": "20.2.4-pve5",
|
|
"Origin": "Proxmox",
|
|
}
|
|
|
|
|
|
def _api_updates(driver, entries=None, error=None):
|
|
api = MagicMock()
|
|
getter = api.nodes.return_value.apt.update.get
|
|
if error:
|
|
getter.side_effect = error
|
|
else:
|
|
getter.return_value = entries or []
|
|
driver._api = api
|
|
return api
|
|
|
|
|
|
class TestAvailableUpdates:
|
|
def test_apt_over_the_exec_path_names_the_suite(self, driver):
|
|
with patch.object(driver, "_exec_ssh_command", return_value=APT + "__APT_RC=0\n") as exec_:
|
|
updates = {u["name"]: u for u in driver.get_available_updates()}
|
|
|
|
exec_.assert_called_once_with(APT_UPGRADABLE_COMMAND)
|
|
assert updates["libssl3t64"]["security"] is True
|
|
assert updates["ceph-common"]["security"] is False
|
|
assert updates["ceph-common"]["origin"] == "stable"
|
|
|
|
def test_the_api_is_the_fallback_with_security_unknown(self, driver):
|
|
_api_updates(driver, [API_ENTRY])
|
|
with patch.object(driver, "_exec_ssh_command", return_value=""):
|
|
updates = driver.get_available_updates()
|
|
|
|
assert updates == [
|
|
{
|
|
"name": "librados2",
|
|
"current_version": "20.2.4-pve4",
|
|
"new_version": "20.2.4-pve5",
|
|
"origin": "Proxmox",
|
|
"security": None,
|
|
}
|
|
]
|
|
|
|
def test_a_failed_apt_falls_back_too(self, driver):
|
|
_api_updates(driver, [API_ENTRY])
|
|
with patch.object(driver, "_exec_ssh_command", return_value="E: lock\n__APT_RC=100\n"):
|
|
assert [u["name"] for u in driver.get_available_updates()] == ["librados2"]
|
|
|
|
def test_nothing_readable_raises_instead_of_reporting_nothing(self, driver):
|
|
_api_updates(driver, error=RuntimeError("API timeout"))
|
|
with patch.object(driver, "_exec_ssh_command", return_value=""):
|
|
with pytest.raises(RuntimeError):
|
|
driver.get_available_updates()
|
|
|
|
def test_nothing_pending_is_an_empty_list(self, driver):
|
|
with patch.object(driver, "_exec_ssh_command", return_value="__APT_RC=0\n"):
|
|
assert driver.get_available_updates() == []
|
|
|
|
|
|
class TestRefresh:
|
|
def test_the_node_refreshes_its_index_through_the_api(self, driver):
|
|
api = _api_updates(driver)
|
|
api.nodes.return_value.apt.update.post.return_value = "UPID:pve1:0001"
|
|
|
|
result = driver.refresh_available_updates()
|
|
|
|
assert result["success"] is True
|
|
api.nodes.return_value.apt.update.post.assert_called_once()
|
|
|
|
def test_a_refused_refresh_says_why(self, driver):
|
|
api = _api_updates(driver)
|
|
api.nodes.return_value.apt.update.post.side_effect = RuntimeError(
|
|
"403 Permission check failed"
|
|
)
|
|
|
|
result = driver.refresh_available_updates()
|
|
|
|
assert result == {"success": False, "output": "403 Permission check failed"}
|
|
|
|
|
|
class TestHostStatus:
|
|
def test_the_node_is_read_over_the_exec_path(self, driver):
|
|
report = (
|
|
"HSTAT_BEGIN\n[kernel]\n7.0.14-19-pve\n[modules]\n7.0.14-19-pve\n7.0.2-6-pve\n"
|
|
"[timers]\napt-daily-upgrade.timer enabled\nHSTAT_END\n"
|
|
)
|
|
with patch.object(driver, "_exec_ssh_command", return_value=report) as exec_:
|
|
status = driver.get_host_status()
|
|
|
|
exec_.assert_called_once_with(HOST_STATUS_COMMAND)
|
|
assert status["reboot_required"] is False
|
|
|
|
def test_the_driver_declares_the_contract(self):
|
|
assert issubclass(ProxmoxDriver, HostStatusMixin)
|