dashboards/tests/test_severity.py

"""The single threshold table and its classifier.

Colour is a judgement, and a judgement made in four places drifts into four meanings.
These tests pin the boundaries (reaching a threshold trips it, in both directions), the
"unknown is not ok" rule, and the shape of the two payload helpers that put `severity`
next to a value.
"""

from __future__ import annotations

import json
import math

import pytest

from dashboards import severity
from dashboards.severity import FAIL, INVERTED, OK, THRESHOLDS, WARN, attach, classify, with_severity

# The ratified table (chat, 2026-10-16). Edit a number here and in THRESHOLDS together —
# that coupling is the point: thresholds are a contract with the pages, not a tuning knob.
RATIFIED = {
    "cpu.usage_pct": (75.0, 90.0),
    "mem.used_pct": (80.0, 92.0),
    "disk.used_pct": (80.0, 90.0),
    "disk.temp_c": (50.0, 70.0),
    "disk.percentage_used": (80.0, 100.0),
    "cpu.temp_c": (75.0, 90.0),
    "llm.kv_cache_usage_pct": (75.0, 90.0),
    "ups.charge_pct": (50.0, 20.0),
    "ups.battery_runtime": (600.0, 300.0),
    "ups.load_pct": (70.0, 90.0),
}


def _samples(metric: str) -> list[tuple[float, str]]:
    """(value, expected) walking across the boundaries of one metric, in its own direction."""
    warn, danger = THRESHOLDS[metric]
    middle = (warn + danger) / 2
    if metric in INVERTED:  # lower is worse: reaching warn trips it, going below danger fails
        step = 0.1
        return [
            (warn + step, OK),
            (warn, WARN),
            (middle, WARN),
            (danger, FAIL),
            (danger - 1.0, FAIL),
        ]
    return [
        (warn - 0.1, OK),
        (warn, WARN),
        (middle, WARN),
        (danger, FAIL),
        (danger + 1.0, FAIL),
    ]


@pytest.mark.parametrize("metric", sorted(THRESHOLDS))
def test_boundaries_hold_for_every_metric_in_the_table(metric: str) -> None:
    for value, expected in _samples(metric):
        assert classify(metric, value) == expected, f"{metric}={value} should be {expected}"


def test_completion_criteria_examples() -> None:
    assert classify("cpu.usage_pct", 95) == "fail"
    assert classify("cpu.usage_pct", 60) == "ok"
    assert classify("cpu.usage_pct", None) is None
    assert classify("ups.charge_pct", 15) == "fail"


def test_warn_edge_is_inclusive_and_just_below_is_still_good() -> None:
    assert classify("cpu.usage_pct", 74.9) == "ok"
    assert classify("cpu.usage_pct", 75) == "warn"
    assert classify("cpu.usage_pct", 75.0001) == "warn"
    assert classify("cpu.usage_pct", 89.9) == "warn"
    assert classify("cpu.usage_pct", 90) == "fail"


def test_inverted_metrics_read_the_other_way() -> None:
    """A flat battery and a flat runtime are the bad end, not the good one."""
    assert classify("ups.charge_pct", 100) == "ok"
    assert classify("ups.charge_pct", 50) == "warn"
    assert classify("ups.charge_pct", 19.9) == "fail"
    assert classify("ups.battery_runtime", 3600) == "ok"
    assert classify("ups.battery_runtime", 600) == "warn"
    assert classify("ups.battery_runtime", 120) == "fail"


def test_a_metric_is_only_in_one_direction() -> None:
    """Sanity: bounds are ordered for their direction, so an edit cannot invert a metric."""
    for metric, (warn, danger) in THRESHOLDS.items():
        assert math.isfinite(warn) and math.isfinite(danger), metric
        if metric in INVERTED:
            assert warn > danger, f"{metric} is inverted: warn must sit above danger"
        else:
            assert warn < danger, f"{metric} is not inverted: warn must sit below danger"


def test_inverted_is_a_subset_of_the_table() -> None:
    assert set(THRESHOLDS) >= INVERTED, "an inverted metric with no thresholds has no colour"


def test_the_ratified_numbers_are_the_ones_in_the_table() -> None:
    assert THRESHOLDS == RATIFIED


def test_unknown_metric_is_none_rather_than_an_exception() -> None:
    assert classify("gpu.vram_used_pct", 99) is None
    assert classify("nope.nope", 0) is None
    assert classify("", 100) is None


def test_gpu_usage_carries_no_verdict_at_any_level() -> None:
    """Phase 06: these are LLM servers — a busy GPU is the job, not a fault.

    The table has **no `gpu.*` row at all** (not a raised ceiling): every utilisation
    level classifies to `None`, so the card renders uncoloured rather than inventing
    amber for the workload the machines exist to run. Deleting the row is not deleting
    the data — the `gpu.busy_pct@N` series are still stored and charted.
    """
    assert not any(metric.startswith("gpu.") for metric in THRESHOLDS)
    assert classify("gpu.busy_pct", 0) is None
    assert classify("gpu.busy_pct", 99) is None
    assert classify("gpu.mem_used_pct", 99) is None
    assert classify("gpu.mem_used_pct", 100) is None
    assert classify("gpu.busy_pct@0", 95) is None, "streams inherit the absent row: nothing"


@pytest.mark.parametrize("metric", sorted(THRESHOLDS))
def test_none_is_never_ok(metric: str) -> None:
    assert classify(metric, None) is None


@pytest.mark.parametrize("metric", sorted(THRESHOLDS))
@pytest.mark.parametrize("value", [float("nan"), float("inf"), float("-inf")])
def test_non_finite_values_are_treated_as_missing(metric: str, value: float) -> None:
    assert classify(metric, value) is None


def test_a_drive_temperature_and_its_wear_are_the_only_disk_verdicts() -> None:
    """Phase 09: the two `/disks` numbers worth a colour, with the bounds the phase set.

    A spinning drive at 70 °C is a fire and an SSD that has consumed its rated wear is a
    replacement, so both trip red; the amber in between is the "watch it" band. An SSD with
    no temperature sensor has **no value**, and therefore no verdict — the absence of a
    number is not a healthy reading (`classify(metric, None)` stays `None`).
    """
    assert classify("disk.temp_c", 49.9) == OK
    assert classify("disk.temp_c", 50) == WARN
    assert classify("disk.temp_c", 69.9) == WARN
    assert classify("disk.temp_c", 70) == FAIL
    assert classify("disk.temp_c", 75) == FAIL
    assert classify("disk.temp_c", None) is None
    assert classify("disk.percentage_used", 79.9) == OK
    assert classify("disk.percentage_used", 80) == WARN
    assert classify("disk.percentage_used", 100) == FAIL
    assert classify("disk.percentage_used", None) is None


def test_sector_counters_and_power_on_hours_are_never_graded() -> None:
    """A non-zero sector count is an alarm, not a threshold breach (`/llm`'s chip rule).

    Colouring 0 reallocated sectors green would call a drive healthy because a number is
    small, and colouring 1 amber would make the chip redundant: the page shows an alarm on
    **presence** and says nothing at all at zero. `power_on_hours` is an odometer with no bad
    value. So none of them may ever grow a row in `THRESHOLDS`.
    """
    counters = ("disk.reallocated", "disk.pending", "disk.uncorrectable",
                "disk.power_on_hours", "disk.health_ok")
    assert set(counters).isdisjoint(THRESHOLDS), "a counter is not a graded number"
    for metric in counters:
        assert classify(metric, 0) is None
        assert classify(metric, 1) is None
        assert classify(metric, 99_000) is None
        assert classify(f"{metric}@sda", 1) is None


def test_per_device_streams_inherit_their_base_metric() -> None:
    assert classify("disk.used_pct@/srv", 95) == "fail"
    assert classify("disk.used_pct@/srv", 10) == "ok"
    assert classify("cpu.usage_pct@core7", 80) == "warn"
    assert classify("disk.temp_c@nvme0n1", 71) == "fail"
    assert classify("disk.percentage_used@nvme0n1", 80) == "warn"
    assert classify("mem.used_pct@any", 95) == "fail"


def test_vocabulary_is_exactly_ok_warn_fail_or_none() -> None:
    seen = {classify(metric, value) for metric in THRESHOLDS for value, _ in _samples(metric)}
    assert seen == {OK, WARN, FAIL}
    inverted = severity.INVERTED
    assert inverted == {"ups.charge_pct", "ups.battery_runtime"}


def test_attach_pairs_a_plain_number_with_its_verdict() -> None:
    assert attach("cpu.usage_pct", 95) == {"value": 95.0, "severity": "fail"}
    assert attach("mem.used_pct", 10) == {"value": 10.0, "severity": "ok"}
    assert attach("ups.load_pct", 70) == {"value": 70.0, "severity": "warn"}


def test_attach_keeps_the_value_when_it_has_no_opinion() -> None:
    assert attach("cpu.usage_pct", None) == {"value": None, "severity": None}
    assert attach("telemetry.mystery", 42) == {"value": 42.0, "severity": None}


def test_attach_never_puts_a_non_finite_number_in_a_payload() -> None:
    for value in (float("nan"), float("inf"), float("-inf")):
        assert attach("cpu.usage_pct", value) == {"value": None, "severity": None}
    assert json.dumps(attach("cpu.usage_pct", float("nan")))  # would raise otherwise


def test_with_severity_is_additive_on_a_llm_shaped_cell() -> None:
    cell = {"value": 82.0, "source": "llamacpp:kv_cache_usage_ratio", "derived": False}
    out = with_severity("llm.kv_cache_usage_pct", cell)
    assert out["source"] == "llamacpp:kv_cache_usage_ratio"
    assert out["derived"] is False
    assert out["value"] == 82.0
    assert out["severity"] == "warn"
    assert cell == {"value": 82.0, "source": "llamacpp:kv_cache_usage_ratio", "derived": False}, \
        "the caller's mapping must not be mutated"


def test_with_severity_handles_missing_and_non_numeric_cells() -> None:
    # purely additive: a missing cell gains `severity` and nothing else
    assert with_severity("llm.kv_cache_usage_pct", None) == {"severity": None}
    assert with_severity("llm.kv_cache_usage_pct", {"value": None}) == {"value": None, "severity": None}
    assert with_severity("llm.decode_tps", {"value": "n/a", "source": None}) == {
        "value": "n/a", "source": None, "severity": None,
    }
    # a value the table has no thresholds for still keeps its other keys
    out = with_severity("llm.ttft_seconds", {"value": 1.2, "source": "computed"})
    assert out == {"value": 1.2, "source": "computed", "severity": None}