dashboards/tests/test_nut.py

"""NUT: the wire protocol, the health rules, and the gates around the only write path.

The protocol tests run against a fake upsd over a real socket — this client speaks a
line protocol with quoting rules, and a mock that agrees with my memory of it proves
nothing. The health rules and the command gates are pure functions and are asserted
directly, because those are what decide whether a battery test gets sent.
"""
from __future__ import annotations

import asyncio
import time
from pathlib import Path

import pytest
import pytest_asyncio

from dashboards.sources import nut
from dashboards.sources.nut import NutClient, NutError, UpsReading, normalise_celsius
from dashboards.sources.nut_sim import NutSimulator, SimUps, from_env
from ups_contract import parse_capture

VAR_ROWS = {
    "battery.charge": "100",
    "battery.runtime": "3720",
    "battery.voltage": "27.3",
    "battery.temperature": "300.4",           # kelvin, as several drivers report it
    "battery.date": "2021/04/01",
    "battery.type": "Pb",
    "input.voltage": "121.8",
    "input.voltage.nominal": "120",
    "input.transfer.low": "92",
    "input.transfer.high": "139",
    "output.voltage": "120.9",
    "ups.load": "34",
    "ups.realpower.nominal": "900",
    "ups.load.high": "1500",
    "ups.status": "OL CHRG",
    "ups.mfr": "Cyber Power Systems",
    "ups.model": "PR1500LCDRT2U",
    "battery.test.result": "PASSED",
    "battery.test.date": "2026-09-01",
}

CMDS = ["beeper.mute", "test.panel.start", "test.battery.start.quick",
        "test.battery.start.deep", "load.off"]


class FakeUpsd:
    """Just enough of upsd's protocol to exercise the client, over a real socket."""

    def __init__(self, *, require_auth: bool = False, close_early: bool = False,
                 hang: bool = False, partial_list: str | None = None,
                 deny: tuple[str, ...] = ()) -> None:
        self.require_auth = require_auth
        self.close_early = close_early
        self.hang = hang
        # Answer this `LIST` kind with `BEGIN` + one row and then go silent: the row is already
        # in the client's buffer when its read times out, which is the exact state a slow upsd
        # leaves a shared connection in.
        self.partial_list = partial_list
        # Answer these `LIST` kinds with a complete `ERR` line — a refusal is the whole reply,
        # so a client may (and must) keep using the connection afterwards.
        self.deny = deny
        self.received: list[str] = []
        self.authenticated = False
        self._server: asyncio.AbstractServer | None = None

    async def start(self) -> int:
        self._server = await asyncio.start_server(self._handle, "127.0.0.1", 0)
        await self._server.start_serving()
        return int(self._server.sockets[0].getsockname()[1])

    async def stop(self) -> None:
        if self._server:
            self._server.close()
            await self._server.wait_closed()

    async def _handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
        if self.close_early:
            writer.close()
            return
        while True:
            raw = await reader.readline()
            if not raw:
                break
            line = raw.decode().strip()
            self.received.append(line)
            verb, _, rest = line.partition(" ")
            await self._respond(writer, verb, rest)

    async def _respond(self, writer: asyncio.StreamWriter, verb: str, rest: str) -> None:
        def send(text: str) -> None:
            writer.write((text + "\n").encode())

        if verb == "LIST":
            kind, _, target = rest.partition(" ")
            if kind in self.deny:
                send("ERR ACCESS-DENIED")
            elif kind == self.partial_list:
                # One row of the real shape, then silence: no END, so the client times out
                # with this row sitting in its socket buffer.
                send(f"BEGIN LIST {kind}")
                send(f'VAR "{_strip(target)}" "ups.status" "OL"' if kind == "VAR"
                     else f'{kind} "{_strip(target)}" "beeper.mute"')
                return
            elif kind == "UPS":
                send("BEGIN LIST UPS")
                send('UPS "cyberpower" "Cyber Power Systems PR1500LCDRT2U"')
                send("END LIST UPS")
            elif kind == "VAR":
                send("BEGIN LIST VAR")
                for key, value in VAR_ROWS.items():
                    send(f'VAR "{_strip(target)}" "{key}" "{value}"')
                send("END LIST VAR")
            elif kind == "CMD":
                send("BEGIN LIST CMD")
                for command in CMDS:
                    send(f'CMD "{_strip(target)}" "{command}"')
                send("END LIST CMD")
            elif kind == "RW":
                send("BEGIN LIST RW")
                send(f'RW "{_strip(target)}" "battery.energysave" "0"')
                send("END LIST RW")
            else:
                send("ERR UNKNOWN-COMMAND")
        elif verb == "USERNAME":
            self.user = rest
            send("OK")
        elif verb == "PASSWORD":
            # The password has to matter: a fake that authenticates on the username alone
            # cannot express a wrong-credential refusal, which is the commonest reason the
            # command buttons stay dead after someone fills in NUT_PASSWORD.
            self.authenticated = not self.require_auth or (
                getattr(self, "user", "") == "monitor" and rest == "pw"
            )
            send("OK" if self.authenticated else "ERR ACCESS-DENIED")
        # No USER/PASS branches on purpose: real upsd 2.8.5 has neither verb, and answering
        # ERR UNKNOWN-COMMAND here (the fallthrough below) is what catches a client that
        # starts sending them again.
        elif verb == "INSTCMD":
            if self.require_auth and not self.authenticated:
                send("ERR ACCESS-DENIED")
            elif self.hang:
                await asyncio.sleep(30)
            else:
                send("OK Send shorthand command successful")
        elif verb == "LOGOUT":
            send("OK Goodbye")
        else:
            send("ERR UNKNOWN-COMMAND")


def _strip(value: str) -> str:
    return value.strip().strip('"')


@pytest_asyncio.fixture
async def upsd():
    server = FakeUpsd()
    port = await server.start()
    yield server, port
    await server.stop()


async def test_list_ups_parses_names_and_descriptions(upsd) -> None:
    _server, port = upsd
    async with NutClient("127.0.0.1", port) as client:
        ups_list = await client.list_ups()
    assert ups_list[0]["name"] == "cyberpower"
    assert ups_list[0]["description"] == "Cyber Power Systems PR1500LCDRT2U"


async def test_list_vars_keeps_spaces_and_quotes(upsd) -> None:
    _server, port = upsd
    async with NutClient("127.0.0.1", port) as client:
        variables = await client.list_vars("cyberpower")
    assert variables["ups.mfr"] == "Cyber Power Systems"
    assert variables["battery.charge"] == "100"
    assert variables["ups.status"] == "OL CHRG"


async def test_list_cmds_returns_what_the_driver_offers(upsd) -> None:
    _server, port = upsd
    async with NutClient("127.0.0.1", port) as client:
        commands = await client.list_cmds("cyberpower")
    assert commands == CMDS


async def test_list_rw_vars_names_the_variables_not_the_ups(upsd) -> None:
    """`RW <ups> <variable>`: reading field 1 returns the UPS name as a writable variable."""
    _server, port = upsd
    async with NutClient("127.0.0.1", port) as client:
        rw = await client.list_rw_vars("cyberpower")
    assert rw == ["battery.energysave"]


async def test_instcmd_without_credentials_refuses_locally(upsd) -> None:
    """No socket write: a dashboard must not even try when it knows it cannot authorise."""
    _server, port = upsd
    async with NutClient("127.0.0.1", port) as client:
        with pytest.raises(NutError) as excinfo:
            await client.instcmd("cyberpower", "test.panel.start")
    assert excinfo.value.code == "ACCESS-DENIED"
    assert "INSTCMD" not in _server.received or not any("INSTCMD" in line for line in _server.received)


async def test_instcmd_with_credentials_succeeds() -> None:
    server = FakeUpsd(require_auth=True)
    port = await server.start()
    try:
        async with NutClient("127.0.0.1", port, username="monitor", password="pw") as client:
            assert await client.login() is True
            reply = await client.instcmd("cyberpower", "test.panel.start")
        assert "OK" in reply
        assert any(line.startswith("INSTCMD") for line in server.received)
    finally:
        await server.stop()


async def test_login_uses_the_verbs_upsd_2_8_5_answers_to() -> None:
    """The verbs are a protocol contract, not a style choice.

    `USER`/`PASS` are not in the deployed upsd's `HELP` list and it answers them
    `ERR UNKNOWN-COMMAND`, which made every login fail. This pins the pair that actually
    authenticates, unquoted, and fails if either legacy verb comes back.
    """
    server = FakeUpsd(require_auth=True)
    port = await server.start()
    try:
        async with NutClient("127.0.0.1", port, username="monitor", password="pw") as client:
            assert await client.login() is True
    finally:
        await server.stop()
    assert server.received[:2] == ["USERNAME monitor", "PASSWORD pw"]
    assert not any(line.split(" ")[0] in ("USER", "PASS") for line in server.received)


async def test_login_reports_a_rejected_password_as_an_upsd_error() -> None:
    """A bad credential must surface as the server's refusal, not a silent False."""
    server = FakeUpsd(require_auth=True)
    port = await server.start()
    try:
        async with NutClient("127.0.0.1", port, username="monitor", password="wrong") as client:
            with pytest.raises(NutError) as excinfo:
                await client.login()
    finally:
        await server.stop()
    assert excinfo.value.code == "ACCESS-DENIED"
    assert server.received[:1] == ["USERNAME monitor"]


async def test_upsd_rejection_carries_the_error_code() -> None:
    server = FakeUpsd(require_auth=True)
    port = await server.start()
    try:
        async with NutClient("127.0.0.1", port, username="nobody", password="pw") as client:
            with pytest.raises(NutError) as excinfo:
                await client.login()
        assert excinfo.value.code == "ACCESS-DENIED"
        assert "upsd.users" in str(excinfo.value) or "refused" in str(excinfo.value) or str(excinfo.value)
    finally:
        await server.stop()


async def test_connection_refused_is_reported_as_unreachable() -> None:
    """A closed port must read as 'upsd is not answering', never as an empty UPS list."""
    client = NutClient("127.0.0.1", 1, timeout=1.0)
    with pytest.raises(NutError) as excinfo:
        await client.connect()
    assert excinfo.value.code == "UNREACHABLE"
    # The reason is the sentence the operator actually reads (health record, alert, feed).
    assert "ConnectionRefusedError" in str(excinfo.value)


async def test_connect_timeout_names_the_timeout_instead_of_an_empty_parenthetical(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """`wait_for()` raises a TimeoutError with no message of its own.

    A firewalled port therefore used to print `… (TimeoutError: )`, which explains nothing.
    This is the exact state the fleet is in while the laptop cannot reach upsd, so the
    message has to carry the timeout it hit.
    """

    async def never_connects(*_args: object, **_kwargs: object) -> None:
        await asyncio.sleep(0.2)

    monkeypatch.setattr(asyncio, "open_connection", never_connects)
    client = NutClient("ups.example.net", 3051, timeout=0.01)
    with pytest.raises(NutError) as excinfo:
        await client.connect()
    assert excinfo.value.code == "UNREACHABLE"
    message = str(excinfo.value)
    assert "timed out after 0.01s" in message
    assert "TimeoutError" not in message


async def test_server_closing_early_is_reported() -> None:
    server = FakeUpsd(close_early=True)
    port = await server.start()
    try:
        async with NutClient("127.0.0.1", port, timeout=2.0) as client:
            with pytest.raises(NutError) as excinfo:
                await client.list_ups()
        assert excinfo.value.code == "CLOSED"
    finally:
        await server.stop()


async def test_a_silent_upsd_times_out_instead_of_hanging() -> None:
    server = FakeUpsd(hang=True)
    port = await server.start()
    try:
        async with NutClient("127.0.0.1", port, timeout=0.3) as client:
            with pytest.raises(NutError) as excinfo:
                await client._send("INSTCMD cyberpower beeper.mute")
                await client._read_line()
        assert excinfo.value.code == "TIMEOUT"
    finally:
        await server.stop()


async def test_a_reply_that_stops_early_makes_the_socket_refuse_the_next_command() -> None:
    """A stalled `LIST` is reported as the stall it was, not as a protocol mystery.

    This fleet's upsd answers every poll within `NUT_TIMEOUT_SECONDS` except every so often,
    and the lines it had already sent stay in the socket buffer. `collectors.NutMonitor`
    tolerates a failed `LIST CMD`/`LIST RW` and reads the next UPS on the same connection, so
    those leftovers used to be taken for that command's `BEGIN` and the poll died saying
    `expected 'BEGIN LIST VAR', got 'RW apc battery.charge.low "10"'` — a sentence about the
    wire protocol, when the truth was a 4-second stall two commands earlier (the pair of
    messages is in the dev `event` table, which is how this got noticed).
    """
    server = FakeUpsd(partial_list="CMD")
    port = await server.start()
    try:
        async with NutClient("127.0.0.1", port, timeout=0.3) as client:
            with pytest.raises(NutError) as stalled:
                await client.list_cmds("apc")
            assert stalled.value.code == "TIMEOUT"
            with pytest.raises(NutError) as next_command:
                await client.list_vars("apc")
    finally:
        await server.stop()
    assert next_command.value.code == "UNUSABLE"
    assert "did not answer" in str(next_command.value), "the cause is named, not the symptom"
    assert "LIST CMD" in str(next_command.value), "and it names the command that stalled"


async def test_an_instant_command_that_never_answers_spends_the_connection() -> None:
    """The guard is on the write path too, where it matters most.

    `instcmd` is the only thing in this app that dials a write. If upsd answers it after we
    gave up, that late `OK` sits on the socket, and a client allowed to ask another question
    would read a *different* command's success as the answer to it — the one mistake the four
    gates cannot survive, since it would print `OK` beside a `load.off` nobody confirmed.
    """
    server = FakeUpsd(hang=True)
    port = await server.start()
    try:
        async with NutClient("127.0.0.1", port, timeout=0.3,
                             username="monitor", password="pw") as client:
            with pytest.raises(NutError) as hung:
                await client.instcmd("apc", "beeper.mute")
            assert hung.value.code == "TIMEOUT"
            with pytest.raises(NutError) as next_command:
                await client.instcmd("apc", "load.off")
    finally:
        await server.stop()
    assert next_command.value.code == "UNUSABLE"
    assert "INSTCMD" in str(next_command.value)


async def test_an_upsd_error_line_leaves_the_connection_usable() -> None:
    """The other half: `ERR …` is a *complete* reply, so the tolerance in `collect()` stands.

    One UPS answering `LIST CMD` with `ACCESS-DENIED` must cost that UPS its command buttons
    and nothing else — the same poll keeps reading the other battery. A guard that poisoned the
    connection on a refusal would turn a per-UPS refusal into a dead UPS page.
    """
    server = FakeUpsd(deny=("CMD",))
    port = await server.start()
    try:
        async with NutClient("127.0.0.1", port, timeout=2.0) as client:
            with pytest.raises(NutError) as denied:
                await client.list_cmds("apc")
            assert denied.value.code == "ACCESS-DENIED"
            variables = await client.list_vars("apc")
    finally:
        await server.stop()
    assert variables["ups.status"] == "OL CHRG"


# ---- reading semantics ---------------------------------------------------- #

def reading(**overrides: str) -> UpsReading:
    values = dict(VAR_ROWS)
    values.update(overrides)
    return UpsReading(name="cyberpower", description="Cyber Power PR1500LCDRT2U",
                      vars=values, commands=list(CMDS))


def test_status_flags_and_state_chips() -> None:
    ups = reading()
    assert ups.status_flags == ["OL", "CHRG"]
    labels = {state["label"] for state in ups.states()}
    assert "online" in labels and "charging" in labels
    assert all(state["severity"] in ("ok", "warn", "fail", "info") for state in ups.states())


def test_unknown_status_flag_is_shown_not_dropped() -> None:
    ups = reading(**{"ups.status": "OL OFF"})
    ids = {state["id"] for state in ups.states()}
    assert "OFF" in ids, "an unrecognised-in-combination flag must still be shown, not swallowed"


def test_on_battery_is_derived_from_the_flag_not_from_charge() -> None:
    assert reading().on_battery is False
    assert reading(**{"ups.status": "OB"}).on_battery is True


def test_alarms_are_split_and_cleaned() -> None:
    ups = reading(**{"ups.alarm": "BATT.VERIFYFAIL, OVERLOAD "})
    assert ups.alarms == ["BATT.VERIFYFAIL", "OVERLOAD"]


def test_number_returns_none_for_absent_and_non_numeric() -> None:
    ups = reading()
    assert ups.number("battery.charge") == 100.0
    assert ups.number("does.not.exist") is None
    assert ups.number("ups.status") is None


@pytest.mark.parametrize("raw,expected", [
    (None, None), (27.4, 27.4), (300.4, 27.25), (150.0, 150.0), (151.0, -122.15),
])
def test_kelvin_is_detected_because_the_driver_cannot_be_trusted(raw, expected) -> None:
    result = normalise_celsius(raw)
    assert result == (None if expected is None else pytest.approx(expected, abs=0.01))


# ---- health assessment ---------------------------------------------------- #

def test_a_healthy_ups_reports_every_check_as_ok() -> None:
    health = nut.assess(reading(**{"battery.temperature": "27.4", "battery.mfr.date": "2026-01-01",
                                   "ups.status": "OL"}), now=1_767_225_600.0)
    assert health["overall"] == "healthy"
    assert health["counts"]["fail"] == 0 and health["counts"]["warn"] == 0
    assert any(check["name"] == "Power path" for check in health["checks"])


def test_on_battery_with_little_runtime_is_a_failure() -> None:
    health = nut.assess(reading(**{"ups.status": "OB", "battery.charge": "18",
                                   "battery.runtime": "90",
                                   "battery.temperature": "27.4"}), now=1_767_225_600.0)
    assert health["overall"] == "at risk"
    runtime = next(c for c in health["checks"] if c["name"] == "Runtime estimate")
    assert runtime["status"] in ("warn", "fail")


def test_old_battery_is_warned_about_with_the_actual_age() -> None:
    health = nut.assess(reading(**{"battery.temperature": "27.4", "battery.mfr.date": "2019-01-01"}),
                        now=1_767_225_600.0)
    age = next(c for c in health["checks"] if c["name"] == "Battery age")
    assert age["status"] == "warn"
    assert "2019" in age["detail"] or "7" in age["detail"]


def test_a_kelvin_temperature_is_not_reported_as_a_melting_battery() -> None:
    health = nut.assess(reading(), now=1_767_225_600.0)   # 300.4 in the fixture
    temp = next(c for c in health["checks"] if c["name"] == "Battery temperature")
    assert temp["status"] == "ok"
    assert "kelvin" in temp["detail"]
    assert 20 < temp["value"] < 35


def test_a_hot_battery_fails() -> None:
    health = nut.assess(reading(**{"battery.temperature": "52"}), now=1_767_225_600.0)
    temp = next(c for c in health["checks"] if c["name"] == "Battery temperature")
    assert temp["status"] == "fail"


def test_a_failed_self_test_is_surfaced() -> None:
    health = nut.assess(reading(**{"battery.test.result": "FAILED", "battery.temperature": "27.4"}),
                        now=1_767_225_600.0)
    check = next(c for c in health["checks"] if c["name"] == "Last self-test")
    assert check["status"] in ("warn", "fail")


def test_missing_test_history_is_unknown_not_ok() -> None:
    values = {key: value for key, value in VAR_ROWS.items()
                if not key.startswith("battery.test")}
    values["battery.temperature"] = "27.4"
    health = nut.assess(nut.UpsReading(name="cyberpower", vars=values, commands=[]),
                        now=1_767_225_600.0)
    check = next(c for c in health["checks"] if c["name"] == "Last self-test")
    assert check["status"] == "unknown"


# ---- the self-test word the hardware actually answers (phase 12 pin) ------- #

#: `ups.test.result` exactly as the two real UPSes answered it, captured from the running
#: upsd on **2026-10-19** from the dev box (which reaches driveripper's 3051 — the transport
#: phase 10 re-verified) with::
#:
#:     printf 'LIST VAR apc ups.test.result\nLIST VAR cyberpower ups.test.result\n' |
#:         nc <upsd-host> 3051
#:
#: These are the sentences the "Last self-test" check has to grade, and they are not the
#: words its pass tuple was written for (AGENTS.md rule 7 — verify against the deployed
#: upstream): the CyberPower has had a test run and answers `Done and passed`, the APC has
#: not and still answers the factory `No test initiated`. The same two words are what
#: `/api/summary` was putting in the Warnings panel on the capture date, one per UPS.
#: Re-captured with `./scripts/capture-fixtures.sh` the same day, so `nut_*.vars` carries
#: them verbatim; if a future capture changes one, the first test below fails and the word
#: map and this table get updated together.
LIVE_SELF_TEST_WORDS = {
    "apc": "No test initiated",
    "cyberpower": "Done and passed",
}


def self_test_check(health: dict) -> dict:
    return next(check for check in health["checks"] if check["name"] == "Last self-test")


def test_the_captured_self_test_words_are_what_the_fixtures_carry() -> None:
    """The capture and the verbatim fixture must not drift apart.

    `assess()` is asserted against `LIVE_SELF_TEST_WORDS` below, so a fixture re-captured
    after somebody runs a real battery test has to land in the same table — the day the two
    disagree, the verdict is being tested against a memory of the hardware instead of the
    hardware (`AGENTS.md` rule 7).
    """
    for ups, word in LIVE_SELF_TEST_WORDS.items():
        capture = (Path(__file__).parent / "fixtures" / f"nut_{ups}.vars").read_text()
        assert f'VAR {ups} ups.test.result "{word}"' in capture, (
            f"nut_{ups}.vars no longer answers the captured word {word!r} — re-run "
            f"./scripts/capture-fixtures.sh and update LIVE_SELF_TEST_WORDS with the diff"
        )
        assert parse_capture(ups).get("test_result") == word


def test_a_passed_self_test_reads_ok_and_not_a_warning() -> None:
    """The task-01 pin, now green: the CyberPower's pass word must read `ok`.

    Before phase 12 `assess()` compared the driver's word against
    `("ok", "passed", "pass", "completed")`, so `Done and passed` fell through to `warn` and
    the overview printed a warning for a battery that passed.
    """
    health = nut.assess(parse_capture("cyberpower"), now=1_767_225_600.0)
    check = self_test_check(health)
    assert check["status"] == "ok", (
        f"the driver reported {check['detail']!r} and it graded {check['status']!r}"
    )
    assert check["detail"].startswith(LIVE_SELF_TEST_WORDS["cyberpower"]), (
        "the driver's own sentence is the useful part of this line")


def test_a_self_test_that_was_never_initiated_is_not_a_warning() -> None:
    """The second half of the same pin: the APC's factory state is not a fault.

    A pack nobody has ever self-tested is a suggestion to run one, so it reads `unknown`
    with the unset branch's sentence instead of `warn`.
    """
    health = nut.assess(parse_capture("apc"), now=1_767_225_600.0)
    check = self_test_check(health)
    assert check["status"] == "unknown", (
        f"never tested ({check['detail']!r}) must read like the unset case, not "
        f"{check['status']!r}"
    )
    assert check["detail"] == nut.NEVER_TESTED_DETAIL
    assert "run a battery test" in check["detail"]


#: The whole `Last self-test` outcome map: the driver's word -> the verdict `assess()` gives.
#: The pass tier holds every pass word this fleet's drivers, the simulator, and the older
#: NUT drivers answer — including `Done and passed`, task 01's live capture and the one the
#: old tuple missed. The fail tier is unchanged: a test that ran and failed is the reason the
#: check exists. `unknown` holds both "no test has ever run" states. `warn` is the deliberate
#: fall-through for a word this map has never seen — a driver inventing a verdict is upstream
#: drift worth a line, and the check prints that word verbatim so it can be added.
SELF_TEST_WORD_MAP = {
    "ok": "ok",
    "passed": "ok",
    "pass": "ok",
    "completed": "ok",
    LIVE_SELF_TEST_WORDS["cyberpower"]: "ok",
    "FAILED": "fail",
    "fail": "fail",
    "Error": "fail",
    "aborted": "fail",
    "test incomplete": "fail",
    LIVE_SELF_TEST_WORDS["apc"]: "unknown",
    "": "unknown",
    "mystery": "warn",
    "DONE AND FAILED": "warn",
}


@pytest.mark.parametrize("word,expected", sorted(SELF_TEST_WORD_MAP.items()))
def test_the_self_test_check_grades_the_outcome_and_not_the_phrasing(word, expected) -> None:
    check = self_test_check(nut.assess(reading(**{"battery.test.result": word}),
                                       now=1_767_225_600.0))
    assert check["status"] == expected, f"{word!r} graded {check['status']!r}"
    if expected == "unknown":
        assert check["detail"] == nut.NEVER_TESTED_DETAIL
    else:
        assert check["detail"].startswith(word), "the raw word is what the operator reads"
        assert "2026-09-01" in check["detail"], "the test_date suffix stays on the line"


@pytest.mark.parametrize("variant", [
    "Done and passed", "DONE AND PASSED", "done and passed", "  Done and passed  ",
])
def test_the_live_pass_word_is_matched_case_and_whitespace_insensitively(variant) -> None:
    assert nut.self_test_outcome(variant) == "ok"
    check = self_test_check(nut.assess(reading(**{"battery.test.result": variant}),
                                       now=1_767_225_600.0))
    assert check["status"] == "ok"


def test_never_tested_says_the_same_sentence_as_no_variable_at_all() -> None:
    """`No test initiated` and a driver exporting nothing are one sentence, not two.

    Both are "run a battery test": a check that invented a second phrasing for the same
    nothing would be two warnings to learn instead of one thing to do.
    """
    values = {key: value for key, value in VAR_ROWS.items()
              if not key.startswith("battery.test")}
    unset = self_test_check(nut.assess(nut.UpsReading(name="apc", vars=values, commands=[]),
                                       now=1_767_225_600.0))
    never = self_test_check(nut.assess(reading(**{
        "battery.test.result": LIVE_SELF_TEST_WORDS["apc"]}), now=1_767_225_600.0))
    assert unset["status"] == never["status"] == "unknown"
    assert never["detail"] == unset["detail"] == nut.NEVER_TESTED_DETAIL


def test_no_word_the_real_drivers_answer_lands_in_warn() -> None:
    """Whatever the two boxes answer must be a word the map knows.

    `warn` is the fall-through, so a captured word reaching it means the overview is warning
    about a battery this code has simply never read before — the exact bug this phase fixes.
    """
    for ups, word in LIVE_SELF_TEST_WORDS.items():
        check = self_test_check(nut.assess(parse_capture(ups), now=1_767_225_600.0))
        assert check["status"] != "warn", f"{ups} answers {word!r} and it graded warn"
        assert check["status"] == nut.self_test_outcome(word)


def test_the_outcome_map_holds_only_verdicts_the_pages_render() -> None:
    """A typo'd tier in the table would reach the browser as an unknown CSS class."""
    assert set(nut.SELF_TEST_OUTCOMES.values()) <= {"ok", "fail", "unknown"}
    assert nut.self_test_outcome(None) == "unknown", "no variable is not a verdict"


def test_the_simulator_only_answers_words_the_map_grades() -> None:
    """The model's `PASSED` / `FAILED` were already in the right tiers (task 02 item 4).

    If the simulator ever learns a third word it goes into `SELF_TEST_OUTCOMES`; it gets no
    special case here, because a model the health check grades differently from the hardware
    is a dev box that lies about `/ups`.
    """
    sim = SimUps("apc", "Smart-UPS 1500", "APC", "AS-1")
    now = time.time()
    assert nut.self_test_outcome(sim.test_result) == "ok"
    assert nut.self_test_outcome("FAILED") == "fail"
    card = nut.payload(UpsReading(name=sim.name, vars=sim.vars(now), commands=[]), now=now)
    assert self_test_check(card["health"])["status"] == "ok"


def test_alarms_drive_the_overall_verdict() -> None:
    health = nut.assess(reading(**{"ups.alarm": "BATT.FAIL", "battery.temperature": "27.4"}),
                        now=1_767_225_600.0)
    assert health["overall"] in ("degraded", "at risk")


# ---- command gating ------------------------------------------------------- #

def base_commands() -> list[str]:
    return list(CMDS)


def by_name(items: list[dict], command: str) -> dict:
    return next(item for item in items if item["command"] == command)


def test_commands_are_disabled_when_actions_are_off() -> None:
    items = nut.available_commands(base_commands(), allow_actions=False, read_only=False, has_credentials=True,
                                   on_battery=False)
    deep = by_name(items, "test.battery.start.deep")
    assert deep["enabled"] is False
    assert "NUT_ALLOW_ACTIONS" in deep["reason"]


def test_commands_are_disabled_without_credentials_that_have_actions() -> None:
    items = nut.available_commands(base_commands(), allow_actions=True, read_only=False, has_credentials=False,
                                   on_battery=False)
    assert by_name(items, "beeper.mute")["enabled"] is False
    assert "upsd.users" in by_name(items, "beeper.mute")["reason"]


def test_dangerous_commands_are_blocked_while_on_battery() -> None:
    """A deep test while already on battery can end the runtime early — that is the whole
    thing the UPS is protecting against, so the UI must not offer it."""
    items = nut.available_commands(base_commands(), allow_actions=True, read_only=False, has_credentials=True,
                                   on_battery=True)
    deep = by_name(items, "test.battery.start.deep")
    assert deep["enabled"] is False
    assert "battery" in deep["reason"].lower()
    safe = by_name(items, "beeper.mute")
    assert safe["enabled"] is True


def test_read_only_overrides_everything_else() -> None:
    items = nut.available_commands(base_commands(), allow_actions=True, read_only=True,
                                   has_credentials=True, on_battery=False)
    assert items and all(item["enabled"] is False for item in items)
    assert any("DASH_READ_ONLY" in (item["reason"] or "") for item in items)


def test_risk_levels_are_labelled_per_command() -> None:
    items = nut.available_commands(base_commands(), allow_actions=True, read_only=False, has_credentials=True,
                                   on_battery=False)
    assert by_name(items, "beeper.mute")["risk"] == nut.RISK_SAFE
    assert by_name(items, "test.battery.start.deep")["risk"] == nut.RISK_DANGEROUS
    assert by_name(items, "load.off")["risk"] == nut.RISK_DANGEROUS


def test_commands_the_driver_does_not_advertise_are_never_offered() -> None:
    items = nut.available_commands(["beeper.mute"], allow_actions=True, read_only=False, has_credentials=True,
                                   on_battery=False)
    assert [item["command"] for item in items] == ["beeper.mute"]


def test_commands_outside_the_risk_catalog_are_never_offered() -> None:
    """A driver can advertise anything; the dashboard only sends what it has a risk label for."""
    items = nut.available_commands(["beeper.mute", "something.new"], allow_actions=True,
                                   read_only=False, has_credentials=True, on_battery=False)
    assert [item["command"] for item in items] == ["beeper.mute"]


def test_command_advice_catalog_is_complete_for_the_ui() -> None:
    advice = nut.command_advice("test.battery.start.deep")
    assert advice and advice["risk"] == nut.RISK_DANGEROUS
    assert advice["label"] and advice["help"]
    assert nut.command_advice("not.a.command") is None


# ---- payload & simulator -------------------------------------------------- #

def test_payload_carries_numbers_health_and_what_the_driver_offers() -> None:
    payload = nut.payload(reading(), now=1_767_225_600.0)
    assert payload["name"] == "cyberpower"
    assert payload["identity"]["model"] == "PR1500LCDRT2U"
    assert payload["numbers"]["charge_pct"] == 100.0
    assert payload["numbers"]["battery_temperature"] == pytest.approx(27.25, abs=0.01)
    assert payload["health"]["overall"] in ("healthy", "degraded", "at risk")
    assert payload["advertised_commands"] == CMDS
    assert payload["on_battery"] is False
    # the raw dump keeps the original kelvin string: normalising for charts must not
    # destroy the value the operator would compare against `upsc`
    assert payload["extra"] or payload["status"] == "OL CHRG"


def test_payload_severity_colours_the_hero_battery_numbers() -> None:
    """charge / load / runtime carry the verdict so the page paints a class, not a threshold."""
    payload = nut.payload(reading(), now=1_767_225_600.0)   # charge 100, runtime 3720 s, load 34 %
    assert payload["severity"] == {"charge_pct": "ok", "load_pct": "ok", "battery_runtime": "ok"}
    # the verdict sits beside the numbers; it does not replace or rename any of them
    assert payload["numbers"]["charge_pct"] == 100.0
    assert payload["numbers"]["battery_runtime"] == 3720.0


@pytest.mark.parametrize("overrides,expected", [
    ({"battery.charge": "15"}, {"charge_pct": "fail"}),
    ({"battery.charge": "50"}, {"charge_pct": "warn"}),      # reaching warn trips it
    ({"battery.charge": "50.1"}, {"charge_pct": "ok"}),
    ({"battery.runtime": "240"}, {"battery_runtime": "fail"}),
    ({"battery.runtime": "600"}, {"battery_runtime": "warn"}),
    ({"ups.load": "95"}, {"load_pct": "fail"}),
    ({"ups.load": "70"}, {"load_pct": "warn"}),
])
def test_severity_trips_the_ratified_ups_thresholds(overrides, expected) -> None:
    payload = nut.payload(reading(**overrides), now=1_767_225_600.0)
    for key, verdict in expected.items():
        assert payload["severity"][key] == verdict, f"{key} with {overrides}"


def test_severity_is_null_when_the_driver_publishes_nothing() -> None:
    """Neither real UPS reports a battery temperature and either can drop `battery.runtime`;
    an absent number gets no colour, and the health check keeps saying `unknown`."""
    values = {key: value for key, value in VAR_ROWS.items()
              if key not in ("battery.runtime", "battery.temperature")}
    payload = nut.payload(nut.UpsReading(name="apc", vars=values, commands=[]),
                          now=1_767_225_600.0)
    assert payload["numbers"]["battery_runtime"] is None
    assert payload["severity"]["battery_runtime"] is None
    assert payload["numbers"]["battery_temperature"] is None
    temp = next(c for c in payload["health"]["checks"] if c["name"] == "Battery temperature")
    assert temp["status"] == "unknown", "severity is additive: assess() keeps its own judgement"


def test_severity_does_not_touch_health_checks_or_numbers() -> None:
    ups = reading()
    payload = nut.payload(ups, now=1_767_225_600.0)
    assert all(check["status"] in ("ok", "warn", "fail", "unknown")
               for check in payload["health"]["checks"])
    assert payload["health"] == nut.assess(ups, now=1_767_225_600.0), \
        "the condition verdict and the value verdict must not contaminate each other"


async def test_simulated_ups_cards_carry_severity_too() -> None:
    """The UPS page is developable without a battery, and still gets real verdicts."""
    for ups in await NutSimulator().read():
        card = nut.payload(ups, now=1_767_225_600.0)
        assert set(card["severity"]) == {"charge_pct", "load_pct", "battery_runtime"}
        assert card["severity"]["charge_pct"] == "ok", "the sim starts at float charge"


async def test_a_fresh_simulator_reads_healthy_so_amber_still_means_something() -> None:
    """The dev model must not train people to ignore the UPS badge.

    Its battery dates are fiction, so they are computed relative to today: a hard-coded
    2019 manufacture date turns every simulated card amber forever, and an amber nobody
    can clear is the same failure `assess()` avoids for a UPS that floats at 90 %.
    """
    for ups in await NutSimulator().read():
        card = nut.payload(ups)
        assert card["health"]["overall"] == "healthy", card["health"]["checks"]
        assert not [c for c in card["health"]["checks"] if c["status"] == "fail"]


def test_history_samples_only_include_real_numbers() -> None:
    values = {key: value for key, value in VAR_ROWS.items() if key != "battery.runtime"}
    samples = nut.history_samples(nut.UpsReading(name="cyberpower", vars=values, commands=[]))
    assert "ups.charge_pct" in samples
    assert "ups.battery_runtime_seconds" not in samples
    assert all(isinstance(value, float) and value == value for value in samples.values())


def test_history_samples_normalise_the_temperature() -> None:
    samples = nut.history_samples(reading())
    assert samples["ups.battery_temperature_c"] == pytest.approx(27.25, abs=0.01)


async def test_simulator_reads_like_upsd_and_is_labelled() -> None:
    readings = await NutSimulator().read()
    assert readings, "the simulator must present at least one UPS"
    for ups in readings:
        assert "(simulated)" in ups.description, "simulated data must be identifiable"
        assert ups.vars["battery.charge"]
        assert "test.battery.start.deep" in ups.commands


async def test_simulator_commands_change_the_model_like_a_real_driver() -> None:
    sim = NutSimulator()
    name = (await sim.read())[0].name
    assert "OK" in sim.command(name, "test.battery.start.quick")
    after = (await sim.read())[0]
    assert after.on_battery is True, "a quick test puts the load on battery"
    assert "OK" in sim.command(name, "test.battery.stop")
    assert (await sim.read())[0].on_battery is False


async def test_simulator_discharges_while_on_battery() -> None:
    """A simulated UPS that never drains cannot exercise a runtime or health path."""
    sim = NutSimulator()
    name = (await sim.read())[0].name
    before = float((await sim.read())[0].vars["battery.charge"])
    sim.command(name, "test.battery.start.deep")
    ups = next(item for item in sim.ups if item.name == name)
    ups.on_battery = True
    for _ in range(30):
        ups.step(60.0)            # half an hour of model time on battery
    assert ups.charge < before


def test_states_never_say_the_same_word_twice() -> None:
    """A driver exporting both the 2.8 token and the legacy flag gets one chip per fact.

    Both spellings arrive at the word "charging"; a chip row that says it twice reads as
    two events. The token wins (it distinguishes trim/boost/full), the subsumed flag goes.
    """
    both = UpsReading(name="x", vars={"battery.charger.status": "chg", "ups.status": "OL CHRG"})
    assert [s["label"] for s in both.states()] == ["charging", "online"]
    dischrg = UpsReading(name="x", vars={"battery.charger.status": "dchg", "ups.status": "OB DISCHRG"})
    assert [s["label"] for s in dischrg.states()] == ["discharging", "on battery"]
    plain = UpsReading(name="x", vars={"ups.status": "OL CHRG"})
    assert [s["label"] for s in plain.states()] == ["online", "charging"], \
        "the flag-only route keeps both facts"


def test_a_recharging_simulated_ups_reports_charging_like_a_real_driver() -> None:
    """A below-float pack on mains is *charging* — APC prints exactly `OL CHRG` there.

    The `/ups` glyph's bolt is keyed on `charger_status == "chg"`, and a model that only
    ever answered `ol`/`OL` made that state unreachable in dev — the flag and the charger
    token also contradicted each other, plain `OL` next to a charge that was climbing.
    """
    ups = SimUps("apc", "Smart-UPS 1500", "APC", "AS-1", charge=60.0)
    vars_ = ups.vars(time.time())
    assert vars_["ups.status"] == "OL CHRG"
    assert vars_["battery.charger.status"] == "chg"
    card = nut.payload(UpsReading(name="apc", vars=vars_, commands=[]))
    assert card["charger_status"] == "chg", "the page derives the bolt from this token only"
    assert card["on_battery"] is False, "OL CHRG is on line, not on the battery"


def test_a_discharging_simulated_ups_is_never_online() -> None:
    """`ol`/`full` next to `OB` is a contradiction; a drawn-from battery reports `dchg`."""
    ups = SimUps("apc", "Smart-UPS 1500", "APC", "AS-1", charge=100.0, on_battery=True)
    vars_ = ups.vars(time.time())
    assert "OL" not in vars_["ups.status"].split()
    assert vars_["battery.charger.status"] == "dchg"


async def test_simulator_command_for_unknown_ups_raises() -> None:
    sim = NutSimulator()
    with pytest.raises(KeyError):
        sim.command("nope", "beeper.mute")


def test_nut_simulate_is_the_only_switch_and_says_so() -> None:
    """The enable rule lives once, next to the model it swaps in.

    `app.py` builds the monitor through this, so `1/true/yes/on` mean the same thing in the
    assembly and in the module — and anything else means *real upsd*, which is the default that
    must never be reachable by an empty or half-set variable.
    """
    assert isinstance(from_env("1"), NutSimulator)
    assert isinstance(from_env(" TRUE "), NutSimulator)
    assert from_env(None) is None
    assert from_env("") is None
    assert from_env("0") is None, "a commented-out or zeroed NUT_SIMULATE must read real upsd"


async def test_every_command_the_simulator_advertises_is_implemented() -> None:
    """`commands()` is what the `/ups` page turns into buttons, so it must not list a lie.

    A command the model advertises but `run_command()` does not implement raises, and the dev
    page would then show `refused` for a button that is supposed to work — the simulator is
    the only driver this app's one write path is ever exercised against, so an unimplemented
    advertised command is a bug in the model, not a missing feature.
    """
    sim = NutSimulator()
    ups = (await sim.read())[0]
    assert ups.commands, "a simulator with no commands cannot exercise the buttons at all"
    for command in ups.commands:
        assert sim.command(ups.name, command).startswith("OK"), command


def test_an_unimplemented_command_is_refused_by_the_model_too() -> None:
    """Silently accepting a command the driver would reject hides a bad catalog entry."""
    ups = SimUps("apc", "Smart-UPS 1500", "APC", "AS-1")
    with pytest.raises(ValueError):
        ups.run_command("outlet.1.switch.off", now=time.time())


def test_the_beeper_commands_change_the_status_a_driver_would_publish() -> None:
    """`beeper.mute` answering OK while nothing changes teaches people to trust a dead button.

    Real drivers report `ups.beeper.status`, and the NUT variable dump on the card shows it,
    so each of the three commands has its own observable state here — `mute` and `disable` are
    different words in NUT and stay different here.
    """
    ups = SimUps("apc", "Smart-UPS 1500", "APC", "AS-1")
    assert ups.vars(time.time())["ups.beeper.status"] == "enabled"
    now = time.time()
    ups.run_command("beeper.mute", now=now)
    assert ups.vars(now)["ups.beeper.status"] == "muted"
    ups.run_command("beeper.disable", now=now)
    assert ups.vars(now)["ups.beeper.status"] == "disabled"
    ups.run_command("beeper.enable", now=now)
    assert ups.vars(now)["ups.beeper.status"] == "enabled"


def test_a_finished_simulated_test_reports_its_result_and_a_stopped_one_reports_nothing() -> None:
    """The command → state change → health-check path the simulator exists for.

    `ups.test.result` is the variable the "Last self-test" check reads, so a quick test that
    ends with the pack back on line but the old result still published shows a card that never
    learned the test happened. An aborted test writes no result: a driver does not.
    """
    ups = SimUps("apc", "Smart-UPS 1500", "APC", "AS-1", test_result="FAILED",
                 test_date="2020-01-01")
    now = time.time()
    ups.run_command("test.battery.start.quick", now=now)
    assert ups.on_battery is True
    ups.tick(now + 20.0)
    assert ups.on_battery is False, "a quick test is over after its ~12s window"
    assert ups.test_result == "PASSED"
    assert ups.test_date == time.strftime("%Y-%m-%d", time.gmtime(now))

    aborted = SimUps("apc", "Smart-UPS 1500", "APC", "AS-1", test_result="FAILED")
    aborted.run_command("test.battery.start.quick", now=now)
    aborted.run_command("test.battery.stop", now=now)
    aborted.tick(now + 20.0)
    assert aborted.test_result == "FAILED", "an aborted test must not be reported as a pass"


def test_a_simulated_alarm_reaches_the_card_and_the_health_check() -> None:
    """The `alarm:` chip and the `Alarms` row are reachable in dev only through the model.

    `assess()` fails that row for the codes that mean the pack or the load is in trouble, and
    neither UPS on this fleet has ever raised one — so the alarm path is pinned here instead of
    against a capture nobody can produce on demand.
    """
    quiet = SimUps("apc", "Smart-UPS 1500", "APC", "AS-1")
    now = time.time()
    assert "ups.alarm" not in quiet.vars(now), "a quiet UPS must not invent an alarm"

    loud = SimUps("apc", "Smart-UPS 1500", "APC", "AS-1", _alarms=["LB"])
    card = nut.payload(UpsReading(name="apc", vars=loud.vars(now), commands=[]))
    assert card["alarms"] == ["LB"]
    check = next(c for c in card["health"]["checks"] if c["name"] == "Alarms")
    assert check["status"] == "fail", "a low-battery alarm is not a warning"