dashboards/tests/test_ups_contract.py

"""The payload contract behind the UPS hero row and battery glyph (phase 05, task 01).

Phase 05 puts a battery glyph and three big colour-coded numbers — charge %, load %, runtime
left — at the top of every UPS card. Nothing is rendered here until the data is proven
present, for the same reason phase 03 pinned `/system` first: a card that shows `n/a` gets
"fixed" by inventing a field, and an invented field is a number the driver never reported
(`AGENTS.md` invariant 2, rule 7).

What makes this file different from `tests/test_nut.py` is *where the variables come from*.
That file's `VAR_ROWS` is a hand-written table; this one drives
`tests/fixtures/nut_apc.vars` and `nut_cyberpower.vars` — verbatim `LIST UPS` / `LIST VAR` /
`LIST CMD` replies recorded from the real upsd by `./scripts/capture-fixtures.sh` — because
the two batteries disagree in ways a typed table never would:

* the APC's `battery.date` is the factory placeholder `2001/09/25` while its
  `battery.mfr.date` is the real one (2021/06/24 → a battery past its window), and the
  separator is `/`, not the `-` every other date in this app uses;
* the CyberPower answers `battery.mfr.date: CPS`, which is not a date and must stay "unknown"
  rather than become a 0-year-old or 56-year-old battery;
* the driver's own `identity.model` is `GX1500U` while the operator's label — the thing on
  the card header — is `PR1500LCDRT2U`;
* **neither** driver exports `battery.temperature` **or** `battery.charger.status`, so the
  glyph's on-battery state has to come from `on_battery` and `charger_status` must be
  optional;
* only the CyberPower exports `output.voltage`.

The path table itself lives in `tests/ups_contract.py` so the live suite
(`tests/test_live_fleet.py::test_ups_hero_fields_are_present_on_the_real_batteries`) asserts
the same list against real hardware and the two cannot drift apart.

Two decisions the task asked to be written down, both asserted below:

* **The glyph derives its state from `charger_status` + `on_battery`.** The NUT flags are
  already parsed server-side into `states[]` (`UPS_STATE_LABELS` / `CHARGER_STATUS`);
  re-parsing `ups.status` in JavaScript duplicates that map and lets it disagree with the API.
* **Runtime prose stays `Dash.duration()`-shaped** (`10m 21s`, `1h 32m`) through the page's
  existing `runtimeText()`, and a runtime nobody measured is `Dash.naText` — `n/a`, never `0s`.
"""
from __future__ import annotations

import re
from collections.abc import Iterator
from pathlib import Path
from typing import Any

import pytest
from fastapi.testclient import TestClient

from dashboards.app import create_app
from dashboards.config import Settings
from dashboards.severity import classify
from dashboards.sources import nut
from dashboards.store import SampleStore
from ups_contract import (
    HERO_FIELDS,
    UPS_SHAPES,
    assert_ups_hero_contract,
    parse_capture,
    with_vars,
)

#: Deterministic "now" for the age maths (same instant `tests/test_nut.py` uses).
NOW = 1_767_225_600.0


# --------------------------------------------------------------------------- #
# Fixtures
# --------------------------------------------------------------------------- #
@pytest.fixture(scope="module")
def captures() -> dict[str, nut.UpsReading]:
    """Each real UPS as the driver reported it — the shape the hero row must tolerate."""
    return {name: parse_capture(name) for name in UPS_SHAPES}


@pytest.fixture(scope="module")
def hero_cards(captures: dict[str, nut.UpsReading]) -> dict[str, dict[str, Any]]:
    """`payload()` of each capture: exactly what `/api/ups/snapshot` serves per UPS."""
    return {name: nut.payload(reading, now=NOW) for name, reading in captures.items()}


@pytest.fixture
def client(tmp_path: Path) -> Iterator[TestClient]:
    """A running app, so the page-side assertions run on the bytes a browser would get.

    There is no JS test runner (a locked anchor), so what can be checked is the served
    script: which helper formats the runtime, and what it returns when there is no runtime.
    """
    db = str(tmp_path / "ups-contract.sqlite3")
    store = SampleStore(db)
    app = create_app(Settings(db_path=db), store)
    test_client = TestClient(app)
    test_client.__enter__()
    try:
        yield test_client
    finally:
        test_client.__exit__(None, None, None)
        store.close()


def _code(body: str) -> str:
    """An asset with its comments stripped — prose may *name* a flag the code must not parse."""
    body = re.sub(r"/\*.*?\*/", " ", body, flags=re.S)
    return re.sub(r"^\s*//.*$", " ", body, flags=re.M)


def _function(js: str, name: str) -> str:
    """One top-level function's source, from its signature to its closing brace at column 0."""
    start = js.index(f"function {name}(")
    return js[start:js.index("\n}", start) + 2]


def _na_text(client: TestClient) -> str:
    """The app's own `n/a`, read out of `app.js` rather than repeated here."""
    match = re.search(r"const naText = '([^']*)';", client.get("/static/app.js").text)
    assert match is not None, "app.js no longer defines the shared naText"
    return match.group(1)


# --------------------------------------------------------------------------- #
# The captures themselves are real, and they are the ones the contract names
# --------------------------------------------------------------------------- #
def test_every_captured_ups_is_in_the_contract() -> None:
    """A third UPS on the wire must join `UPS_SHAPES`, or its driver goes uncontracted."""
    on_disk = {path.stem.removeprefix("nut_")
               for path in (Path(__file__).parent / "fixtures").glob("nut_*.vars")}
    assert on_disk == set(UPS_SHAPES)


def test_each_capture_is_a_real_upsd_reply_not_an_empty_stub(captures: dict[str, nut.UpsReading],
                                                            hero_cards: dict[str, dict[str, Any]]) -> None:
    """50+ variables, a status, a description, advertised commands.

    Without this, a truncated fixture would satisfy "the key exists" and the contract would
    prove nothing: the whole point of capturing the wire is that reality has 50 variables in
    it, not the 15 anybody would type from memory.
    """
    for name, reading in captures.items():
        card = hero_cards[name]
        assert len(reading.vars) > 40, f"{name}: {len(reading.vars)} vars — this is not a capture"
        assert reading.commands, f"{name}: no CMD lines recorded"
        assert card["description"], f"{name}: LIST UPS gave no description"
        assert card["status"], f"{name}: ups.status is missing — every state would be unknown"
        assert card["identity"]["model"], f"{name}: no model to put beside the numbers"


# --------------------------------------------------------------------------- #
# Every field the hero row and the glyph read
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("name", UPS_SHAPES)
def test_every_hero_and_glyph_field_is_served_by_a_real_driver(name: str,
                                                               hero_cards: dict[str, dict[str, Any]]) -> None:
    assert_ups_hero_contract(hero_cards[name])


def test_the_hero_paths_come_from_the_table_not_from_memory(hero_cards: dict[str, dict[str, Any]]) -> None:
    """The table is the contract: every listed path is under `numbers`/`severity`/`health` or
    one of the three glyph fields, and the three hero numbers each travel with a verdict."""
    assert set(HERO_FIELDS) >= {"charge %", "load %", "runtime left", "glyph state", "card header"}
    for label, paths in HERO_FIELDS.items():
        assert paths, f"{label} lists no payload path"
    for card in hero_cards.values():
        for metric in ("charge_pct", "load_pct", "battery_runtime"):
            assert card["severity"][metric] == classify(f"ups.{metric}", card["numbers"][metric]), (
                f"{metric}: the card colour does not match severity.py"
            )


@pytest.mark.parametrize("name", UPS_SHAPES)
def test_the_hero_numbers_are_plausible_on_the_real_batteries(name: str,
                                                              hero_cards: dict[str, dict[str, Any]]) -> None:
    """Not just present: sane. 273 V on the battery line is the classic unit bug here."""
    numbers = hero_cards[name]["numbers"]
    assert 0.0 <= numbers["charge_pct"] <= 100.0
    assert 0.0 <= numbers["load_pct"] <= 200.0, "over 100 % is legal (overloaded), 0-200 is the honest band"
    assert numbers["battery_runtime"] > 0, "a runtime of 0 while online means the estimate is not reported"
    assert 12.0 <= numbers["battery_voltage"] <= 60.0, "these are 24 V battery packs, not line voltage"


# --------------------------------------------------------------------------- #
# The real-world shape: what the drivers do *not* report
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("name", UPS_SHAPES)
def test_neither_real_driver_reports_a_battery_temperature(name: str,
                                                          hero_cards: dict[str, dict[str, Any]]) -> None:
    """`None`, never `0 °C` — and never a key the page has to guess at (invariant 2)."""
    card = hero_cards[name]
    assert "battery_temperature" in card["numbers"]
    assert card["numbers"]["battery_temperature"] is None
    check = next(c for c in card["health"]["checks"] if c["name"] == "Battery temperature")
    assert check["status"] == "unknown"
    assert check["value"] is None


def test_the_apc_date_is_a_placeholder_and_its_mfr_date_is_the_real_one(
        captures: dict[str, nut.UpsReading], hero_cards: dict[str, dict[str, Any]]) -> None:
    """`battery.date` means nothing on this APC; `battery.mfr.date` is the fact (AGENTS.md)."""
    apc = captures["apc"]
    # the driver's own separator (`2001/09/25`), normalised here so the assertion is about the
    # date and not about whether this NUT build used slashes
    assert apc.vars["battery.date"].replace("/", "-") == "2001-09-25", \
        "the factory placeholder this driver ships"
    assert (apc.get("battery_date") or "").replace("/", "-") == "2021-06-24", \
        "VAR_MAP must prefer battery.mfr.date over the placeholder"
    age = next(c for c in hero_cards["apc"]["health"]["checks"] if c["name"] == "Battery age")
    assert age["value"] is not None and 4.0 <= age["value"] < 6.0, age
    assert age["status"] == "warn", "a 2021 battery is past the replacement window; the card says so"


def test_the_cyberpower_date_is_not_a_date_so_its_age_stays_unknown(
        captures: dict[str, nut.UpsReading], hero_cards: dict[str, dict[str, Any]]) -> None:
    """`battery.mfr.date: CPS` is a vendor string. Unknown, never an age invented from it."""
    cyberpower = captures["cyberpower"]
    assert cyberpower.vars["battery.mfr.date"] == "CPS"
    age = next(c for c in hero_cards["cyberpower"]["health"]["checks"] if c["name"] == "Battery age")
    assert age["status"] == "unknown" and age["value"] is None


def test_the_card_header_names_the_box_the_operator_labels_it(
        hero_cards: dict[str, dict[str, Any]]) -> None:
    """The driver calls the CyberPower `GX1500U`; the UPS on the shelf is a PR1500LCDRT2U."""
    card = hero_cards["cyberpower"]
    assert card["identity"]["model"] == "GX1500U"
    assert "PR1500LCDRT2U" in card["description"], "LIST UPS' description is the operator's label"
    assert hero_cards["apc"]["identity"]["mfr"] == "American Power Conversion"


def test_the_two_drivers_disagree_about_what_they_publish(hero_cards: dict[str, dict[str, Any]]) -> None:
    """One `numbers` dict, two drivers: a missing variable is `None` on one, a number on the other."""
    assert hero_cards["cyberpower"]["numbers"]["output_voltage"] is not None
    assert hero_cards["apc"]["numbers"]["output_voltage"] is None, "no output.voltage on the apc driver"
    assert hero_cards["cyberpower"]["numbers"]["realpower"] is not None
    assert hero_cards["apc"]["numbers"]["realpower"] is None, "only ups.realpower.nominal there"


# --------------------------------------------------------------------------- #
# The three verdicts the hero row paints with
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("charge,expected", [
    (100, "ok"),
    (45, "warn"),
    (15, "fail"),
])
def test_charge_severity_trips_the_inverted_threshold(captures: dict[str, nut.UpsReading],
                                                     charge: int, expected: str) -> None:
    """Inverted: a *smaller* number is worse, and `severity.py` owns that direction."""
    card = nut.payload(with_vars(captures["apc"], **{"battery.charge": str(charge)}), now=NOW)
    assert card["severity"]["charge_pct"] == expected
    assert card["numbers"]["charge_pct"] == float(charge), "the verdict is additive, not a replacement"


@pytest.mark.parametrize("runtime,expected", [
    (3600, "ok"),
    (400, "warn"),
    (200, "fail"),
])
def test_runtime_severity_trips_the_inverted_threshold(captures: dict[str, nut.UpsReading],
                                                       runtime: int, expected: str) -> None:
    """Seconds of runtime left: 10 min is amber, 5 min is red, an hour is fine."""
    card = nut.payload(with_vars(captures["apc"], **{"battery.runtime": str(runtime)}), now=NOW)
    assert card["severity"]["battery_runtime"] == expected
    assert card["numbers"]["battery_runtime"] == float(runtime)


def test_a_runtime_the_driver_cannot_estimate_has_no_verdict(captures: dict[str, nut.UpsReading]) -> None:
    """No number → no colour → `n/a`. `None` in, `None` out, never a green pass."""
    card = nut.payload(with_vars(captures["cyberpower"], **{"battery.runtime": None}), now=NOW)
    assert card["numbers"]["battery_runtime"] is None, (
        "`battery.runtime.low` is a shutdown threshold, not time left — borrowing it for the "
        "hero would print a confident 5m for a battery that may have 30"
    )
    assert card["severity"]["battery_runtime"] is None
    check = next(c for c in card["health"]["checks"] if c["name"] == "Runtime estimate")
    assert check["status"] == "unknown"
    # the threshold itself is still published, under its own name, for whoever wants it
    assert card["numbers"]["battery_runtime_low"] == 300.0


def test_load_severity_trips_the_ordinary_threshold(captures: dict[str, nut.UpsReading]) -> None:
    """Load is not inverted: 90 % of rated capacity is the red end, unlike charge/runtime."""
    for load, expected in (("35", "ok"), ("70", "warn"), ("95", "fail")):
        card = nut.payload(with_vars(captures["cyberpower"], **{"ups.load": load}), now=NOW)
        assert card["severity"]["load_pct"] == expected, f"load {load} %"


# --------------------------------------------------------------------------- #
# The glyph's state: `charger_status` + `on_battery`, never a flag re-parse in JS
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("charger,status,expected_on_battery,label", [
    ("ol", "OL", False, "online"),
    ("full", "OL", False, "fully charged"),
    ("chg", "OL CHRG", False, "charging"),
    ("dchg", "OB", True, "discharging"),
])
def test_the_glyph_states_are_published_as_fields(captures: dict[str, nut.UpsReading], charger: str,
                                                  status: str, expected_on_battery: bool,
                                                  label: str) -> None:
    """The four states a battery glyph draws, from the fields the API already sends."""
    card = nut.payload(with_vars(captures["apc"], **{"battery.charger.status": charger,
                                                     "ups.status": status}), now=NOW)
    assert card["charger_status"] == charger
    assert card["on_battery"] is expected_on_battery
    assert label in [state["label"] for state in card["states"]]


def test_a_driver_that_capitalises_the_charger_status_is_still_a_known_state(
        captures: dict[str, nut.UpsReading]) -> None:
    """`payload()` normalises the token to the `CHARGER_STATUS` keys.

    `states()` and `on_battery` already lowercase what they compare; if the field the glyph
    reads kept the driver's casing, `charger_status === "full"` would miss on a driver that
    answers `FULL` and the battery would stop drawing its state.
    """
    card = nut.payload(with_vars(captures["apc"], **{"battery.charger.status": " FULL "}), now=NOW)
    assert card["charger_status"] == "full"
    assert "fully charged" in [state["label"] for state in card["states"]]
    assert_ups_hero_contract(card)


def test_full_charge_on_line_never_claims_the_battery_is_charging(
        captures: dict[str, nut.UpsReading]) -> None:
    """`charger_status == "full"` with `on_battery == False`: the glyph is full, not filling.

    "Charging" is the word an operator reads as "it was flat recently", so a driver that
    reports a rested, floating, fully-charged battery must not produce it anywhere.
    """
    card = nut.payload(with_vars(captures["apc"], **{"battery.charger.status": "full",
                                                     "ups.status": "OL"}), now=NOW)
    assert card["charger_status"] == "full"
    assert card["on_battery"] is False
    words = [state["label"] for state in card["states"]] + [card["status"] or "",
                                                            card["charger_status"] or ""]
    assert "charging" not in words, f"a full, online UPS is not charging: {words}"
    assert not any("charging" in word for word in words), words


def test_on_battery_makes_the_glyphs_on_battery_state_available(captures: dict[str, nut.UpsReading],
                                                               ) -> None:
    """Both routes to on-battery work, including the one with no `ups.status` at all."""
    from_flag = nut.payload(with_vars(captures["apc"], **{"ups.status": "OB DISCHRG",
                                                          "battery.charger.status": None}), now=NOW)
    assert from_flag["on_battery"] is True
    assert {"id": "OB", "label": "on battery", "severity": "warn",
            "detail": "Running on battery"} in from_flag["states"]
    assert from_flag["charger_status"] is None, "the APC is proof this field is optional"

    from_charger = nut.payload(with_vars(captures["apc"], **{"ups.status": None,
                                                            "battery.charger.status": "dchg"}), now=NOW)
    assert from_charger["on_battery"] is True
    assert "discharging" in [state["label"] for state in from_charger["states"]]


@pytest.mark.parametrize("name", UPS_SHAPES)
def test_the_real_captures_omit_charger_status_so_on_battery_carries_the_state(
        name: str, captures: dict[str, nut.UpsReading],
        hero_cards: dict[str, dict[str, Any]]) -> None:
    """The assumption, recorded against hardware: `battery.charger.status` is not there.

    Neither usbhid-ups driver exports the NUT 2.8 rich status, so a glyph that only looked at
    `charger_status` would render "unknown" on two healthy batteries. `on_battery` is derived
    from `ups.status` server-side and is the field that must drive the glyph here.
    """
    assert "battery.charger.status" not in captures[name].vars
    card = hero_cards[name]
    assert card["charger_status"] is None
    assert card["on_battery"] is False
    labels = [state["label"] for state in card["states"]]
    assert "online" in labels, f"the glyph still needs a state to show: {labels}"


def test_the_page_does_not_re_parse_nut_status_flags(client: TestClient) -> None:
    """`page-ups.js` may render `states[]` and the raw `status` chip; it may not interpret flags.

    `UPS_STATE_LABELS` and `CHARGER_STATUS` in `sources/nut.py` are the map. A copy of it in
    JavaScript is a second opinion about the wire, and one of the two will be wrong.
    """
    js = _code(client.get("/static/page-ups.js").text)
    assert "status.split(" not in js, "the page is parsing ups.status itself"
    for flag in ("OL", "OB", "LB", "RB", "CHRG", "DISCHRG", "FUB", "CAL", "OVER", "NULL"):
        assert f'"{flag}"' not in js and f"'{flag}'" not in js, f"the page special-cases the {flag} flag"
    assert "item.states" in js, "the states the API labelled are what the card renders"


# --------------------------------------------------------------------------- #
# The page-side formatting of "no runtime"
# --------------------------------------------------------------------------- #
def test_runtime_of_none_is_na_not_zero_seconds(client: TestClient) -> None:
    """`runtimeText(null)` is the app's `n/a`. `0s` would be a different, alarming sentence.

    Asserted on the served script because there is no JS test runner — the promise under test
    is that the runtime cell goes through the one helper and that the helper's unknown branch
    returns the shared text before any number can be formatted.
    """
    js = client.get("/static/page-ups.js").text
    na = _na_text(client)
    assert na == "n/a"

    body = _function(js, "runtimeText")
    guard = re.search(r"seconds == null.*?return\s+([^;]+);", body)
    assert guard is not None, "runtimeText() no longer guards against an absent runtime"
    token = guard.group(1).strip()
    rendered = na if token == "naText" else token.strip("'\"")
    assert rendered == na, f"runtimeText() invents its own unknown text: {token}"

    # the unknown branch is first: nothing may format a number before it
    assert body.index("seconds == null") < body.index("s`;"), "null could reach the number formatter"
    assert "return '0s'" not in body

    # and the card's runtime cell is written through that helper, from the hero's own number
    assert re.search(r"runtimeText\([^)]*battery_runtime\s*\)", js), (
        "the runtime cell no longer formats numbers.battery_runtime through runtimeText()"
    )


def test_the_page_reads_the_keys_the_contract_lists(client: TestClient) -> None:
    """The hero row's keys are the contract's keys, on the served script and nowhere newer.

    Task 02/03 may restructure the card, but a rename here (`n.chargeRemaining`) means the
    page and `/api/ups/snapshot` no longer speak, and the only symptom is a card of `n/a`.
    """
    js = client.get("/static/page-ups.js").text
    assert re.search(r"item\.numbers\b", js), "the card no longer reads the payload's numbers object"
    for key in ("charge_pct", "load_pct", "battery_runtime", "battery_voltage",
                "battery_temperature", "line_voltage"):
        assert f".{key}" in js, f"the page never reads numbers.{key} — renamed or dropped"
    assert "item.health" in js, "the card header still reads health.overall / health.counts"