From b6b1c8adca6a9c4f9c62455c03d98e13c44b279f Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 14 Apr 2026 22:07:39 -0400 Subject: [PATCH] test(host_agent): agent resource collectors unit tests Co-Authored-By: Claude Sonnet 4.6 --- .../host_agent/test_agent_collectors.py | 101 ++++++++++++++++++ 1 file changed, 101 insertions(+) create mode 100644 tests/plugins/host_agent/test_agent_collectors.py diff --git a/tests/plugins/host_agent/test_agent_collectors.py b/tests/plugins/host_agent/test_agent_collectors.py new file mode 100644 index 0000000..9f3d1fa --- /dev/null +++ b/tests/plugins/host_agent/test_agent_collectors.py @@ -0,0 +1,101 @@ +# tests/plugins/host_agent/test_agent_collectors.py +"""Unit tests for per-resource collectors, driven by fixture /proc files.""" +from unittest.mock import patch +import pytest +from plugins.host_agent import agent as a + + +PROC_STAT_SAMPLE_1 = "cpu 100 0 50 850 0 0 0 0 0 0\n" +PROC_STAT_SAMPLE_2 = "cpu 200 0 100 1700 0 0 0 0 0 0\n" + +PROC_MEMINFO = ( + "MemTotal: 16000000 kB\n" + "MemFree: 2000000 kB\n" + "MemAvailable: 6000000 kB\n" + "SwapTotal: 4000000 kB\n" + "SwapFree: 3500000 kB\n" +) + +PROC_LOADAVG = "0.42 0.55 0.61 1/234 5678\n" +PROC_UPTIME = "482934.12 1000000.00\n" +OS_RELEASE = 'PRETTY_NAME="Ubuntu 24.04 LTS"\nNAME="Ubuntu"\n' + + +def test_collect_memory_uses_available(tmp_path): + p = tmp_path / "meminfo" + p.write_text(PROC_MEMINFO) + with patch.object(a, "MEMINFO_PATH", str(p)): + mem = a.collect_memory() + assert mem["total_bytes"] == 16000000 * 1024 + assert mem["available_bytes"] == 6000000 * 1024 + assert mem["used_bytes"] == (16000000 - 6000000) * 1024 + assert mem["swap_used_bytes"] == (4000000 - 3500000) * 1024 + + +def test_collect_load(tmp_path): + p = tmp_path / "loadavg" + p.write_text(PROC_LOADAVG) + with patch.object(a, "LOADAVG_PATH", str(p)): + load = a.collect_load() + assert load == {"1m": 0.42, "5m": 0.55, "15m": 0.61} + + +def test_collect_uptime(tmp_path): + p = tmp_path / "uptime" + p.write_text(PROC_UPTIME) + with patch.object(a, "UPTIME_PATH", str(p)): + assert a.collect_uptime() == 482934 + + +def test_collect_cpu_reads_twice(): + # Two sequential _read_file calls return two stat snapshots. + reads = iter([PROC_STAT_SAMPLE_1, PROC_STAT_SAMPLE_2]) + + def _read(_path): + return next(reads) + + with patch.object(a, "_read_file", _read), \ + patch.object(a.time, "sleep", lambda _s: None): + pct = a.collect_cpu(sample_window=0.0) + # totals: 1000 → 2000 (delta 1000). idle+iowait: 850 → 1700 (delta 850). + # busy = 150. pct = 15.0. + assert pct == pytest.approx(15.0, abs=0.1) + + +def test_collect_storage(): + fake = {"/": (1000, 400, 600), "/mnt/data": (2000, 500, 1500)} + + def _fake_usage(path): + total, used, free = fake[path] + + class R: + pass + + r = R() + r.total = total + r.used = used + r.free = free + return r + + with patch.object(a.shutil, "disk_usage", side_effect=_fake_usage): + out = a.collect_storage(["/", "/mnt/data"]) + assert out == [ + {"mount": "/", "total_bytes": 1000, "used_bytes": 400}, + {"mount": "/mnt/data", "total_bytes": 2000, "used_bytes": 500}, + ] + + +def test_collect_metadata(tmp_path): + p = tmp_path / "os-release" + p.write_text(OS_RELEASE) + + class Uname: + release = "6.8.0-45-generic" + machine = "x86_64" + + with patch.object(a, "OS_RELEASE_PATH", str(p)), \ + patch.object(a.os, "uname", return_value=Uname()): + meta = a.collect_metadata() + assert meta["kernel"] == "6.8.0-45-generic" + assert meta["arch"] == "x86_64" + assert "Ubuntu 24.04" in meta["distro"]