185 lines
5.8 KiB
Python
185 lines
5.8 KiB
Python
# ruff: noqa: INP001
|
|
"""Unit tests for system_health service helpers.
|
|
|
|
All tests are pure-Python — no gateway connection required.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta
|
|
|
|
import pytest
|
|
|
|
from app.services.openclaw.system_health import (
|
|
HealthSnapshot,
|
|
HealthHistory,
|
|
parse_health_response,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
FIXTURE_FULL = {
|
|
"cpu": {"usage": 45.2, "cores": 8, "loadAvg": [1.2, 1.5, 1.8]},
|
|
"memory": {"used": 4_294_967_296, "total": 16_777_216_000, "percent": 25.6},
|
|
"disk": {"used": 50_000_000_000, "total": 500_000_000_000, "percent": 10.0},
|
|
"uptime": 86400,
|
|
"platform": "linux",
|
|
"hostname": "gateway-host",
|
|
}
|
|
|
|
FIXTURE_ALT_KEYS = {
|
|
"cpuUsage": 33.3,
|
|
"memUsed": 2_000_000_000,
|
|
"memTotal": 8_000_000_000,
|
|
"diskUsed": 100_000_000_000,
|
|
"diskTotal": 1_000_000_000_000,
|
|
"uptimeSeconds": 3600,
|
|
}
|
|
|
|
FIXTURE_MINIMAL = {"uptime": 120}
|
|
|
|
FIXTURE_EMPTY = {}
|
|
FIXTURE_NONE = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# parse_health_response
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestParseHealthResponse:
|
|
|
|
def test_full_response_parsed(self):
|
|
snap = parse_health_response(FIXTURE_FULL)
|
|
assert isinstance(snap, HealthSnapshot)
|
|
|
|
def test_cpu_percent(self):
|
|
snap = parse_health_response(FIXTURE_FULL)
|
|
assert snap.cpu_pct == pytest.approx(45.2)
|
|
|
|
def test_memory_percent(self):
|
|
snap = parse_health_response(FIXTURE_FULL)
|
|
assert snap.memory_pct == pytest.approx(25.6)
|
|
|
|
def test_disk_percent(self):
|
|
snap = parse_health_response(FIXTURE_FULL)
|
|
assert snap.disk_pct == pytest.approx(10.0)
|
|
|
|
def test_uptime_seconds(self):
|
|
snap = parse_health_response(FIXTURE_FULL)
|
|
assert snap.uptime_seconds == 86400
|
|
|
|
def test_hostname(self):
|
|
snap = parse_health_response(FIXTURE_FULL)
|
|
assert snap.hostname == "gateway-host"
|
|
|
|
def test_alt_key_cpu(self):
|
|
snap = parse_health_response(FIXTURE_ALT_KEYS)
|
|
assert snap.cpu_pct == pytest.approx(33.3)
|
|
|
|
def test_alt_key_memory_computed(self):
|
|
snap = parse_health_response(FIXTURE_ALT_KEYS)
|
|
# 2GB / 8GB = 25%
|
|
assert snap.memory_pct == pytest.approx(25.0, abs=1)
|
|
|
|
def test_alt_key_disk_computed(self):
|
|
snap = parse_health_response(FIXTURE_ALT_KEYS)
|
|
# 100GB / 1000GB = 10%
|
|
assert snap.disk_pct == pytest.approx(10.0, abs=1)
|
|
|
|
def test_alt_key_uptime(self):
|
|
snap = parse_health_response(FIXTURE_ALT_KEYS)
|
|
assert snap.uptime_seconds == 3600
|
|
|
|
def test_minimal_response(self):
|
|
snap = parse_health_response(FIXTURE_MINIMAL)
|
|
assert snap.uptime_seconds == 120
|
|
assert snap.cpu_pct is None
|
|
assert snap.memory_pct is None
|
|
|
|
def test_empty_response(self):
|
|
snap = parse_health_response(FIXTURE_EMPTY)
|
|
assert snap.cpu_pct is None
|
|
assert snap.memory_pct is None
|
|
assert snap.uptime_seconds is None
|
|
|
|
def test_none_response(self):
|
|
snap = parse_health_response(FIXTURE_NONE)
|
|
assert snap is not None # always returns a snapshot
|
|
assert snap.cpu_pct is None
|
|
|
|
def test_recorded_at_is_set(self):
|
|
snap = parse_health_response(FIXTURE_FULL)
|
|
assert isinstance(snap.recorded_at, datetime)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# HealthHistory
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestHealthHistory:
|
|
|
|
def _make_snap(self, offset_hours: float = 0) -> HealthSnapshot:
|
|
from app.core.time import utcnow
|
|
snap = parse_health_response(FIXTURE_FULL)
|
|
snap.recorded_at = utcnow() - timedelta(hours=offset_hours)
|
|
return snap
|
|
|
|
def test_add_snapshot(self):
|
|
history = HealthHistory()
|
|
history.add("gw-1", self._make_snap())
|
|
assert len(history.get("gw-1")) == 1
|
|
|
|
def test_multiple_snapshots(self):
|
|
history = HealthHistory()
|
|
for _ in range(5):
|
|
history.add("gw-1", self._make_snap())
|
|
assert len(history.get("gw-1")) == 5
|
|
|
|
def test_old_snapshots_pruned(self):
|
|
history = HealthHistory(window_hours=24)
|
|
# Add a snapshot 25 hours old
|
|
old = self._make_snap(offset_hours=25)
|
|
history.add("gw-1", old)
|
|
# Add a recent snapshot
|
|
history.add("gw-1", self._make_snap())
|
|
snaps = history.get("gw-1")
|
|
assert len(snaps) == 1 # old one pruned
|
|
|
|
def test_different_gateways_isolated(self):
|
|
history = HealthHistory()
|
|
history.add("gw-1", self._make_snap())
|
|
history.add("gw-2", self._make_snap())
|
|
assert len(history.get("gw-1")) == 1
|
|
assert len(history.get("gw-2")) == 1
|
|
|
|
def test_unknown_gateway_returns_empty(self):
|
|
history = HealthHistory()
|
|
assert history.get("nonexistent") == []
|
|
|
|
def test_snapshots_ordered_oldest_first(self):
|
|
history = HealthHistory()
|
|
newer = self._make_snap(offset_hours=0)
|
|
older = self._make_snap(offset_hours=1)
|
|
history.add("gw-1", newer)
|
|
history.add("gw-1", older)
|
|
snaps = history.get("gw-1")
|
|
# oldest first
|
|
assert snaps[0].recorded_at <= snaps[1].recorded_at
|
|
|
|
def test_latest_snapshot(self):
|
|
history = HealthHistory()
|
|
old = self._make_snap(offset_hours=2)
|
|
new = self._make_snap(offset_hours=0)
|
|
history.add("gw-1", old)
|
|
history.add("gw-1", new)
|
|
latest = history.latest("gw-1")
|
|
assert latest is not None
|
|
assert latest.recorded_at >= old.recorded_at
|
|
|
|
def test_latest_none_when_empty(self):
|
|
history = HealthHistory()
|
|
assert history.latest("gw-1") is None
|