# tests/plugins/host_agent/test_agent_updates.py """#5765: the agent says what the package manager has waiting and whether the host needs a reboot, without root, without the network, and without a slow apt ever holding up a sample.""" import os import subprocess import threading import time from plugins.host_agent import agent as a def test_apt_check_counts_are_read_from_its_stderr(): assert a.parse_apt_check("12;3") == (12, 3) assert a.parse_apt_check("") is None def test_apt_list_counts_security_updates_by_their_suite(): out = ("Listing...\n" "libc6/noble-updates,noble-security 2.39-0ubuntu8.4 amd64 [upgradable from: 2.39-0ubuntu8.3]\n" "curl/noble-updates 8.5.0-2ubuntu10.5 amd64 [upgradable from: 8.5.0-2ubuntu10.4]\n") assert a.parse_apt_list(out) == (2, 1) def test_dnf_check_update_counts_packages_and_stops_at_obsoletes(): out = ("\nkernel.x86_64 5.14.0-503 baseos\nopenssl.x86_64 3.2.2-6 baseos\n" "Obsoleting Packages\ngrub2.x86_64 2.06 baseos\n") assert a.parse_dnf_check_update(out) == 2 def test_apt_hosts_are_checked_with_apt_check(monkeypatch): calls = [] def fake_run(cmd, deadline=a.UPDATES_DEADLINE_SECONDS): calls.append(cmd[0]) return (0, "", "7;2") if cmd[0].endswith("apt-check") else None monkeypatch.setattr(a, "_run", fake_run) got = a.collect_updates() assert (got["manager"], got["pending"], got["security"]) == ("apt", 7, 2) assert calls == ["/usr/lib/update-notifier/apt-check"] def test_a_package_manager_that_hangs_is_an_error_not_a_stall(monkeypatch): def hang(cmd, deadline=a.UPDATES_DEADLINE_SECONDS): raise subprocess.TimeoutExpired(cmd, deadline) monkeypatch.setattr(a, "_run", hang) assert "didn't answer" in a.collect_updates()["error"] def test_the_check_runs_in_the_background_once_an_hour(monkeypatch): started, release = threading.Event(), threading.Event() def slow(): started.set() release.wait(5) return {"manager": "apt", "pending": 1, "security": 0, "checked_at": "x"} monkeypatch.setattr(a, "collect_updates", slow) state: dict = {} a.refresh_updates(state, now_mono=1000.0) assert started.wait(2) a.refresh_updates(state, now_mono=1001.0) # still running: not started twice release.set() for _ in range(50): if not state.get("updates_running"): break time.sleep(0.02) assert state["updates"]["pending"] == 1 monkeypatch.setattr(a, "collect_updates", lambda: {"pending": 2}) a.refresh_updates(state, now_mono=1000.0 + 600) # inside the hour: no new check assert state["updates"]["pending"] == 1 state["updates_due"] = True # asked for now: checked now a.refresh_updates(state, now_mono=1000.0 + 601) for _ in range(50): if state["updates"].get("pending") == 2: break time.sleep(0.02) assert state["updates"]["pending"] == 2 def test_a_reboot_is_required_when_debian_says_so(tmp_path): marker = tmp_path / "reboot-required" marker.write_text("*** System restart required ***\n") (tmp_path / "reboot-required.pkgs").write_text("linux-image-6.8.0-146-generic\nlibc6\nlibc6\n") os.utime(marker, (1792000000, 1792000000)) got = a.collect_reboot(str(marker)) assert got["required"] is True assert got["packages"] == ["libc6", "linux-image-6.8.0-146-generic"] assert got["since"].startswith("2026-10-14") # ── pacman, zypper and yum (#5803) ─────────────────────────────────────────── CHECKUPDATES = ("linux 6.11.1.arch1-1 -> 6.11.2.arch1-1\n" "openssl 3.3.2-1 -> 3.4.0-1\n" "python-requests 2.32.3-1 -> 2.32.3-2\n") ZYPPER_UPDATES = ( "S | Repository | Name | Current Version | Available Version | Arch\n" "--+-----------------+-------------+---------------------+---------------------+-------\n" "v | repo-sle-update | libopenssl3 | 3.1.4-150600.5.7.1 | 3.1.4-150600.5.10.1 | x86_64\n" "v | repo-oss | vim | 9.1.0330-1.1 | 9.1.0698-1.1 | x86_64\n") ZYPPER_PATCHES = ( "Repository | Name | Category | Severity | Interactive | Status | Since | Summary\n" "----------------+-----------------------------+----------+-----------+-------------+------------+-------+--------\n" "repo-sle-update | openSUSE-SLE-15.6-2024-3601 | security | important | --- | needed | - | Security update for openssl-3\n" "repo-sle-update | openSUSE-SLE-15.6-2024-3500 | security | moderate | --- | applied | - | Security update for curl\n" "repo-sle-update | openSUSE-SLE-15.6-2024-3488 | security | low | --- | not needed | - | Security update for ruby\n") YUM_CHECK_UPDATE = ("Loaded plugins: fastestmirror\n\n" "kernel.x86_64 3.10.0-1160.119.1.el7 updates\n" "openssl.x86_64 1:1.0.2k-26.el7_9 updates\n") def test_each_managers_list_is_counted(): assert a.parse_checkupdates(CHECKUPDATES) == 3 assert a.parse_checkupdates("") == 0 assert a.parse_zypper_updates(ZYPPER_UPDATES) == 2 assert a.parse_zypper_updates("No updates found.\n") == 0 assert a.parse_zypper_patches(ZYPPER_PATCHES) == 1 # applied and not-needed aren't waiting assert a.parse_dnf_check_update(YUM_CHECK_UPDATE) == 2 def _host(monkeypatch, answers: dict, binaries=()): """A host whose commands answer from `answers` (keyed by the whole command, its first two words, or its first) and with `binaries` on the PATH.""" calls = [] def fake_run(cmd, deadline=a.UPDATES_DEADLINE_SECONDS): calls.append(" ".join(cmd)) for key in (" ".join(cmd), " ".join(cmd[:2]), cmd[0]): if key in answers: return answers[key] return None monkeypatch.setattr(a, "_run", fake_run) monkeypatch.setattr(a.shutil, "which", lambda name: f"/usr/bin/{name}" if name in binaries else None) return calls def test_a_pacman_host_is_checked_with_checkupdates(monkeypatch): _host(monkeypatch, {"checkupdates": (0, CHECKUPDATES, "")}, binaries=("pacman",)) got = a.collect_updates() assert (got["manager"], got["pending"], got["security"]) == ("pacman", 3, None) _host(monkeypatch, {"checkupdates": (2, "", "")}, binaries=("pacman",)) # 2: up to date assert a.collect_updates()["pending"] == 0 def test_a_pacman_host_without_checkupdates_says_what_is_missing(monkeypatch): _host(monkeypatch, {}, binaries=("pacman",)) got = a.collect_updates() assert got["manager"] == "pacman" and "pacman-contrib" in got["error"] def test_a_zypper_host_reads_updates_security_patches_and_reboot(monkeypatch): monkeypatch.setattr(a, "_os_release_id", lambda path="/etc/os-release": "opensuse-leap") _host(monkeypatch, { "zypper --no-refresh -q list-updates": (0, ZYPPER_UPDATES, ""), "zypper --no-refresh -q list-patches --category security": (0, ZYPPER_PATCHES, ""), "zypper needs-rebooting": (102, "", "")}) # 102: a reboot is needed got = a.collect_updates() assert (got["manager"], got["pending"], got["security"]) == ("zypper", 2, 1) assert got["reboot_required"] is True def test_tumbleweed_counts_what_dup_would_change_and_has_no_security_patches(monkeypatch): monkeypatch.setattr(a, "_os_release_id", lambda path="/etc/os-release": "opensuse-tumbleweed") calls = _host(monkeypatch, {"zypper --no-refresh": (0, ZYPPER_UPDATES, ""), "zypper needs-rebooting": (0, "", "")}) got = a.collect_updates() assert got["pending"] == 2 and got["security"] is None and got["reboot_required"] is False assert "zypper --no-refresh -q list-updates --dup" in calls assert not any("list-patches" in c for c in calls) def test_a_yum_host_without_dnf_is_checked_like_dnf(monkeypatch): _host(monkeypatch, {"yum -q": (100, YUM_CHECK_UPDATE, ""), "needs-restarting -r": (1, "", "")}) got = a.collect_updates() assert (got["manager"], got["pending"]) == ("yum", 2) assert got["reboot_required"] is True def test_a_failed_check_names_its_manager(monkeypatch): _host(monkeypatch, {"zypper --no-refresh": (7, "", "System management is locked by PackageKit")}) monkeypatch.setattr(a, "_os_release_id", lambda path="/etc/os-release": "sles") got = a.collect_updates() assert got["manager"] == "zypper" and "locked" in got["error"] def test_a_pacman_host_needs_a_reboot_once_its_running_kernel_is_replaced(monkeypatch, tmp_path): monkeypatch.setattr(a.shutil, "which", lambda name: "/usr/bin/pacman" if name == "pacman" else None) modules = tmp_path / "modules" (modules / os.uname().release).mkdir(parents=True) no_marker = str(tmp_path / "reboot-required") assert a.collect_reboot(no_marker, str(modules)) == {"required": False} (modules / os.uname().release).rmdir() # the upgrade removed it assert a.collect_reboot(no_marker, str(modules)) == {"required": True} def test_a_host_with_no_reboot_marker_to_read_says_nothing(monkeypatch, tmp_path): monkeypatch.setattr(a.shutil, "which", lambda name: None) assert a.collect_reboot(str(tmp_path / "reboot-required")) is None