feat: name only a missing base64; lldpd, auc and updates are facts (0.3.0)
CI / test (3.10) (push) Successful in 26s
CI / test (3.11) (push) Successful in 23s
CI / test (3.12) (push) Successful in 26s
CI / test (3.10) (pull_request) Successful in 24s
CI / test (3.11) (pull_request) Successful in 22s
CI / test (3.12) (pull_request) Successful in 25s
CI / test (3.10) (push) Successful in 26s
CI / test (3.11) (push) Successful in 23s
CI / test (3.12) (push) Successful in 26s
CI / test (3.10) (pull_request) Successful in 24s
CI / test (3.11) (pull_request) Successful in 22s
CI / test (3.12) (pull_request) Successful in 25s
get_device_warnings no longer names lldpd_not_installed, updates_available or update_notifications_disabled. The packages and get_available_updates are the facts; netOrk derives the warnings with the same reach (netOrk #1111). The old update read swallowed failures, get_available_updates raises them. no_base64 stays: BusyBox may provide base64 as an applet, so no package tells the caller whether it is there.
This commit is contained in:
1 parent
9f43b8c797
commit
34d80bd994
3 files changed
+53
-62
No files matched your search
@@ -391,69 +391,17 @@ class OpenWrtSystemMixin:
|
|||||||
return {"success": True, "output": output}
|
return {"success": True, "output": output}
|
||||||
|
|
||||||
def get_device_warnings(self) -> list[dict[str, Any]]:
|
def get_device_warnings(self) -> list[dict[str, Any]]:
|
||||||
"""Return a list of warning dicts for issues detected on this device.
|
"""Only a missing ``base64`` binary, which BusyBox may lack.
|
||||||
|
|
||||||
Currently detects:
|
It is not a package the caller can look up -- BusyBox may provide it as
|
||||||
- lldpd not installed (LLDP neighbor discovery unavailable)
|
an applet -- so it stays here until the shell commands callers send get
|
||||||
- package updates available (uses local package cache, no network call)
|
a public method. lldpd, auc and the pending updates are facts
|
||||||
- update notifications not configured (auc not installed, opkg only)
|
(``get_packages``, ``get_available_updates``); whether they are a
|
||||||
|
problem is the caller's to decide.
|
||||||
"""
|
"""
|
||||||
warnings: list[dict[str, Any]] = []
|
if self._send_command("command -v base64 2>/dev/null").strip():
|
||||||
|
return []
|
||||||
# 1. LLDP daemon
|
return [{"code": "no_base64"}]
|
||||||
lldpd_path = self._send_command("which lldpd 2>/dev/null").strip()
|
|
||||||
if not lldpd_path:
|
|
||||||
warnings.append({"code": "lldpd_not_installed"})
|
|
||||||
|
|
||||||
pm = self._pm_type()
|
|
||||||
|
|
||||||
# 2. Package updates available (local cache only – no opkg update)
|
|
||||||
def _pkg_name(line: str, pm_type: str) -> str:
|
|
||||||
"""Extract just the package name from an upgradable line.
|
|
||||||
|
|
||||||
apk: 'luci-app-firewall-26.143~abc < 26.151~def' → 'luci-app-firewall'
|
|
||||||
opkg: 'luci-app-firewall - 1.0 - 1.1' → 'luci-app-firewall'
|
|
||||||
"""
|
|
||||||
import re as _re
|
|
||||||
if pm_type == "apk":
|
|
||||||
# Strip trailing ' < ...' then remove the version suffix (-\d...)
|
|
||||||
name_ver = line.split(" ")[0]
|
|
||||||
m = _re.match(r'^(.*?)-\d', name_ver)
|
|
||||||
return m.group(1) if m else name_ver
|
|
||||||
else:
|
|
||||||
return line.split(" - ")[0].strip()
|
|
||||||
|
|
||||||
try:
|
|
||||||
if pm == "apk":
|
|
||||||
raw_upg = self._send_command("apk version 2>/dev/null | grep '<'")
|
|
||||||
else:
|
|
||||||
raw_upg = self._send_command("opkg list-upgradable 2>/dev/null")
|
|
||||||
upgradable = [ln.strip() for ln in raw_upg.splitlines() if ln.strip()]
|
|
||||||
except Exception:
|
|
||||||
upgradable = []
|
|
||||||
|
|
||||||
if upgradable:
|
|
||||||
pkg_names = [_pkg_name(ln, pm) for ln in upgradable]
|
|
||||||
warnings.append({
|
|
||||||
"code": "updates_available",
|
|
||||||
"meta": {
|
|
||||||
"count": len(upgradable),
|
|
||||||
"packages": pkg_names[:10],
|
|
||||||
},
|
|
||||||
})
|
|
||||||
|
|
||||||
# 3. Attended sysupgrade client not installed (opkg systems only)
|
|
||||||
if pm == "opkg":
|
|
||||||
auc_path = self._send_command("which auc 2>/dev/null").strip()
|
|
||||||
if not auc_path:
|
|
||||||
warnings.append({"code": "update_notifications_disabled"})
|
|
||||||
|
|
||||||
# 4. base64 not available — needed for efficient config apply
|
|
||||||
b64_path = self._send_command("command -v base64 2>/dev/null").strip()
|
|
||||||
if not b64_path:
|
|
||||||
warnings.append({"code": "no_base64"})
|
|
||||||
|
|
||||||
return warnings
|
|
||||||
|
|
||||||
def get_available_updates(self) -> list[dict[str, Any]]:
|
def get_available_updates(self) -> list[dict[str, Any]]:
|
||||||
"""Return list of upgradable packages from the local package manager cache.
|
"""Return list of upgradable packages from the local package manager cache.
|
||||||
|
|||||||
+1
-1
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "napalm-openwrt"
|
name = "napalm-openwrt"
|
||||||
version = "0.2.0"
|
version = "0.3.0"
|
||||||
description = "NAPALM driver for OpenWrt routers/access-points"
|
description = "NAPALM driver for OpenWrt routers/access-points"
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
license = { text = "Apache-2.0" }
|
license = { text = "Apache-2.0" }
|
||||||
|
|||||||
@@ -0,0 +1,43 @@
|
|||||||
|
"""What the driver still names itself (netOrk #1111).
|
||||||
|
|
||||||
|
lldpd, auc and the pending updates are facts the caller reads (``get_packages``,
|
||||||
|
``get_available_updates``) and decides on. A missing ``base64`` stays: BusyBox
|
||||||
|
may provide it as an applet, so no package says whether it is there.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from napalm_openwrt import OpenWrtDriver
|
||||||
|
|
||||||
|
|
||||||
|
def _driver(answers: dict[str, str]) -> OpenWrtDriver:
|
||||||
|
d = OpenWrtDriver("ap", "root", "pw")
|
||||||
|
issued: list[str] = []
|
||||||
|
|
||||||
|
def send(command: str) -> str:
|
||||||
|
issued.append(command)
|
||||||
|
return next((out for frag, out in answers.items() if frag in command), "")
|
||||||
|
|
||||||
|
d._send_command = send
|
||||||
|
d.issued = issued
|
||||||
|
return d
|
||||||
|
|
||||||
|
|
||||||
|
def test_nothing_installed_names_only_base64():
|
||||||
|
"""No lldpd, no auc and an opkg system: the caller names those from the packages."""
|
||||||
|
d = _driver({"opkg list-upgradable": "luci - 1.0 - 1.1\n"})
|
||||||
|
|
||||||
|
assert [w["code"] for w in d.get_device_warnings()] == ["no_base64"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_base64_present_names_nothing():
|
||||||
|
d = _driver({"command -v base64": "/usr/bin/base64\n"})
|
||||||
|
|
||||||
|
assert d.get_device_warnings() == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_one_command_per_poll():
|
||||||
|
d = _driver({})
|
||||||
|
d.get_device_warnings()
|
||||||
|
|
||||||
|
assert d.issued == ["command -v base64 2>/dev/null"]
|
||||||
Reference in new issue
Block a user