Pipeline/backend/tests/test_system_health.py

# 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