Add homelab dashboards: system, LLM, and UPS on one domain
Three dashboards served by one FastAPI app with a SQLite history store
(raw / 1m / 1h) and hand-written SVG charts: no Prometheus, no Grafana,
no CDN, no build step. Nothing is fetched from the internet at runtime.
* system: a stdlib-only agent (agent/host_agent.py) reports /proc and
/sys counters; the dashboard turns them into rates over a labelled
window, so every number says how it was measured.
* llm: gufo/llama.cpp and vLLM /metrics are parsed by an in-house
exposition parser. Measured delta/elapsed throughput beats a native
per-request gauge, and each value carries its provenance.
* ups: a NUT wire-protocol client plus health checks that name what
they looked at. Commands are gated by NUT_ALLOW_ACTIONS,
DASH_READ_ONLY and a DANGER confirmation, re-checked server-side.
Missing values render as n/a rather than 0, and an unconfigured source
answers 404 {"configured": false} without dialling anything, so "not
deployed here" never looks like an outage.
253 tests at 94% coverage against captured real fixtures; ruff and
pyright clean.
LLM-Generated: true
This commit is contained in:
@@ -0,0 +1,486 @@
|
||||
"""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_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")
|
||||
Reference in New Issue
Block a user