dashboards/tests/test_host_source.py

"""Agent snapshot -> dashboard payload: rates, null-safety, and honest fleet totals.

The rule this file defends hardest is *null in, null out*. A missing sensor rendered as 0
is a dashboard that says "the GPU is idle" when it means "the driver said nothing", and on
a machine you are debugging at 2am that difference is the whole point.
"""
from __future__ import annotations

import copy
import math
from typing import Any

import pytest

from dashboards.rates import RateEstimator
from dashboards.severity import classify
from dashboards.sources.host import HostCollector, fleet_summary, parse_host_snapshot

TICKS_PER_SECOND = 100.0


def make_collector() -> HostCollector:
    return HostCollector(RateEstimator(span=60.0))


def advance(snapshot: dict[str, Any], dt: float = 10.0, *, total_ticks: float = 1000.0,
            idle_ticks: float = 900.0, read_bytes: float = 0.0, write_bytes: float = 0.0,
            rx_bytes: float = 0.0, tx_bytes: float = 0.0) -> dict[str, Any]:
    """A second scrape `dt` later with counters moved by known amounts."""
    snap = copy.deepcopy(snapshot)
    snap["ts"] = float(snapshot["ts"]) + dt
    modes = snap["cpu"]["modes"]
    idle_parts = {"idle": idle_ticks * 0.9, "iowait": idle_ticks * 0.1}
    for key, value in modes.items():
        if key == "total":
            modes["total"] = value + total_ticks
        elif key in idle_parts:
            modes[key] = value + idle_parts[key]
        elif key != "guest" and key != "guest_nice":
            modes[key] = value + (total_ticks - idle_ticks) / 5
    for device in snap["disk_io"]["devices"]:
        device["read_bytes_total"] += read_bytes
        device["write_bytes_total"] += write_bytes
        device["reads_total"] += read_bytes / 4096
        device["writes_total"] += write_bytes / 4096
    for iface in snap["network"]["interfaces"]:
        iface["rx_bytes_total"] += rx_bytes
        iface["tx_bytes_total"] += tx_bytes
    return snap


def parse_twice(agent_snapshot) -> tuple[dict[str, Any], dict[str, Any]]:
    collector = make_collector()
    first = parse_host_snapshot("driveripper", agent_snapshot, collector, now=1_000.0)
    second_snap = advance(agent_snapshot, 10.0, read_bytes=4_096_000, write_bytes=2_048_000,
                          rx_bytes=100_000, tx_bytes=50_000)
    second = parse_host_snapshot("driveripper", second_snap, collector, now=1_010.0)
    return first.payload, second.payload


def test_first_scrape_has_no_rates_but_reports_instantaneous_values(agent_snapshot) -> None:
    collector = make_collector()
    payload = parse_host_snapshot("driveripper", agent_snapshot, collector, now=1_000.0).payload
    assert payload["cpu"]["usage_pct"] is None, "one sample cannot make a rate"
    assert payload["cpu"]["load1"] is not None, "load is a gauge, available immediately"
    assert payload["memory"]["used_pct"] is not None
    assert payload["disks"]["read_bytes_per_s"] is None
    assert payload["network"]["rx_bytes_per_s"] is None
    assert payload["window_seconds"] is None


def test_second_scrape_measures_the_window_it_claims(agent_snapshot) -> None:
    """Two scrapes are enough — a dashboard that needs three looks to say anything about
    CPU is a dashboard whose first minute of history is a hole."""
    _first, payload = parse_twice(agent_snapshot)
    assert payload["window_seconds"] == 10.0
    usage = payload["cpu"]["usage_pct"]
    assert usage is not None
    # 1000 ticks total, 900 idle over 10s => exactly 10% busy, whatever the host was doing
    assert usage == 10.0


def test_iowait_counts_as_idle_but_is_still_reported(agent_snapshot) -> None:
    _first, payload = parse_twice(agent_snapshot)
    modes = payload["cpu"]["modes_pct"]
    assert modes["iowait"] is not None and modes["iowait"] > 0
    assert sum(v for v in modes.values() if v is not None) <= 100.0001


def test_per_core_series_is_one_point_per_logical_cpu(agent_snapshot) -> None:
    collector = make_collector()
    parse_host_snapshot("driveripper", agent_snapshot, collector, now=1_000.0)
    payload = parse_host_snapshot("driveripper", advance(agent_snapshot, 10.0), collector,
                                  now=1_010.0).payload
    cores = payload["cpu"]["core_usage_pct"]
    assert len(cores) == payload["info"]["cpu_cores_logical"] == 48
    assert all(value is None or 0.0 <= value <= 100.0 for value in cores)


def test_disk_and_network_rates_are_bytes_per_second_over_the_window(agent_snapshot) -> None:
    _first, payload = parse_twice(agent_snapshot)
    devices = payload["disks"]["devices"]
    assert devices, "captured host has block devices"
    # the fixture scrape added the same amount to every device/interface, so the per-device
    # figure is delta/dt and the headline total is the sum of what the payload shows
    assert devices[0]["read_bytes_per_s"] == 4_096_000 / 10.0
    assert payload["disks"]["read_bytes_per_s"] == sum(
        d["read_bytes_per_s"] or 0.0 for d in devices)
    assert payload["disks"]["write_bytes_per_s"] == sum(
        d["write_bytes_per_s"] or 0.0 for d in devices)
    interfaces = payload["network"]["interfaces"]
    assert payload["network"]["rx_bytes_per_s"] == 100_000 / 10.0 * len(interfaces)
    assert payload["network"]["tx_bytes_per_s"] == 50_000 / 10.0 * len(interfaces)


def test_idle_machine_reports_zero_not_none_once_measured(agent_snapshot) -> None:
    """0 and n/a are different claims: 'nothing happened' vs 'we could not tell'."""
    collector = make_collector()
    parse_host_snapshot("driveripper", agent_snapshot, collector, now=1_000.0)
    quiet = advance(agent_snapshot, 10.0, total_ticks=1_000.0, idle_ticks=1_000.0)
    payload = parse_host_snapshot("driveripper", quiet, collector, now=1_010.0).payload
    assert payload["cpu"]["usage_pct"] == 0.0
    assert payload["disks"]["read_bytes_per_s"] == 0.0
    assert payload["network"]["rx_bytes_per_s"] == 0.0


def test_missing_sensors_stay_missing(agent_snapshot) -> None:
    snap = copy.deepcopy(agent_snapshot)
    snap["temperatures"] = []                    # the agent reports a flat sensor list
    snap["cpu_temperature_celsius"] = None       # ...plus its own pick for "the CPU temp"
    snap["cpu"]["freq_mhz"] = None
    for gpu in snap["gpus"]:
        gpu["power_watts"] = None
        gpu["temps"] = {}
    collector = make_collector()
    payload = parse_host_snapshot("driveripper", snap, collector, now=1_000.0).payload
    assert payload["cpu"]["temp_c"] is None
    assert payload["cpu"]["freq_mhz"] is None
    assert payload["temperatures"]["max_c"] is None
    assert payload["temperatures"]["sensors"] == []
    for gpu in payload["gpus"]:
        assert gpu["power_watts"] is None
        assert gpu["temps"] == {}
        assert gpu["temp_c"] is None


def test_no_sample_is_ever_written_as_zero_for_a_missing_value(agent_snapshot) -> None:
    snap = copy.deepcopy(agent_snapshot)
    snap["cpu"]["freq_mhz"] = None
    snap["gpus"] = []
    collector = make_collector()
    result = parse_host_snapshot("driveripper", snap, collector, now=1_000.0)
    for sample in result.samples:
        assert sample.value is not None
        assert math.isfinite(sample.value), f"{sample.metric} is {sample.value}"
    assert not any(s.metric == "cpu.freq_mhz" for s in result.samples)


def test_gpu_pool_percent_accounts_for_vram_and_gtt(agent_snapshot) -> None:
    """On an APU the model lives in GTT; reporting only vram says '32 MB used' on a 128 GB box."""
    collector = make_collector()
    payload = parse_host_snapshot("driveripper", agent_snapshot, collector, now=1_000.0).payload
    gpu = payload["gpus"][0]
    assert gpu["model_full"] and gpu["model"]
    assert gpu["vram_total_bytes"] > 0
    assert gpu["mem_used_pct"] is not None and 0.0 <= gpu["mem_used_pct"] <= 100.0
    assert "gtt_total_bytes" in gpu


def test_reboot_is_reported_as_an_event(agent_snapshot) -> None:
    collector = make_collector()
    parse_host_snapshot("driveripper", agent_snapshot, collector, now=1_000.0)
    snap = advance(agent_snapshot, 10.0)
    snap["host"]["boot_ts"] = float(agent_snapshot["host"]["boot_ts"]) + 5_000
    result = parse_host_snapshot("driveripper", snap, collector, now=1_010.0)
    reboots = [e for e in result.events if e.kind == "reboot"]
    assert len(reboots) == 1
    assert reboots[0].severity == "warn"
    assert "rebooted" in reboots[0].message


def test_agent_errors_travel_to_the_ui(agent_snapshot) -> None:
    snap = copy.deepcopy(agent_snapshot)
    snap["errors"] = ["disk_io: permission denied reading /sys/block/sdb"]
    payload = parse_host_snapshot("driveripper", snap, make_collector(), now=1_000.0).payload
    assert payload["agent"]["errors"], "a partial snapshot must be labelled partial"


def test_counter_reset_from_a_reboot_does_not_chart_a_spike(agent_snapshot) -> None:
    collector = make_collector()
    parse_host_snapshot("driveripper", agent_snapshot, collector, now=1_000.0)
    snap = copy.deepcopy(agent_snapshot)
    snap["cpu"]["modes"]["total"] = 1.0   # counter restarted from zero
    payload = parse_host_snapshot("driveripper", snap, collector, now=1_010.0).payload
    assert payload["cpu"]["usage_pct"] is None


def test_fleet_summary_ignores_hosts_that_are_not_answering(agent_snapshot) -> None:
    payload = parse_host_snapshot("driveripper", agent_snapshot, make_collector(),
                                  now=1_000.0).payload
    live = copy.deepcopy(payload)
    live["cpu"]["usage_pct"] = 40.0
    live["info"]["cpu_cores_logical"] = 48
    dead = copy.deepcopy(payload)
    dead["name"] = "deskwork"
    dead["cpu"]["usage_pct"] = None
    fleet = fleet_summary([live, dead])
    assert fleet["hosts_total"] == 2
    assert fleet["hosts_live"] == 1
    assert fleet["cpu_usage_pct_avg"] == 40.0, "a dead host must not drag the average to 0"


def test_fleet_summary_weights_cpu_by_cores(agent_snapshot) -> None:
    base = parse_host_snapshot("driveripper", agent_snapshot, make_collector(),
                               now=1_000.0).payload
    big = copy.deepcopy(base)
    big["cpu"]["usage_pct"] = 10.0
    big["info"]["cpu_cores_logical"] = 100
    small = copy.deepcopy(base)
    small["name"] = "gateway"
    small["cpu"]["usage_pct"] = 90.0
    small["info"]["cpu_cores_logical"] = 2
    fleet = fleet_summary([big, small])
    assert fleet["cpu_usage_pct_avg"] == pytest.approx((10 * 100 + 90 * 2) / 102)
    assert fleet["cpu_cores"] == 102


def test_fleet_summary_worst_disk_and_memory_totals(agent_snapshot) -> None:
    base = parse_host_snapshot("driveripper", agent_snapshot, make_collector(),
                               now=1_000.0).payload
    fleet = fleet_summary([base, base])
    assert fleet["disk_worst_used_pct"] == pytest.approx(base["disks"]["worst_used_pct"])
    assert fleet["memory_total_bytes"] == pytest.approx(2 * base["memory"]["total_bytes"])
    assert fleet["memory_used_pct"] == pytest.approx(
        base["memory"]["used_bytes"] / base["memory"]["total_bytes"] * 100)


# ---- severity: the verdict travels with the value ------------------------- #

def test_payloads_carry_a_severity_beside_every_rated_value(agent_snapshot) -> None:
    """The page paints a class; it must never hold its own opinion about where amber starts."""
    _first, payload = parse_twice(agent_snapshot)
    assert payload["cpu"]["severity"] == {
        "usage_pct": classify("cpu.usage_pct", payload["cpu"]["usage_pct"]),
        "temp_c": classify("cpu.temp_c", payload["cpu"]["temp_c"]),
    }
    assert payload["cpu"]["severity"]["usage_pct"] == "ok"          # the window is 10 % busy
    assert payload["cpu"]["severity"]["temp_c"] == "ok"             # 68 °C in the capture
    assert payload["memory"]["severity"]["used_pct"] == "ok"        # 68 % of 128 GB
    assert payload["disks"]["severity"]["worst_used_pct"] == "ok"   # 74.6 % worst mount
    gpu = payload["gpus"][0]
    assert gpu["severity"] == {
        "busy_pct": classify("gpu.busy_pct", gpu["busy_pct"]),
        "mem_used_pct": classify("gpu.mem_used_pct", gpu["mem_used_pct"]),
    }
    assert gpu["severity"]["busy_pct"] == "ok"


def test_severity_trips_on_the_ratified_boundaries(agent_snapshot) -> None:
    """Values pushed onto the table's edges in the real fixture, not on a hand-drawn snapshot."""
    snap = copy.deepcopy(agent_snapshot)
    snap["memory"]["used_bytes"] = snap["memory"]["total_bytes"] * 0.92   # mem.used_pct danger
    snap["cpu_temperature_celsius"] = 90.0                                 # cpu.temp_c danger
    for fs in snap["filesystems"]:
        fs["used_pct"] = 80.0                                              # disk.used_pct warn
    snap["gpus"][0]["busy_pct"] = 95.0                                     # gpu.busy_pct danger
    snap["gpus"][0]["gtt_used_bytes"] = 65e9   # whole-pool usage well past danger
    payload = parse_host_snapshot("driveripper", snap, make_collector(), now=1_000.0).payload
    assert payload["memory"]["severity"]["used_pct"] == "fail"
    assert payload["cpu"]["severity"]["temp_c"] == "fail"
    assert payload["disks"]["severity"]["worst_used_pct"] == "warn"
    assert payload["gpus"][0]["severity"]["busy_pct"] == "fail"
    assert payload["gpus"][0]["severity"]["mem_used_pct"] == "fail"


def test_a_value_the_agent_did_not_report_gets_no_colour(agent_snapshot) -> None:
    """Unknown is not ok: `severity: null` renders `n/a`, a green pass would be a lie."""
    snap = copy.deepcopy(agent_snapshot)
    snap["cpu_temperature_celsius"] = None
    snap["gpus"][0]["busy_pct"] = None
    snap["filesystems"] = []
    payload = parse_host_snapshot("driveripper", snap, make_collector(), now=1_000.0).payload
    assert payload["cpu"]["severity"]["temp_c"] is None
    assert payload["cpu"]["severity"]["usage_pct"] is None, "one scrape is not a rate"
    assert payload["gpus"][0]["severity"]["busy_pct"] is None
    assert payload["disks"]["severity"]["worst_used_pct"] is None
    # the numbers themselves are unchanged by the verdict sitting next to them
    assert payload["memory"]["used_pct"] == pytest.approx(
        snap["memory"]["used_bytes"] / snap["memory"]["total_bytes"] * 100)


def test_attaching_severity_renamed_or_removed_nothing(agent_snapshot) -> None:
    """Additive-only anchor, at the source: the flat fields the pages read are all still there."""
    payload = parse_host_snapshot("driveripper", agent_snapshot, make_collector(),
                                  now=1_000.0).payload
    expected = {
        "cpu": {"usage_pct", "modes_pct", "load1", "load5", "load15", "cores",
                "core_usage_pct", "freq_mhz", "freq_max_mhz", "context_switches_per_s",
                "forks_per_s", "procs_running", "procs_blocked", "temp_c"},
        "memory": {"total_bytes", "used_bytes", "available_bytes", "used_pct", "cached_bytes",
                   "buffers_bytes", "swap_total_bytes", "swap_used_bytes", "swap_used_pct",
                   "committed_bytes"},
        "disks": {"filesystems", "worst_used_pct", "worst_mount", "read_bytes_per_s",
                  "write_bytes_per_s", "reads_per_s", "writes_per_s", "util_pct", "devices"},
    }
    for section, keys in expected.items():
        assert keys <= set(payload[section]), f"{section} lost {keys - set(payload[section])}"
    for gpu in payload["gpus"]:
        assert {"busy_pct", "mem_used_pct"} <= set(gpu)


def test_fleet_summary_of_nothing_is_empty_not_zero() -> None:
    fleet = fleet_summary([])
    assert fleet["hosts_total"] == 0
    assert fleet["cpu_usage_pct_avg"] is None
    assert fleet["memory_used_bytes"] is None