Compare commits
3
Commits
c8fc46c373
...
799d1ce749
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
799d1ce749 | ||
|
|
ce40299033 | ||
|
|
27027eec56 |
@@ -126,6 +126,9 @@ class LinuxDriver(OSDriver):
|
|||||||
TYPE_LABEL = "Linux"
|
TYPE_LABEL = "Linux"
|
||||||
VENDOR = "Linux"
|
VENDOR = "Linux"
|
||||||
DRIVER_NAME = "linux"
|
DRIVER_NAME = "linux"
|
||||||
|
# A general-purpose host runs through a full init sequence; NAS derivatives
|
||||||
|
# (OpenMediaVault, QNAP) inherit this and are, if anything, slower.
|
||||||
|
REBOOT_SETTLE_SECONDS = 90
|
||||||
SNMP_FINGERPRINT = [
|
SNMP_FINGERPRINT = [
|
||||||
FingerprintRule("linux", weight=5.0),
|
FingerprintRule("linux", weight=5.0),
|
||||||
]
|
]
|
||||||
@@ -1019,7 +1022,7 @@ class LinuxDriver(OSDriver):
|
|||||||
success = not any(kw in low for kw in ("error:", "failed", "not found", "is not installed", "no packages"))
|
success = not any(kw in low for kw in ("error:", "failed", "not found", "is not installed", "no packages"))
|
||||||
return {"success": success, "output": raw.strip()}
|
return {"success": success, "output": raw.strip()}
|
||||||
|
|
||||||
def get_pending_updates(self) -> list[UpdateDict]:
|
def get_available_updates(self) -> list[UpdateDict]:
|
||||||
if self._pkg_manager == "apt":
|
if self._pkg_manager == "apt":
|
||||||
return self._get_updates_apt()
|
return self._get_updates_apt()
|
||||||
if self._pkg_manager in ("dnf", "yum"):
|
if self._pkg_manager in ("dnf", "yum"):
|
||||||
@@ -1032,10 +1035,6 @@ class LinuxDriver(OSDriver):
|
|||||||
f"Package manager '{self._pkg_manager}' is not supported"
|
f"Package manager '{self._pkg_manager}' is not supported"
|
||||||
)
|
)
|
||||||
|
|
||||||
def get_available_updates(self) -> list[UpdateDict]:
|
|
||||||
"""Alias for get_pending_updates(); called by the netork API backend."""
|
|
||||||
return self.get_pending_updates()
|
|
||||||
|
|
||||||
def get_device_warnings(self) -> List[dict[str, Any]]:
|
def get_device_warnings(self) -> List[dict[str, Any]]:
|
||||||
"""Return warning dicts for issues detected on this device.
|
"""Return warning dicts for issues detected on this device.
|
||||||
|
|
||||||
|
|||||||
+13
-5
@@ -22,6 +22,11 @@ def driver():
|
|||||||
d._secret = "pass" # noqa: S105
|
d._secret = "pass" # noqa: S105
|
||||||
d._forced_pkg_manager = None
|
d._forced_pkg_manager = None
|
||||||
d._pkg_manager = "apt"
|
d._pkg_manager = "apt"
|
||||||
|
# Set by __init__, which this fixture bypasses via __new__. Without it every
|
||||||
|
# call through _sudo() raises AttributeError, which the callers' broad
|
||||||
|
# `except Exception` turns into a plain {"success": False} -- so the tests
|
||||||
|
# failed for a reason that had nothing to do with what they were testing.
|
||||||
|
d._sudo_password = None
|
||||||
d.netmiko_optional_args = {}
|
d.netmiko_optional_args = {}
|
||||||
d._device = MagicMock()
|
d._device = MagicMock()
|
||||||
return d
|
return d
|
||||||
@@ -166,7 +171,7 @@ def test_get_packages_apt(driver):
|
|||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# get_pending_updates (apt)
|
# get_available_updates (apt)
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
@@ -177,10 +182,10 @@ APT_UPGRADABLE = (
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_get_pending_updates_apt(driver):
|
def test_get_available_updates_apt(driver):
|
||||||
driver._pkg_manager = "apt"
|
driver._pkg_manager = "apt"
|
||||||
with patch.object(driver, "_send", side_effect=["", APT_UPGRADABLE]):
|
with patch.object(driver, "_send", return_value=APT_UPGRADABLE):
|
||||||
updates = driver.get_pending_updates()
|
updates = driver.get_available_updates()
|
||||||
assert len(updates) == 2
|
assert len(updates) == 2
|
||||||
assert updates[0]["name"] == "openssh-server"
|
assert updates[0]["name"] == "openssh-server"
|
||||||
assert updates[0]["current_version"] == "1:9.2p1-2+deb12u1"
|
assert updates[0]["current_version"] == "1:9.2p1-2+deb12u1"
|
||||||
@@ -312,7 +317,10 @@ def test_apply_updates_apt_all_packages(driver):
|
|||||||
driver._pkg_manager = "apt"
|
driver._pkg_manager = "apt"
|
||||||
sent_commands = []
|
sent_commands = []
|
||||||
|
|
||||||
def capture_send(cmd):
|
def capture_send(cmd, **kwargs):
|
||||||
|
# _sudo() passes read_timeout as a keyword; without **kwargs this raises
|
||||||
|
# TypeError, which the caller's `except Exception` reports as a failed
|
||||||
|
# upgrade rather than a broken test double.
|
||||||
sent_commands.append(cmd)
|
sent_commands.append(cmd)
|
||||||
return APT_UPGRADE_SUCCESS
|
return APT_UPGRADE_SUCCESS
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user