102 lines
3.0 KiB
Python
102 lines
3.0 KiB
Python
# 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"]
|