fix(ups): offer every advertised command and gate its aliases too
Two halves of the same bug: the page offered less than the hardware does, and the gate matched only the name it rendered. `available_commands()` drops anything `COMMANDS.get()` misses, and 12 of the 23 commands the two real UPSes advertise had no entry (`beeper.on/off`, `load.on/off.delay`, `shutdown.default/reboot/stayoff`, `driver.reload*`, `driver.killpower`, `test.panel.stop`) — an invisible button the operator had no way to learn about. Every one is catalogued now with a risk tier, and the power-cutting ones are pinned to `RISK_DANGEROUS` (`shutdown.stop` is the exception the name hides: it cancels a shutdown). Both real drivers advertise `beeper.enable` *and* `beeper.on`, `beeper.disable` *and* `beeper.off`, so rendering every advertised name drew "Beeper on" twice. `COMMAND_ALIASES` collapses one action to one entry (named for a name this driver answers, so a driver that only implements the alias still gets its button) and records the other names in `also`. The POST gate resolved the request through that table instead of comparing the literal string: previously `beeper.off` walked past a `disabled` verdict recorded against `beeper.disable` — the same action, and the gate is the control, not the button. LLM-Generated: true
This commit is contained in:
@@ -446,9 +446,12 @@ def ups_router() -> APIRouter:
|
||||
if advice["risk"] == nut.RISK_DANGEROUS and confirm != "DANGER":
|
||||
raise HTTPException(422, f"{advice['label']} can cut power: send "
|
||||
f"confirm=\"DANGER\" to prove you read the warning")
|
||||
if any(item["command"] == command and not item["enabled"] for item in card["commands"]):
|
||||
blocked = next(item for item in card["commands"] if item["command"] == command)
|
||||
raise HTTPException(409, blocked["reason"] or f"{command} is not available")
|
||||
# Resolved through the alias table: `beeper.off` and `beeper.disable` are one action,
|
||||
# and only one of them is listed. Matching the literal name would let the unlisted
|
||||
# alias past a `disabled` verdict — on battery, that is a power cut with a typo.
|
||||
entry = nut.resolve_command(card["commands"], command)
|
||||
if entry is not None and not entry["enabled"]:
|
||||
raise HTTPException(409, entry["reason"] or f"{command} is not available")
|
||||
|
||||
try:
|
||||
if monitor.simulated:
|
||||
|
||||
@@ -675,6 +675,8 @@ COMMANDS: dict[str, dict[str, str]] = {
|
||||
"help": "Aborts a running battery test and returns to line power."},
|
||||
"test.panel.start": {"label": "Panel test", "risk": RISK_SAFE,
|
||||
"help": "Lights and beeps the front panel for a few seconds."},
|
||||
"test.panel.stop": {"label": "Stop panel test", "risk": RISK_SAFE,
|
||||
"help": "Ends a front-panel test."},
|
||||
"test.system.start": {"label": "Full system self-test", "risk": RISK_CAUTION,
|
||||
"help": "Vendor self-test; behaviour varies by model and may move "
|
||||
"the load to battery."},
|
||||
@@ -682,6 +684,12 @@ COMMANDS: dict[str, dict[str, str]] = {
|
||||
"beeper.disable": {"label": "Beeper off", "risk": RISK_SAFE,
|
||||
"help": "Silences audible alarms. Faults are still reported in this dashboard."},
|
||||
"beeper.mute": {"label": "Beeper mute", "risk": RISK_SAFE, "help": "Silence the current alarm only."},
|
||||
# `beeper.on`/`beeper.off` are what the APC and CyberPower drivers actually advertise;
|
||||
# `beeper.enable`/`beeper.disable` are the names other drivers use. Both pairs are real,
|
||||
# so both are catalogued — a command the driver offers must never go missing from the page.
|
||||
"beeper.on": {"label": "Beeper on", "risk": RISK_SAFE, "help": "Re-enable audible alarms."},
|
||||
"beeper.off": {"label": "Beeper off", "risk": RISK_SAFE,
|
||||
"help": "Silences audible alarms. Faults are still reported in this dashboard."},
|
||||
"calibrate.start": {
|
||||
"label": "Start calibration",
|
||||
"risk": RISK_DANGEROUS,
|
||||
@@ -693,8 +701,15 @@ COMMANDS: dict[str, dict[str, str]] = {
|
||||
"help": "Clears the recorded input voltage extremes."},
|
||||
"load.on": {"label": "Energise output", "risk": RISK_CAUTION,
|
||||
"help": "Turns the outlets on."},
|
||||
"load.on.delay": {"label": "Energise output after delay", "risk": RISK_CAUTION,
|
||||
"help": "Turns the outlets on after the driver's shutdown delay. "
|
||||
"Cancels a pending power-down on some models."},
|
||||
"load.off": {"label": "Cut output power", "risk": RISK_DANGEROUS,
|
||||
"help": "Kills power to everything plugged in, immediately, on battery or not."},
|
||||
"load.off.delay": {"label": "Cut output power after delay", "risk": RISK_DANGEROUS,
|
||||
"help": "Cuts outlet power after the driver's shutdown delay. Attached "
|
||||
"systems are not asked to shut down first — this is a hard cut "
|
||||
"on a timer."},
|
||||
"load.reboot": {"label": "Power-cycle outlets", "risk": RISK_DANGEROUS,
|
||||
"help": "Cuts and restores output power. Everything attached reboots."},
|
||||
"shutdown.return": {"label": "Graceful shutdown, then restore", "risk": RISK_DANGEROUS,
|
||||
@@ -704,6 +719,32 @@ COMMANDS: dict[str, dict[str, str]] = {
|
||||
"help": "Shuts attached systems down and stays off."},
|
||||
"shutdown.stop": {"label": "Cancel shutdown", "risk": RISK_SAFE,
|
||||
"help": "Aborts a pending shutdown sequence."},
|
||||
"shutdown.stayoff": {"label": "Shut down and stay off", "risk": RISK_DANGEROUS,
|
||||
"help": "Cuts output power after the delay and does not bring it back "
|
||||
"when utility returns."},
|
||||
"shutdown.default": {"label": "Default shutdown sequence", "risk": RISK_DANGEROUS,
|
||||
"help": "The driver's own shutdown sequence — usually the same as "
|
||||
"shutdown.return, but the order and timing are the driver's "
|
||||
"choice, not yours."},
|
||||
"shutdown.reboot": {"label": "Shut down, then power-cycle", "risk": RISK_DANGEROUS,
|
||||
"help": "Shuts attached systems down, cuts power, and restores it after "
|
||||
"the delay. Everything attached comes back on its own."},
|
||||
# Driver-level commands. `driver.reload` re-reads the driver's configuration; it interrupts
|
||||
# status reporting for a moment but leaves the load alone. `driver.killpower` does not.
|
||||
"driver.reload": {"label": "Reload driver config", "risk": RISK_CAUTION,
|
||||
"help": "Makes the driver re-read ups.conf. Status reporting blips for a "
|
||||
"few seconds; output power is unaffected."},
|
||||
"driver.reload-or-error": {"label": "Reload config, exit on error", "risk": RISK_CAUTION,
|
||||
"help": "Reloads the driver configuration and stops the driver if "
|
||||
"it is invalid — this UPS then stops reporting until it is "
|
||||
"started again."},
|
||||
"driver.reload-or-exit": {"label": "Reload config, or exit", "risk": RISK_CAUTION,
|
||||
"help": "Reloads the driver configuration, exiting if the reload "
|
||||
"cannot be done."},
|
||||
"driver.killpower": {"label": "Kill power now", "risk": RISK_DANGEROUS,
|
||||
"help": "Cuts inverter output immediately with no shutdown sequencing "
|
||||
"and no delay — the fastest way to take everything attached "
|
||||
"down hard."},
|
||||
}
|
||||
|
||||
|
||||
@@ -711,6 +752,39 @@ def command_advice(command: str) -> dict[str, str] | None:
|
||||
return COMMANDS.get(command)
|
||||
|
||||
|
||||
# Driver names that mean the same action. Both real drivers advertise `beeper.enable` AND
|
||||
# `beeper.on`, and `beeper.disable` AND `beeper.off`, so offering every advertised command
|
||||
# drew "Beeper on" and "Beeper off" twice each. Both names stay in `COMMANDS` — a driver that
|
||||
# offers only one of them must still get a button — but one action gets one entry, named for a
|
||||
# name this driver actually answers, in the preference order below.
|
||||
COMMAND_ALIASES: tuple[tuple[str, ...], ...] = (
|
||||
("beeper.enable", "beeper.on"),
|
||||
("beeper.disable", "beeper.off"),
|
||||
)
|
||||
|
||||
|
||||
def alias_group(command: str) -> tuple[str, ...]:
|
||||
"""Every driver name for the same action, preferred first. Just itself if not an alias."""
|
||||
for group in COMMAND_ALIASES:
|
||||
if command in group:
|
||||
return group
|
||||
return (command,)
|
||||
|
||||
|
||||
def resolve_command(commands: list[dict[str, Any]], command: str) -> dict[str, Any] | None:
|
||||
"""The offered entry a command name refers to, aliases included.
|
||||
|
||||
The POST gate compares the request against the command list. Matching on `command` alone
|
||||
would let an alias the page never rendered (`beeper.off`) walk past a `disabled` verdict
|
||||
recorded against the name the page did render (`beeper.disable`) — the same action, and
|
||||
the gate is the control, not the button.
|
||||
"""
|
||||
for item in commands:
|
||||
if item["command"] == command or command in (item.get("also") or []):
|
||||
return item
|
||||
return None
|
||||
|
||||
|
||||
def available_commands(advertised: list[str], *, allow_actions: bool,
|
||||
read_only: bool, has_credentials: bool,
|
||||
on_battery: bool = False) -> list[dict[str, Any]]:
|
||||
@@ -718,12 +792,21 @@ def available_commands(advertised: list[str], *, allow_actions: bool,
|
||||
|
||||
Filtering server-side (not just hiding buttons) is the point: the response says
|
||||
`enabled: false` plus `reason`, so the page never offers a button that 403s.
|
||||
|
||||
One entry per *action*: where a driver advertises both names of an alias group, the
|
||||
preferred name is emitted and the others ride along in `also`, so the page cannot draw the
|
||||
same button twice while `resolve_command()` still refuses the alias when the action is off.
|
||||
"""
|
||||
out: list[dict[str, Any]] = []
|
||||
advertised_set = set(advertised)
|
||||
for name in advertised:
|
||||
advice = COMMANDS.get(name)
|
||||
if advice is None:
|
||||
continue
|
||||
group = alias_group(name)
|
||||
preferred = next(candidate for candidate in group if candidate in advertised_set)
|
||||
if name != preferred:
|
||||
continue # already represented by the preferred name of this action
|
||||
reason = None
|
||||
if read_only:
|
||||
reason = "the server is in DASH_READ_ONLY mode"
|
||||
@@ -741,6 +824,9 @@ def available_commands(advertised: list[str], *, allow_actions: bool,
|
||||
"enabled": reason is None,
|
||||
"reason": reason,
|
||||
"confirm_text": None if advice["risk"] != RISK_DANGEROUS else "DANGER",
|
||||
# other names this driver answers for the same action; the page shows one button,
|
||||
# and POSTing any of them hits this same entry's verdict
|
||||
"also": [other for other in advertised if other in group and other != name],
|
||||
})
|
||||
order = {RISK_SAFE: 0, RISK_CAUTION: 1, RISK_DANGEROUS: 2}
|
||||
out.sort(key=lambda item: (order[item["risk"]], item["label"]))
|
||||
|
||||
@@ -143,6 +143,107 @@ def test_each_capture_is_a_real_upsd_reply_not_an_empty_stub(captures: dict[str,
|
||||
assert card["identity"]["model"], f"{name}: no model to put beside the numbers"
|
||||
|
||||
|
||||
def test_every_command_the_hardware_advertises_is_catalogued(captures: dict[str, nut.UpsReading]) -> None:
|
||||
"""An advertised command with no `COMMANDS` entry is a button that silently does not exist.
|
||||
|
||||
`available_commands()` skips anything `COMMANDS.get()` misses, so the page renders a
|
||||
shorter list than the driver offers and the operator has no way to learn the command is
|
||||
there — 12 of the 23 commands these two UPSes advertise were invisible this way
|
||||
(`beeper.on/off`, `load.off.delay`, `shutdown.default/reboot/stayoff`, `driver.*`,
|
||||
`test.panel.stop`). The captures are ground truth, so a command that appears on real
|
||||
hardware must be catalogued with a risk tier before the fixture is allowed to drift.
|
||||
"""
|
||||
advertised = {command for reading in captures.values() for command in reading.commands}
|
||||
missing = sorted(advertised - set(nut.COMMANDS))
|
||||
assert not missing, f"advertised but not in nut.COMMANDS (the UI cannot offer them): {missing}"
|
||||
|
||||
|
||||
def test_every_catalogued_command_carries_a_risk_tier_and_help() -> None:
|
||||
"""No tier means no confirmation gate; no help means an unlabeled dangerous button."""
|
||||
tiers = {nut.RISK_SAFE, nut.RISK_CAUTION, nut.RISK_DANGEROUS}
|
||||
for name, advice in nut.COMMANDS.items():
|
||||
assert advice["risk"] in tiers, f"{name}: risk {advice['risk']!r} is not a known tier"
|
||||
assert advice["label"].strip(), f"{name}: no label"
|
||||
assert len(advice["help"].strip()) > 15, f"{name}: help text too thin to be a warning"
|
||||
|
||||
|
||||
# The command list is attached by `collectors`, not by `payload()`, so these two tests build it
|
||||
# with `available_commands()` — the function under test — from each capture's real `LIST CMD`.
|
||||
offered = {"allow_actions": True, "read_only": False, "has_credentials": True}
|
||||
|
||||
|
||||
def test_no_two_offered_buttons_carry_the_same_label(captures: dict[str, nut.UpsReading]) -> None:
|
||||
"""`beeper.on`/`beeper.enable` are one action with two driver names and one label.
|
||||
|
||||
Both names are advertised by both real drivers, so a page that renders every advertised
|
||||
command shows "Beeper on" twice — which reads as a rendering bug and makes the operator
|
||||
wonder whether the two buttons do different things. They do not, and only one appears.
|
||||
"""
|
||||
for name, reading in captures.items():
|
||||
labels = [item["label"] for item in nut.available_commands(reading.commands, **offered)]
|
||||
dupes = sorted({label for label in labels if labels.count(label) > 1})
|
||||
assert not dupes, f"{name}: duplicated button labels {dupes}"
|
||||
|
||||
|
||||
def test_alias_names_collapse_one_button_per_action(captures: dict[str, nut.UpsReading]) -> None:
|
||||
"""The collapsed entry says which other names the driver answers, so nothing is hidden."""
|
||||
for ups, reading in captures.items():
|
||||
cmds = nut.available_commands(reading.commands, **offered)
|
||||
by_command = {item["command"]: item for item in cmds}
|
||||
assert len(cmds) == len(by_command), f"{ups}: a command is offered twice"
|
||||
on = by_command.get("beeper.enable")
|
||||
assert on is not None, f"{ups}: lost the beeper-on button entirely"
|
||||
if "beeper.on" in reading.commands:
|
||||
assert on["also"] == ["beeper.on"], f"{ups}: beeper.on not recorded on the entry"
|
||||
assert "beeper.on" not in by_command, f"{ups}: beeper.on rendered as its own button"
|
||||
# every advertised command is still reachable, by the name offered or by its alias
|
||||
for advertised_name in reading.commands:
|
||||
assert nut.resolve_command(cmds, advertised_name) is not None, \
|
||||
f"{ups}: {advertised_name} is advertised but nothing in the UI answers for it"
|
||||
|
||||
|
||||
def test_a_driver_offering_only_the_alias_still_gets_its_button() -> None:
|
||||
"""Emit the name this driver answers, not the preferred one it does not have.
|
||||
|
||||
Sending `beeper.enable` to a driver that only implements `beeper.on` is a refusal the
|
||||
operator cannot act on — the button looks right and the UPS says the command does not exist.
|
||||
"""
|
||||
cmds = nut.available_commands(["beeper.on"], allow_actions=True, read_only=False,
|
||||
has_credentials=True)
|
||||
assert [c["command"] for c in cmds] == ["beeper.on"]
|
||||
assert cmds[0]["label"] == "Beeper on"
|
||||
|
||||
|
||||
def test_an_alias_cannot_walk_past_a_disabled_verdict() -> None:
|
||||
"""The gate compares against the entry, so the unlisted alias is refused too."""
|
||||
cmds = nut.available_commands(["load.off", "load.off.delay", "beeper.disable", "beeper.off"],
|
||||
allow_actions=True, read_only=False,
|
||||
has_credentials=True, on_battery=True)
|
||||
off = nut.resolve_command(cmds, "load.off")
|
||||
assert off is not None and not off["enabled"], "power cut must be refused on battery"
|
||||
# an alias-shaped name for a collapsed action resolves to the same verdict
|
||||
disabled = {"command": "beeper.disable", "enabled": False, "also": ["beeper.off"]}
|
||||
assert nut.resolve_command([disabled], "beeper.off") is disabled
|
||||
assert nut.resolve_command([disabled], "beeper.disable") is disabled
|
||||
assert nut.resolve_command([disabled], "load.off") is None
|
||||
|
||||
|
||||
def test_every_command_that_can_cut_power_is_in_the_dangerous_tier() -> None:
|
||||
"""The tier is the safety control, not the label. `RISK_DANGEROUS` is what makes the page
|
||||
ask for the word `DANGER` and what makes the server refuse the command while the UPS is
|
||||
already on battery — so a power-cutting command left at `safe` or `caution` would be a
|
||||
one-click outage. Derived from the command names the drivers use, not from a list of them.
|
||||
"""
|
||||
# `shutdown.stop` is the exception the name hides: it cancels a pending shutdown, so it
|
||||
# keeps power on and belongs in `safe`, next to `beeper.mute`.
|
||||
for name, advice in nut.COMMANDS.items():
|
||||
can_cut_power = (name.startswith(("load.off", "shutdown.")) and name != "shutdown.stop") \
|
||||
or name in {"load.reboot", "driver.killpower", "calibrate.start"}
|
||||
if can_cut_power:
|
||||
assert advice["risk"] == nut.RISK_DANGEROUS, \
|
||||
f"{name} can cut output power but its risk is {advice['risk']!r}"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Every field the hero row and the glyph read
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
Reference in New Issue
Block a user