diff --git a/napalm_openwrt/system_mixin.py b/napalm_openwrt/system_mixin.py index 2a569e0..6694940 100644 --- a/napalm_openwrt/system_mixin.py +++ b/napalm_openwrt/system_mixin.py @@ -18,6 +18,20 @@ import time as _time from typing import Any + +#: Printed after a command so its exit status survives the shell session. +_RC_MARKER = "__OWRT_RC=" +_RC_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE) +#: What apk prints when its index is not in RAM (after a reboot). +_APK_INDEX_MISSING = re.compile(r"opening from cache .*No such file or directory") + + +def _split_status(raw: str) -> tuple[str, int | None]: + """``(output, exit_status)``; the status is None when the marker never arrived.""" + statuses = _RC_RE.findall(raw or "") + output = _RC_RE.sub("", raw or "").strip() + return output, int(statuses[-1]) if statuses else None + class OpenWrtSystemMixin: """Mixin providing system-level NAPALM getters (environment, NTP, SNMP, users, services, updates, device actions).""" @@ -593,37 +607,64 @@ class OpenWrtSystemMixin: return warnings def get_available_updates(self) -> list[dict[str, Any]]: - """Return list of upgradable packages from the local package manager cache.""" - import re as _re - pm = self._pm_type() - updates: list[dict[str, Any]] = [] + """Return list of upgradable packages from the local package manager cache. - if pm == "apk": - # Output format: "pkgname-current_ver < new_ver" - raw = self._send_command("apk version 2>/dev/null | grep '<'") - for line in raw.splitlines(): - line = line.strip() - m = _re.match(r'^(.+)-(\d\S*)\s+<\s+(\S+)', line) - if m: - updates.append({ - "name": m.group(1), - "current_version": m.group(2), - "new_version": m.group(3), - }) + OpenWrt keeps its package index in RAM. Without it -- after a reboot, + before ``refresh_available_updates`` -- ``apk version`` only warns and + ``opkg list-upgradable`` prints nothing. Both raise here instead of + reporting "nothing pending". + """ + if self._pm_type() == "apk": + updates = self._updates_apk() else: - # opkg output: "pkgname - current_ver - new_ver" - raw = self._send_command("opkg list-upgradable 2>/dev/null") - for line in raw.splitlines(): - parts = [p.strip() for p in line.split(" - ")] - if len(parts) == 3: - updates.append({ - "name": parts[0], - "current_version": parts[1], - "new_version": parts[2], - }) - + updates = self._updates_opkg() return sorted(updates, key=lambda u: u["name"]) + def _updates_apk(self) -> list[dict[str, Any]]: + import re as _re + + output, status = _split_status(self._send_command(f"apk version 2>&1; echo {_RC_MARKER}$?")) + if status != 0: + raise RuntimeError(f"apk version failed (exit {status}): {output[-200:]}") + if _APK_INDEX_MISSING.search(output): + raise RuntimeError("The package index is not loaded (run apk update first)") + updates: list[dict[str, Any]] = [] + for line in output.splitlines(): + # Output format: "pkgname-current_ver < new_ver" + m = _re.match(r"^(.+)-(\d\S*)\s+<\s+(\S+)", line.strip()) + if m: + updates.append({ + "name": m.group(1), + "current_version": m.group(2), + "new_version": m.group(3), + }) + return updates + + def _updates_opkg(self) -> list[dict[str, Any]]: + raw = self._send_command( + '[ -n "$(ls /var/opkg-lists 2>/dev/null)" ] || echo __NO_INDEX__; ' + "opkg list-upgradable 2>/dev/null" + ) + if "__NO_INDEX__" in raw: + raise RuntimeError("The package index is not loaded (run opkg update first)") + updates: list[dict[str, Any]] = [] + for line in raw.splitlines(): + # opkg output: "pkgname - current_ver - new_ver" + parts = [p.strip() for p in line.split(" - ")] + if len(parts) == 3: + updates.append({ + "name": parts[0], + "current_version": parts[1], + "new_version": parts[2], + }) + return updates + + def refresh_available_updates(self) -> dict[str, Any]: + """Load the package index into RAM (``apk update`` / ``opkg update``).""" + cmd = "apk update" if self._pm_type() == "apk" else "opkg update" + output, status = _split_status(self._send_command(f"{cmd} 2>&1; echo {_RC_MARKER}$?")) + return {"success": status == 0, "output": output} + def apply_updates(self, packages: list[str]) -> dict[str, Any]: """Upgrade the given packages using the device's package manager.""" import re as _re diff --git a/tests/unit/test_updates.py b/tests/unit/test_updates.py new file mode 100644 index 0000000..b86b3fb --- /dev/null +++ b/tests/unit/test_updates.py @@ -0,0 +1,113 @@ +"""Pending updates on OpenWrt: never "nothing pending" when the index is gone. + +OpenWrt keeps its package index in RAM. After a reboot ``apk version`` only +warns that it cannot open the cached index and exits 0, and ``opkg +list-upgradable`` prints nothing: both looked like "no updates". netOrk keeps +"pending since" per package, so a reader that cannot read raises instead, and +``refresh_available_updates`` loads the index (netOrk MVP 5). + +The apk warning is the real output of an OpenWrt 25.12.5 access point after a +reboot. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from napalm_openwrt.openwrt import OpenWrtDriver + +APK_INDEX_MISSING = ( + "WARNING: opening from cache https://downloads.openwrt.org/releases/25.12.5/targets/" + "ath79/generic/packages/packages.adb: No such file or directory\n" + "WARNING: opening from cache https://downloads.openwrt.org/releases/25.12.5/packages/" + "mips_24kc/base/packages.adb: No such file or directory\n" + "__OWRT_RC=0\n" +) +APK_PENDING = ( + "Installed: Available:\n" + "dnsmasq-2.90-r4 < 2.90-r5\n" + "hostapd-common-2024.09.15-r2 < 2024.09.15-r3\n" + "__OWRT_RC=0\n" +) +OPKG_PENDING = "dnsmasq - 2.90-4 - 2.90-5\n" + + +@pytest.fixture +def driver(): + with patch("napalm_openwrt.openwrt.ConnectHandler"): + drv = OpenWrtDriver(hostname="192.168.1.1", username="root", password="") + drv.device = MagicMock() + yield drv + + +def _answer(driver, pm: str, reply: str): + """Answer the package-manager probe with *pm* and everything else with *reply*.""" + sent: list = [] + + def send(cmd, **kwargs): + sent.append(cmd) + if cmd.startswith("command -v apk"): + return "/usr/bin/apk" if pm == "apk" else "__no_apk__" + return reply + + driver._send_command = send + return sent + + +class TestAvailableUpdates: + def test_apk_without_its_index_raises_instead_of_reporting_nothing(self, driver): + _answer(driver, "apk", APK_INDEX_MISSING) + + with pytest.raises(RuntimeError, match="index"): + driver.get_available_updates() + + def test_apk_with_its_index_lists_what_is_pending(self, driver): + _answer(driver, "apk", APK_PENDING) + + updates = driver.get_available_updates() + + assert [u["name"] for u in updates] == ["dnsmasq", "hostapd-common"] + assert updates[0]["current_version"] == "2.90-r4" + assert updates[0]["new_version"] == "2.90-r5" + + def test_apk_that_failed_raises(self, driver): + _answer(driver, "apk", "ERROR: unable to lock database\n__OWRT_RC=99\n") + + with pytest.raises(RuntimeError): + driver.get_available_updates() + + def test_opkg_without_its_lists_raises(self, driver): + _answer(driver, "opkg", "__NO_INDEX__\n") + + with pytest.raises(RuntimeError, match="index"): + driver.get_available_updates() + + def test_opkg_with_its_lists(self, driver): + _answer(driver, "opkg", OPKG_PENDING) + + assert driver.get_available_updates() == [ + {"name": "dnsmasq", "current_version": "2.90-4", "new_version": "2.90-5"} + ] + + +class TestRefresh: + def test_apk_loads_its_index(self, driver): + sent = _answer(driver, "apk", "OK: 8912 distinct packages available\n__OWRT_RC=0\n") + + assert driver.refresh_available_updates()["success"] is True + assert sent[-1].startswith("apk update") + + def test_opkg_loads_its_lists(self, driver): + sent = _answer(driver, "opkg", "Updated list of available packages\n__OWRT_RC=0\n") + + assert driver.refresh_available_updates()["success"] is True + assert sent[-1].startswith("opkg update") + + def test_a_failed_refresh_says_why(self, driver): + _answer(driver, "apk", "ERROR: network unreachable\n__OWRT_RC=1\n") + + result = driver.refresh_available_updates() + + assert result == {"success": False, "output": "ERROR: network unreachable"}