ducoterra
1f20a5e57f
`NutClient.login()` sent `USER`/`PASS`. The deployed upsd 2.8.5 implements neither — its own `HELP` advertises `USERNAME`/`PASSWORD` — so each line answered `ERR UNKNOWN-COMMAND`, `_read_line()` raised, and `POST /api/ups/command` could only ever 502 before `instcmd()`, credentials or not. Values go out unquoted: upsd takes the rest of the line literally, so quoting them would make the quotes part of the credential. `ERROR_HINTS["ACCESS-DENIED"]` blamed the upsd.conf ACL/LISTEN lines, but that code is only reachable on an authenticated session (an unauthenticated client gets `ERR USERNAME-REQUIRED`), so the cause is the account's `instcmds` grant. The hint now names the one thing the operator can change. The fake upsd in tests has no `USER`/`PASS` branch, so a regression answers `ERR UNKNOWN-COMMAND` exactly like the real server, and its `PASSWORD` branch checks the secret so a wrong-credential refusal is expressible. LLM-Generated: true
731 lines
32 KiB
Python
731 lines
32 KiB
Python
"""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
|
|
|
|
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
|
|
|
|
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 == "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"
|
|
|
|
|
|
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"
|
|
|
|
|
|
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"
|