fix: never report "nothing pending" when the package index is gone, and load it on request

OpenWrt keeps its package index in RAM. After a reboot `apk version` only
warns that it cannot open its cached index and exits 0, and `opkg
list-upgradable` prints nothing: both read as "no updates". netOrk keeps
"pending since" per package (MVP 5), so the reader now raises instead:

- apk: on apk's own warning ("opening from cache ... No such file"), or a
  failed exit status.
- opkg: when /var/opkg-lists is empty.
- refresh_available_updates(): apk update / opkg update, success from the
  exit status.
This commit is contained in:
Christian Manivong
2026-10-06 00:20:09 +02:00
parent d45c082355
commit 388118cd1b
2 changed files with 181 additions and 27 deletions
+68 -27
View File
@@ -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