99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
"""Pure-function tests for host_agent metric expansion."""
|
|
from datetime import datetime, timezone
|
|
|
|
from plugins.host_agent.routes import _expand_sample_to_metrics
|
|
|
|
|
|
NOW = datetime(2026, 4, 14, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def _sample() -> dict:
|
|
return {
|
|
"ts": NOW.isoformat(),
|
|
"cpu_pct": 12.5,
|
|
"mem": {
|
|
"total_bytes": 16_000_000_000,
|
|
"used_bytes": 8_000_000_000,
|
|
"available_bytes": 8_000_000_000,
|
|
"swap_used_bytes": 0,
|
|
},
|
|
"load": {"1m": 0.1, "5m": 0.2, "15m": 0.3},
|
|
"uptime_secs": 1234,
|
|
"storage": [
|
|
{"mount": "/", "total_bytes": 1000, "used_bytes": 400},
|
|
{"mount": "/mnt/data", "total_bytes": 2000, "used_bytes": 1800},
|
|
],
|
|
}
|
|
|
|
|
|
def _by_metric(rows, name):
|
|
return [r for r in rows if r.metric_name == name]
|
|
|
|
|
|
def test_expand_emits_cpu_mem_load_uptime():
|
|
rows = _expand_sample_to_metrics(_sample(), "testhost", NOW)
|
|
names = {r.metric_name for r in rows}
|
|
assert "cpu_pct" in names
|
|
assert "mem_used_pct" in names
|
|
assert "mem_available_bytes" in names
|
|
assert "swap_used_bytes" in names
|
|
assert "load_1m" in names and "load_5m" in names and "load_15m" in names
|
|
assert "uptime_secs" in names
|
|
|
|
|
|
def test_expand_source_module_and_recorded_at():
|
|
rows = _expand_sample_to_metrics(_sample(), "testhost", NOW)
|
|
assert rows, "expected non-empty rows"
|
|
for r in rows:
|
|
assert r.source_module == "host_agent"
|
|
assert r.recorded_at == NOW
|
|
|
|
|
|
def test_expand_cpu_value_and_resource():
|
|
cpu = _by_metric(_expand_sample_to_metrics(_sample(), "testhost", NOW), "cpu_pct")
|
|
assert len(cpu) == 1
|
|
assert cpu[0].value == 12.5
|
|
assert cpu[0].resource_name == "testhost"
|
|
|
|
|
|
def test_expand_mem_used_pct_math():
|
|
# used = total - available = 50%
|
|
used = _by_metric(_expand_sample_to_metrics(_sample(), "testhost", NOW), "mem_used_pct")
|
|
assert len(used) == 1
|
|
assert abs(used[0].value - 50.0) < 0.01
|
|
|
|
|
|
def test_expand_per_mount_disk_metrics():
|
|
rows = _expand_sample_to_metrics(_sample(), "testhost", NOW)
|
|
per_mount = _by_metric(rows, "disk_used_pct")
|
|
assert {r.resource_name for r in per_mount} == {"testhost:/", "testhost:/mnt/data"}
|
|
# Values: / is 40%, /mnt/data is 90%
|
|
by_res = {r.resource_name: r.value for r in per_mount}
|
|
assert abs(by_res["testhost:/"] - 40.0) < 0.01
|
|
assert abs(by_res["testhost:/mnt/data"] - 90.0) < 0.01
|
|
|
|
|
|
def test_expand_disk_worst_is_max():
|
|
worst = _by_metric(_expand_sample_to_metrics(_sample(), "testhost", NOW), "disk_used_pct_worst")
|
|
assert len(worst) == 1
|
|
assert worst[0].resource_name == "testhost"
|
|
assert abs(worst[0].value - 90.0) < 0.01
|
|
|
|
|
|
def test_expand_skips_missing_cpu_and_load():
|
|
sample = {"mem": {"total_bytes": 100, "available_bytes": 50}}
|
|
rows = _expand_sample_to_metrics(sample, "h", NOW)
|
|
names = {r.metric_name for r in rows}
|
|
assert "cpu_pct" not in names
|
|
assert "load_1m" not in names
|
|
assert "uptime_secs" not in names
|
|
assert "mem_used_pct" in names
|
|
|
|
|
|
def test_expand_empty_storage_emits_no_disk_worst():
|
|
sample = {"cpu_pct": 1.0, "storage": []}
|
|
rows = _expand_sample_to_metrics(sample, "h", NOW)
|
|
names = {r.metric_name for r in rows}
|
|
assert "disk_used_pct_worst" not in names
|
|
assert "cpu_pct" in names
|