feat: report where an update comes from and whether it is a security fix, refresh the index, read the host status
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).
This commit is contained in:
@@ -0,0 +1,120 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user