56 lines
2.3 KiB
Python
56 lines
2.3 KiB
Python
from unittest.mock import patch
|
|
from plugins.host_agent import agent as a
|
|
|
|
|
|
def test_build_sample_shape():
|
|
fake_mem = {"total_bytes": 100, "used_bytes": 40,
|
|
"available_bytes": 60, "swap_used_bytes": 0}
|
|
fake_load = {"1m": 0.1, "5m": 0.2, "15m": 0.3}
|
|
fake_storage = [{"mount": "/", "total_bytes": 1000, "used_bytes": 400}]
|
|
|
|
with patch.object(a, "collect_cpu", return_value=12.5), \
|
|
patch.object(a, "collect_memory", return_value=fake_mem), \
|
|
patch.object(a, "collect_load", return_value=fake_load), \
|
|
patch.object(a, "collect_uptime", return_value=1234), \
|
|
patch.object(a, "collect_storage", return_value=fake_storage):
|
|
sample = a.build_sample(mounts=["/"])
|
|
|
|
assert sample["cpu_pct"] == 12.5
|
|
assert sample["mem"]["used_bytes"] == 40
|
|
assert sample["load"]["1m"] == 0.1
|
|
assert sample["uptime_secs"] == 1234
|
|
assert sample["storage"][0]["mount"] == "/"
|
|
assert "ts" in sample
|
|
# ISO-8601 UTC string
|
|
assert "T" in sample["ts"]
|
|
|
|
|
|
def test_build_sample_tolerates_collector_failure():
|
|
# If a collector raises, the sample should still be built with None for that field.
|
|
def _boom():
|
|
raise RuntimeError("no /proc")
|
|
|
|
with patch.object(a, "collect_cpu", side_effect=_boom), \
|
|
patch.object(a, "collect_memory", return_value={"total_bytes": 1, "used_bytes": 0,
|
|
"available_bytes": 1, "swap_used_bytes": 0}), \
|
|
patch.object(a, "collect_load", return_value={"1m": 0.0, "5m": 0.0, "15m": 0.0}), \
|
|
patch.object(a, "collect_uptime", return_value=0), \
|
|
patch.object(a, "collect_storage", return_value=[]):
|
|
sample = a.build_sample(mounts=["/"])
|
|
assert sample["cpu_pct"] is None
|
|
assert sample["mem"]["total_bytes"] == 1
|
|
|
|
|
|
def test_build_payload_wraps_samples():
|
|
metadata = {"kernel": "6.8", "distro": "Ubuntu", "arch": "x86_64"}
|
|
payload = a.build_payload(
|
|
samples=[{"ts": "2026-04-14T00:00:00+00:00", "cpu_pct": 5.0}],
|
|
hostname="myhost",
|
|
metadata=metadata,
|
|
)
|
|
assert payload["agent_version"] == a.AGENT_VERSION
|
|
assert payload["hostname"] == "myhost"
|
|
assert payload["metadata"] == metadata
|
|
assert len(payload["samples"]) == 1
|
|
assert payload["samples"][0]["cpu_pct"] == 5.0
|