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 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

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) -> None:
        self.require_auth = require_auth
        self.close_early = close_early
        self.hang = hang
        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 == "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 == "USER":
            self.user = rest
            send("OK")
        elif verb == "PASS":
            self.authenticated = not self.require_auth or getattr(self, "user", "") == "monitor"
            send("OK" if self.authenticated else "ERR ACCESS-DENIED")
        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_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"


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()


# ---- 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"


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"


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


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