fix(linux): fix_snmp installs snmpd without refreshing apt cache first

A fresh cloud image's apt cache is stale/effectively empty — installing
snmpd without an apt-get update first could fail outright or hang on
unreachable mirrors, and the install call wasn't guarded, so a timeout
propagated as an opaque unguarded exception instead of a clean failure
result.

Also adds a new apt_update_upgrade device action (apt-get update +
upgrade), used by netork's VM-provisioning bootstrap alongside the
existing fix_apt_proxy action to fully prep a freshly provisioned VM's
apt before installing anything on it.
This commit is contained in:
Christian Manivong
2026-07-08 17:09:19 +02:00
parent 3ca2ed72ed
commit 8436013dcd
2 changed files with 190 additions and 2 deletions
+139
View File
@@ -661,3 +661,142 @@ def test_get_facts_fallback_vendor_when_dmi_empty(driver):
patch.object(driver, "_send", side_effect=["host", "host.local", "Alpine Linux 3.19", "eth0"]):
facts = driver.get_facts()
assert facts["vendor"] == "Linux" # fallback to VENDOR class attribute
# ---------------------------------------------------------------------------
# _action_fix_snmp / _action_apt_update_upgrade
# ---------------------------------------------------------------------------
def _fix_snmp_send_side_effect(command: str, read_timeout: float = 100) -> str:
"""Canned responses covering every _send() call _action_fix_snmp makes."""
if command.startswith("sudo -n true"):
return "" # passwordless sudo works
if command.startswith("command -v apt"):
return "/usr/bin/apt"
if command.startswith("command -v"):
return "" # ufw/iptables not found, other pkg managers not found
if command.startswith("ss -tnp"):
return "" # no netork_ip detected — skips firewall step
if command.startswith("cat /etc/snmp/snmpd.conf"):
return "agentAddress udp:161\nrocommunity public\n"
if command.startswith("snmpget"):
return "STRING: Linux test"
return ""
class TestActionFixSnmp:
def test_runs_apt_get_update_before_install(self, driver):
driver._pkg_manager = "apt"
driver._sudo_password = None
sudo_calls: list[str] = []
def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
sudo_calls.append(command)
return ""
with (
patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
):
driver._action_fix_snmp()
update_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get update" in c)
install_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get install" in c)
assert update_idx < install_idx
def test_install_exception_returns_failure_instead_of_raising(self, driver):
driver._pkg_manager = "apt"
driver._sudo_password = None
def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
if "apt-get install" in command:
raise TimeoutError("connection timed out")
return ""
with (
patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
):
result = driver._action_fix_snmp()
assert result["success"] is False
assert "install failed" in result["output"]
def test_apt_get_update_failure_is_non_fatal(self, driver):
"""apt-get update failing (e.g. transient network issue) must not
abort the whole action — install is still attempted."""
driver._pkg_manager = "apt"
driver._sudo_password = None
def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
if "apt-get update" in command:
raise TimeoutError("network unreachable")
return ""
with (
patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
):
result = driver._action_fix_snmp()
assert "apt-get update failed" in result["output"]
class TestActionAptUpdateUpgrade:
def test_skips_when_not_apt(self, driver):
driver._pkg_manager = "dnf"
result = driver._action_apt_update_upgrade()
assert result["success"] is True
assert "Skipped" in result["output"]
def test_fails_when_sudo_needs_password_and_none_configured(self, driver):
driver._pkg_manager = "apt"
driver._sudo_password = None
with patch.object(driver, "_send", return_value="__SUDO_NEEDS_PW__"):
result = driver._action_apt_update_upgrade()
assert result["success"] is False
assert "sudo requires a password" in result["output"]
def test_runs_update_then_upgrade_successfully(self, driver):
driver._pkg_manager = "apt"
driver._sudo_password = "secret" # noqa: S105
with (
patch.object(driver, "_send", return_value=""),
patch.object(
driver,
"_sudo",
side_effect=["Reading package lists... Done", "0 upgraded, 0 newly installed"],
) as mock_sudo,
):
result = driver._action_apt_update_upgrade()
assert result["success"] is True
assert "[update]" in result["output"]
assert "[upgrade]" in result["output"]
update_call, upgrade_call = mock_sudo.call_args_list
assert "apt-get update" in update_call.args[0]
assert "apt-get upgrade" in upgrade_call.args[0]
def test_exception_during_upgrade_returns_failure(self, driver):
driver._pkg_manager = "apt"
driver._sudo_password = "secret" # noqa: S105
with (
patch.object(driver, "_send", return_value=""),
patch.object(
driver, "_sudo", side_effect=["update ok", TimeoutError("connection lost")]
),
):
result = driver._action_apt_update_upgrade()
assert result["success"] is False
assert "[error]" in result["output"]
class TestRunDeviceActionDispatch:
def test_apt_update_upgrade_action_dispatches(self, driver):
with patch.object(
driver, "_action_apt_update_upgrade", return_value={"success": True, "output": ""}
) as mock_action:
driver.run_device_action("apt_update_upgrade")
mock_action.assert_called_once()