phase: 05_ups_hero_battery_glyph
**Phase 05 final pass — all criteria met; whole project 100 % green.**
Verified (live upsd + isolated simulator):
- Real `apc`/`cyberpower`: Charge/Load/Runtime are the card's three largest elements (38.4 px vs 15.2 px next), coloured only by API `severity` — live right now 100 % ok / 79.0 % **warn** / 5m **fail**; glyph fill = charge (100/100 viewBox units, `--ok` token), `<title>` "apc: Charge 100 percent, on line"; SIMULATED hidden; battery °C `n/a` uncoloured; every button disabled with its reason; `confirm_text: DANGER` on power-cutting commands; **0 non-GET requests** → `data/logs/ups-verify-live-{card,full}.png`.
- Simulator on `:8091` (own `/tmp` DB, killed after): caution dialog → Cancel, dangerous dialog → Run disabled until `DANGER` → Cancel, **0 POSTs**; one deliberate quick test *to the sim only* flipped the card to `OB` (glyph `stroke-dasharray="7 5"`, "on battery", health "discharging, on battery") then `OL CHRG` (bolt) then `OL` → `ups-verify-sim-{dialog-caution,dialog-danger,onbattery-button,onbattery-page,charging-button}.png`; glyph matrix `ups-glyph-*.png` (100/mid/low/unknown/empty/unrated, dark/light/200 %) present; `grep -c "function battery"` = 1.
Defects found in earlier work and fixed:
1. **`poller.py` could stop polling in silence** — 4 of 6 dev-server sources had frozen `attempts` (ages 12–38 min) while `/api/health` still said `ok: true`: the store-write half of `run_once` was outside the `try`, and a dead task's exception is never retrieved. Now `_commit()` moves `last_ok` only after the write lands, `_store_refused()` reports it (`ok:false`, real reason, one `store-error` event + one `recording` on recovery, payload kept), and `_loop()` catches anything that escapes and keeps polling — 4 new tests in `tests/test_poller.py`.
2. **`nut_sim.py` (modified by this phase) sat at 83 % coverage**; its whole command path was untested. Added tests (every advertised command implemented, unknown → `ValueError`, alarms → card + `assess()` fail, `NUT_SIMULATE` truthy rule) and fixed two model lies: `beeper.*` answered OK while changing nothing (now publishes `ups.beeper.status` enabled/muted/disabled) and a finished self-test never wrote `ups.test.result` (aborted writes none). → 100 %.
3. `nut_sim.from_env()` was dead code duplicating `app._truthy`; `app.py` now builds the simulator through it, so the enable gate lives once.
Tests / lint (exact): `bash .agents/validate.sh` → **EXIT=0** — `uv run pytest --cov --cov-fail-under=94` **543 passed, 8 skipped, coverage 97.91 %**; `uv run ruff check src tests agent` clean; `uv run pyright` 0 errors; `node --check` clean; repo hygiene clean. `uv run pytest -m live --live` **with `.env.dev` sourced → 8 passed** (one earlier run: 2 transient `9121/snapshot timed out` against the 4 s `HOST_TIMEOUT`; three later runs 8/8). `uv pip list --outdated` empty. Dev server restarted on the final code: `health ok`, 6/6 sources fresh.
Docs: AGENTS.md — invariant 4 extended ("nothing may stop polling in silence" + store-refusal semantics), `poller.py`/`nut_sim.py` responsibility lines, 533→543 and 96.9 %→97.9 %, "~97 %"→"~98 %". README.md — poller bullet gained the same guarantee, Testing gained "source `.env.dev` before `-m live --live`" (bare runs skip 7 of 8 live tests, which reads like a pass), Battery-test section gained the `deploy/nut-sudoers` install line and the read-before-write warning.
Notes/deviations: no live instant command was ever sent to real hardware (the four gates leave them disabled — AGENTS gap 1 unchanged); the on-battery proof was driven through the real button against `NUT_SIMULATE=1`, which is stronger than the NOTES hand-driven states. `NOTES-verify.md` under `todo/` left untouched per instructions. Next phase: **none** — 05 is the last in the PLAN.md roadmap.
This commit is contained in:
Vendored
+75
@@ -0,0 +1,75 @@
|
||||
# Verbatim upsd reply (LIST UPS / LIST VAR / LIST CMD), recorded by
|
||||
# ./scripts/capture-fixtures.sh — one `VAR`/`CMD` line per protocol line.
|
||||
# Volatile values (ups.load, battery.runtime, the voltages) move every poll,
|
||||
# so a diff here on re-capture is expected; the contract asserts shape, units
|
||||
# and which variables the drivers omit, never those numbers.
|
||||
UPS apc "APC Back-UPS RS 1500MS (USB)"
|
||||
VAR apc battery.charge "100"
|
||||
VAR apc battery.charge.low "10"
|
||||
VAR apc battery.charge.warning "50"
|
||||
VAR apc battery.date "2001/09/25"
|
||||
VAR apc battery.mfr.date "2021/06/24"
|
||||
VAR apc battery.runtime "648"
|
||||
VAR apc battery.runtime.low "120"
|
||||
VAR apc battery.type "PbAc"
|
||||
VAR apc battery.voltage "27.3"
|
||||
VAR apc battery.voltage.nominal "24.0"
|
||||
VAR apc device.mfr "American Power Conversion"
|
||||
VAR apc device.model "Back-UPS RS 1500MS"
|
||||
VAR apc device.serial "SCRUBBEDSERIAL1 "
|
||||
VAR apc device.type "ups"
|
||||
VAR apc driver.debug "0"
|
||||
VAR apc driver.flag.allow_killpower "0"
|
||||
VAR apc driver.name "usbhid-ups"
|
||||
VAR apc driver.parameter.interrupt_pipe_no_events_tolerance "-1"
|
||||
VAR apc driver.parameter.pollfreq "30"
|
||||
VAR apc driver.parameter.pollinterval "2"
|
||||
VAR apc driver.parameter.port "auto"
|
||||
VAR apc driver.parameter.productid "0002"
|
||||
VAR apc driver.parameter.synchronous "auto"
|
||||
VAR apc driver.parameter.vendorid "051d"
|
||||
VAR apc driver.state "quiet"
|
||||
VAR apc driver.version "2.8.5"
|
||||
VAR apc driver.version.data "APC HID 0.101"
|
||||
VAR apc driver.version.internal "0.71"
|
||||
VAR apc driver.version.usb "libusb-1.0.30 (API: 0x0100010C)"
|
||||
VAR apc input.sensitivity "medium"
|
||||
VAR apc input.transfer.high "144"
|
||||
VAR apc input.transfer.low "88"
|
||||
VAR apc input.voltage "123.0"
|
||||
VAR apc input.voltage.nominal "120"
|
||||
VAR apc ups.beeper.status "disabled"
|
||||
VAR apc ups.delay.shutdown "20"
|
||||
VAR apc ups.firmware "952.e5 .D"
|
||||
VAR apc ups.firmware.aux "e5 "
|
||||
VAR apc ups.load "48"
|
||||
VAR apc ups.mfr "American Power Conversion"
|
||||
VAR apc ups.mfr.date "2021/01/08"
|
||||
VAR apc ups.model "Back-UPS RS 1500MS"
|
||||
VAR apc ups.productid "0002"
|
||||
VAR apc ups.realpower.nominal "900"
|
||||
VAR apc ups.serial "SCRUBBEDSERIAL1 "
|
||||
VAR apc ups.status "OL"
|
||||
VAR apc ups.test.result "No test initiated"
|
||||
VAR apc ups.timer.reboot "0"
|
||||
VAR apc ups.timer.shutdown "-1"
|
||||
VAR apc ups.vendorid "051d"
|
||||
CMD apc beeper.disable
|
||||
CMD apc beeper.enable
|
||||
CMD apc beeper.mute
|
||||
CMD apc beeper.off
|
||||
CMD apc beeper.on
|
||||
CMD apc driver.killpower
|
||||
CMD apc driver.reload
|
||||
CMD apc driver.reload-or-error
|
||||
CMD apc driver.reload-or-exit
|
||||
CMD apc load.off
|
||||
CMD apc load.off.delay
|
||||
CMD apc shutdown.default
|
||||
CMD apc shutdown.reboot
|
||||
CMD apc shutdown.stop
|
||||
CMD apc test.battery.start.deep
|
||||
CMD apc test.battery.start.quick
|
||||
CMD apc test.battery.stop
|
||||
CMD apc test.panel.start
|
||||
CMD apc test.panel.stop
|
||||
Vendored
+79
@@ -0,0 +1,79 @@
|
||||
# Verbatim upsd reply (LIST UPS / LIST VAR / LIST CMD), recorded by
|
||||
# ./scripts/capture-fixtures.sh — one `VAR`/`CMD` line per protocol line.
|
||||
# Volatile values (ups.load, battery.runtime, the voltages) move every poll,
|
||||
# so a diff here on re-capture is expected; the contract asserts shape, units
|
||||
# and which variables the drivers omit, never those numbers.
|
||||
UPS cyberpower "CyberPower PR1500LCDRT2U (USB)"
|
||||
VAR cyberpower battery.charge "100"
|
||||
VAR cyberpower battery.charge.low "10"
|
||||
VAR cyberpower battery.charge.warning "20"
|
||||
VAR cyberpower battery.mfr.date "CPS"
|
||||
VAR cyberpower battery.runtime "3350"
|
||||
VAR cyberpower battery.runtime.low "300"
|
||||
VAR cyberpower battery.status "100%"
|
||||
VAR cyberpower battery.type "PbAcid"
|
||||
VAR cyberpower battery.voltage "27.2"
|
||||
VAR cyberpower battery.voltage.nominal "24"
|
||||
VAR cyberpower device.mfr "CPS"
|
||||
VAR cyberpower device.model "GX1500U"
|
||||
VAR cyberpower device.serial "SCRUBBEDSERIAL1"
|
||||
VAR cyberpower device.type "ups"
|
||||
VAR cyberpower driver.debug "0"
|
||||
VAR cyberpower driver.flag.allow_killpower "0"
|
||||
VAR cyberpower driver.name "usbhid-ups"
|
||||
VAR cyberpower driver.parameter.interrupt_pipe_no_events_tolerance "-1"
|
||||
VAR cyberpower driver.parameter.pollfreq "12"
|
||||
VAR cyberpower driver.parameter.pollinterval "2"
|
||||
VAR cyberpower driver.parameter.port "auto"
|
||||
VAR cyberpower driver.parameter.productid "0601"
|
||||
VAR cyberpower driver.parameter.synchronous "auto"
|
||||
VAR cyberpower driver.parameter.vendorid "0764"
|
||||
VAR cyberpower driver.state "quiet"
|
||||
VAR cyberpower driver.version "2.8.5"
|
||||
VAR cyberpower driver.version.data "CyberPower HID 0.85"
|
||||
VAR cyberpower driver.version.internal "0.71"
|
||||
VAR cyberpower driver.version.usb "libusb-1.0.30 (API: 0x0100010C)"
|
||||
VAR cyberpower input.sensitivity "normal"
|
||||
VAR cyberpower input.transfer.high "139"
|
||||
VAR cyberpower input.transfer.low "100"
|
||||
VAR cyberpower input.voltage "122.0"
|
||||
VAR cyberpower input.voltage.nominal "120"
|
||||
VAR cyberpower output.voltage "122.0"
|
||||
VAR cyberpower ups.beeper.status "disabled"
|
||||
VAR cyberpower ups.delay.shutdown "60"
|
||||
VAR cyberpower ups.delay.start "120"
|
||||
VAR cyberpower ups.firmware "CR01906AC911"
|
||||
VAR cyberpower ups.load "14"
|
||||
VAR cyberpower ups.mfr "CPS"
|
||||
VAR cyberpower ups.model "GX1500U"
|
||||
VAR cyberpower ups.power "131"
|
||||
VAR cyberpower ups.power.nominal "1500"
|
||||
VAR cyberpower ups.productid "0601"
|
||||
VAR cyberpower ups.realpower "127"
|
||||
VAR cyberpower ups.realpower.nominal "900"
|
||||
VAR cyberpower ups.serial "SCRUBBEDSERIAL1"
|
||||
VAR cyberpower ups.status "OL"
|
||||
VAR cyberpower ups.test.result "No test initiated"
|
||||
VAR cyberpower ups.timer.shutdown "-60"
|
||||
VAR cyberpower ups.timer.start "-60"
|
||||
VAR cyberpower ups.vendorid "0764"
|
||||
CMD cyberpower beeper.disable
|
||||
CMD cyberpower beeper.enable
|
||||
CMD cyberpower beeper.mute
|
||||
CMD cyberpower beeper.off
|
||||
CMD cyberpower beeper.on
|
||||
CMD cyberpower driver.killpower
|
||||
CMD cyberpower driver.reload
|
||||
CMD cyberpower driver.reload-or-error
|
||||
CMD cyberpower driver.reload-or-exit
|
||||
CMD cyberpower load.off
|
||||
CMD cyberpower load.off.delay
|
||||
CMD cyberpower load.on
|
||||
CMD cyberpower load.on.delay
|
||||
CMD cyberpower shutdown.default
|
||||
CMD cyberpower shutdown.return
|
||||
CMD cyberpower shutdown.stayoff
|
||||
CMD cyberpower shutdown.stop
|
||||
CMD cyberpower test.battery.start.deep
|
||||
CMD cyberpower test.battery.start.quick
|
||||
CMD cyberpower test.battery.stop
|
||||
+260
-7
@@ -575,17 +575,20 @@ def test_hero_reads_the_verdict_instead_of_a_threshold(client) -> None:
|
||||
|
||||
|
||||
def test_severity_for_is_kept_as_the_documented_fallback(client) -> None:
|
||||
"""The old call sites still work until the remaining phases convert each page.
|
||||
"""Every page call site is gone now that phase 05 converted `/ups`.
|
||||
|
||||
`/` was converted in phase 02, `/system` in phase 03 and `/llm` in phase 04: each
|
||||
colours itself from the `severity` verdicts the API already sends, so it carries no
|
||||
threshold of its own. `/ups` is phase 05's job and must keep compiling against the
|
||||
fallback until it is converted — deleting it early would break that page.
|
||||
`/` was converted in phase 02, `/system` in phase 03, `/llm` in phase 04 and `/ups` in
|
||||
phase 05, so no page colours anything from a threshold it wrote itself. The function
|
||||
itself stays: it is the documented fallback `setBar()` uses when a caller names no
|
||||
verdict, and a page that ever needs a verdict the API does not send should add it to
|
||||
`severity.py` rather than here — which is what the docstring on it says.
|
||||
"""
|
||||
js = client.get("/static/app.js").text
|
||||
assert "FALLBACK ONLY" in js
|
||||
assert "severityFor" in client.get("/static/page-ups.js").text, \
|
||||
"/ups still uses the fallback until phase 05 converts it"
|
||||
assert "severityFor" in js, "the documented fallback itself was deleted"
|
||||
for page in ("page-index.js", "page-system.js", "page-llm.js", "page-ups.js"):
|
||||
assert "severityFor" not in client.get(f"/static/{page}").text, \
|
||||
f"{page} colours something from a threshold of its own"
|
||||
|
||||
|
||||
def test_overview_colours_from_the_payload_not_from_js_thresholds(client) -> None:
|
||||
@@ -826,6 +829,256 @@ def test_sparkline_ships_the_fixed_scale_and_verdict_options(client) -> None:
|
||||
assert "transition:" not in body and "animation:" not in body
|
||||
|
||||
|
||||
# ---- the battery glyph (phase 05 task 02) ---------------------------------
|
||||
#
|
||||
# `DashCharts.battery()` is the one visual in this app whose *shape* is the data: the fill
|
||||
# is the charge. That makes the honesty rules heavier here than anywhere else — a fill
|
||||
# rounded up to look reassuring, a colour invented for an unrated cell, or a state only
|
||||
# readable as green-vs-red would each turn a glance at the battery into a wrong claim.
|
||||
# There is no JS test runner (a locked anchor), so the construction is pinned on the served
|
||||
# bytes and shown in a browser (`data/logs/ups-glyph-*.png`).
|
||||
|
||||
def _battery_js(client: TestClient) -> str:
|
||||
"""The glyph's code as served (state table + body), comments stripped.
|
||||
|
||||
Prose cannot satisfy these assertions, and the phrase table the accessible name is
|
||||
built from is part of the component, so the slice starts above `function battery`.
|
||||
"""
|
||||
charts = client.get("/static/charts.js").text
|
||||
body = charts[charts.index("const BATTERY_STATES"):charts.index("window.DashCharts")]
|
||||
assert "function battery(container, opts)" in body
|
||||
return _code(body)
|
||||
|
||||
|
||||
def test_battery_glyph_is_exported_from_the_shared_chart_library(client) -> None:
|
||||
"""The phase's own grep criterion, as a test: exactly one glyph, and it is public."""
|
||||
charts = client.get("/static/charts.js").text
|
||||
assert charts.count("function battery") == 1
|
||||
exports = charts[charts.index("window.DashCharts"):]
|
||||
assert re.search(r"\bbattery\b", exports), "DashCharts.battery is not exported"
|
||||
# the wrapper the stylesheet sizes, added by the component rather than each caller
|
||||
assert ".battery {" in _css(client) and ".battery svg" in _css(client)
|
||||
|
||||
|
||||
def test_battery_fill_is_the_charge_and_is_never_padded(client) -> None:
|
||||
"""2 % must look nearly empty, and 0 % must leave an empty shell, not a stray pixel."""
|
||||
body = _battery_js(client)
|
||||
# clamped to the scale, and the clamp is the only thing that touches the number: a
|
||||
# minimum visible width would be a reassuring battery over a flat one
|
||||
assert "Math.max(0, Math.min(100," in body
|
||||
assert "(pctValue / 100) * inner.w" in body, "the fill stopped being a fraction of charge"
|
||||
assert not re.search(r"Math\.max\(\s*[0-9.]*[1-9]\s*,", body), \
|
||||
"the charge fill got a minimum width — a small charge would look bigger than it is"
|
||||
# nothing is drawn for a cell at zero, and the plate it fills is a clip, so a low fill
|
||||
# cannot leak outside the rounded interior
|
||||
assert "pctValue > 0" in body
|
||||
assert "<clipPath" in body and 'clip-path="url(#' in body
|
||||
# an empty cell is the sunken plate, never a transparent hole in the card
|
||||
assert "cssVar('--bg-sunken')" in body
|
||||
|
||||
|
||||
def test_battery_colours_come_from_the_verdict_tokens_only(client) -> None:
|
||||
"""The verdict paints; an unrated cell draws a grey shape and no colour at all."""
|
||||
body = _battery_js(client)
|
||||
for token in ("cssVar('--ok')", "cssVar('--warn')", "cssVar('--fail')"):
|
||||
assert token in body, f"the glyph stopped taking its colour from {token}"
|
||||
# no verdict → muted outline, and a measured charge in faint grey at half strength: a
|
||||
# shape, not a judgement (an unknown number must never look green)
|
||||
assert "cssVar('--text-muted')" in body and "cssVar('--text-faint')" in body
|
||||
assert ' opacity=".5"' in body, "an unrated charge is drawn with the same weight as a rated one"
|
||||
assert ".sev-" not in body, "the glyph invented a parallel severity vocabulary"
|
||||
assert "severityFor(" not in body, "the glyph compares the charge to a threshold itself"
|
||||
|
||||
|
||||
def test_battery_states_are_readable_without_colour(client) -> None:
|
||||
"""Every state has a *shape*, so the glyph survives monochrome and colour-blindness.
|
||||
|
||||
on battery = dashed outline, charging = bolt, unknown = a dash pattern through the
|
||||
body, online = a plain solid shell. A word the page did not intend falls back to
|
||||
`unknown` rather than to a confident "on line".
|
||||
"""
|
||||
body = _battery_js(client)
|
||||
assert 'BATTERY_STATES.indexOf(o.state) >= 0' in body, \
|
||||
"an unrecognised state would be drawn as some other state"
|
||||
assert 'state === \'onbattery\'' in body and 'stroke-dasharray="7 5"' in body, \
|
||||
"the on-battery outline is no longer dashed"
|
||||
assert "state === 'charging'" in body and 'd="M13 2 L3 14' in body, "the charging bolt is gone"
|
||||
assert "state === 'unknown'" in body and 'stroke-dasharray="5 6"' in body, \
|
||||
"an unknown charge no longer draws a dash through the body"
|
||||
# and the unknown case really is *no fill*: the shape has to say it on its own
|
||||
assert "pctValue != null" in body
|
||||
|
||||
|
||||
def test_battery_glyph_names_itself_and_can_say_it_once(client) -> None:
|
||||
"""`role="img"` + `<title>` built from the inputs, with an explicit `aria-hidden` opt-out."""
|
||||
body = _battery_js(client)
|
||||
assert 'role="img"' in body and "<title>" in body and "aria-label=" in body
|
||||
for phrase in ("Charge ", " percent", "Charge unknown", "on line", "on battery",
|
||||
"charging", "state unknown"):
|
||||
assert phrase in body, f"the accessible name lost {phrase!r}"
|
||||
# the label is a UPS name from upsd, so it goes through the one escaper
|
||||
assert body.count("escapeHtml(name)") == 2, "<title> or aria-label stopped escaping the label"
|
||||
assert 'aria-hidden="true"' in body, "the decorative opt-out for a duplicate hero number is gone"
|
||||
|
||||
|
||||
def test_battery_glyph_scales_with_its_container_and_never_animates(client) -> None:
|
||||
"""One `viewBox` + `preserveAspectRatio` for a phone and a 27" monitor; no motion at all."""
|
||||
body = _battery_js(client)
|
||||
assert 'viewBox="0 0 ${W} ${H}"' in body
|
||||
assert 'preserveAspectRatio="xMidYMid meet"' in body, "the glyph stopped scaling with its box"
|
||||
assert "ResizeObserver" not in body, "the glyph redraws on resize instead of scaling"
|
||||
assert "transition" not in body and "animation" not in body
|
||||
css = _code(_css(client))
|
||||
wrapper = re.search(r"\.battery \{([^}]*)\}", css)
|
||||
inner = re.search(r"\.battery svg \{([^}]*)\}", css)
|
||||
assert wrapper and inner, "dash.css no longer sizes the glyph box"
|
||||
assert "width: 100%" in wrapper.group(1) and "max-width" in wrapper.group(1), \
|
||||
"the glyph is not capped inside its container"
|
||||
assert "width: 100%" in inner.group(1) and "height: auto" in inner.group(1), \
|
||||
"the svg would stretch instead of keeping the viewBox ratio"
|
||||
both = wrapper.group(1) + inner.group(1)
|
||||
assert "transition" not in both and "animation" not in both
|
||||
|
||||
|
||||
# ---- the `/ups` hero row (phase 05 task 03) ------------------------------
|
||||
#
|
||||
# The card used to be nine equal `.metric` cells in which charge, load and runtime were the
|
||||
# same size as the battery amperage. These pin the shape that replaced it on the served
|
||||
# bytes (a locked anchor: no JS test runner), because the whole point of the row is a size
|
||||
# and colour hierarchy a later "tidy up" can quietly flatten back to nine equal cells.
|
||||
|
||||
def _ups_card_js(client: TestClient) -> str:
|
||||
"""The card markup as served, from `upsCard()` to `renderCard()`."""
|
||||
js = client.get("/static/page-ups.js").text
|
||||
return js[js.index("function upsCard("):js.index("function renderCard(")]
|
||||
|
||||
|
||||
def test_ups_card_opens_with_the_glyph_and_three_heroes(client) -> None:
|
||||
"""Charge / Load / Runtime are the card's three largest elements, next to the glyph.
|
||||
|
||||
`data-sev` names the verdict path each slot colours itself from, the same convention the
|
||||
`/system` host card uses — so "which number is coloured by what" is readable in the
|
||||
markup, and a rename that disconnects the page from `/api/ups/snapshot` fails here
|
||||
instead of shipping a card of `n/a`.
|
||||
"""
|
||||
card = _ups_card_js(client)
|
||||
assert '<div class="ups-hero">' in card, "the hero row left the top of the card"
|
||||
assert 'data-k="glyph"' in card, "the battery glyph is not part of the hero row"
|
||||
slots = re.findall(r'<div class="hero[^"]*" data-k="([^"]+)"', card)
|
||||
assert slots == ["charge", "load", "runtime"], \
|
||||
"the hero row is no longer charge / load / runtime in the agreed order"
|
||||
for hook in ("severity.charge_pct", "severity.load_pct", "severity.battery_runtime"):
|
||||
assert hook in card, f"the card no longer names {hook} as a colour source"
|
||||
# the hero row is the first thing in the card body, above the health checks and commands
|
||||
assert card.index("ups-hero") < card.index('data-k="checks"')
|
||||
assert card.index("ups-hero") < card.index('data-k="commands"')
|
||||
|
||||
|
||||
def test_ups_heroes_take_their_colour_from_the_payload(client) -> None:
|
||||
"""Three `Dash.hero()` calls, three payload verdicts, no JS threshold.
|
||||
|
||||
The charge hero carries **no** `.bar`: the glyph beside it is that same value drawn as a
|
||||
shape, and one value drawn twice in one row is the pattern this app avoids (load, which
|
||||
has no other shape, keeps its bar). Runtime prose comes from `runtimeText()` alone, and
|
||||
a runtime nobody estimated is passed as `null` so `hero()` renders the faint, uncoloured
|
||||
unknown rather than the string `n/a` styled like a measurement.
|
||||
"""
|
||||
js = client.get("/static/page-ups.js").text
|
||||
body = js[js.index("function renderCard("):js.index("// ---- command dialog")]
|
||||
assert body.count("Dash.hero(q(") == 3, "the hero row is not three hero numbers"
|
||||
for verdict in ("verdicts.charge_pct", "verdicts.load_pct", "verdicts.battery_runtime"):
|
||||
assert verdict in body, f"{verdict} never reaches a hero"
|
||||
charge = body[body.index("Dash.hero(q('charge')"):body.index("Dash.hero(q('load')")]
|
||||
assert "bar" not in charge, "charge grew a second shape when the glyph already is one"
|
||||
load = body[body.index("Dash.hero(q('load')"):body.index("Dash.hero(q('runtime')")]
|
||||
assert "bar: true" in load, "load lost its bar"
|
||||
assert "runtimeText(n.battery_runtime)" in body, "runtime prose bypasses runtimeText()"
|
||||
runtime = body[body.index("Dash.hero(q('runtime')"):]
|
||||
assert "n.battery_runtime == null ? null" in runtime, \
|
||||
"an unmeasured runtime is rendered as a measurement"
|
||||
assert "severityFor(" not in body
|
||||
|
||||
|
||||
def test_ups_secondary_electrical_row_is_demoted_not_deleted(client) -> None:
|
||||
"""Line in, output, battery volts and amps, battery °C: still there, visibly second.
|
||||
|
||||
These are the numbers you reach for *while* the UPS is on battery, so they were demoted
|
||||
rather than folded into the disclosure — and they stay one step down (smaller, muted) so
|
||||
the three verdicts remain the largest elements, which is also what keeps the card from
|
||||
growing so tall that two stop fitting side by side on a laptop.
|
||||
"""
|
||||
card = _ups_card_js(client)
|
||||
assert 'class="metrics ups-electrical"' in card
|
||||
for key in ("line", "out", "batt", "amps", "temp"):
|
||||
assert f'data-k="{key}"' in card, f"the electrical row lost {key}"
|
||||
assert card.index("ups-electrical") > card.index("ups-hero")
|
||||
css = _code(_css(client))
|
||||
rule = re.search(r"\.ups-electrical \.metric \.value\.sm \{([^}]*)\}", css)
|
||||
assert rule, "dash.css no longer styles the secondary row"
|
||||
assert "var(--text-muted)" in rule.group(1), "the secondary row is not muted"
|
||||
sizes = re.findall(r"font-size: ([\d.]+)rem", rule.group(1))
|
||||
assert sizes and float(sizes[0]) < 1.3, "the secondary row competes with the heroes"
|
||||
# `n/a` is fainter than a muted reading — neither real driver reports a battery temperature
|
||||
null_rule = re.search(r"\.ups-electrical \.metric \.value\.sm\.is-null \{([^}]*)\}", css)
|
||||
assert null_rule and "var(--text-faint)" in null_rule.group(1), \
|
||||
"an unmeasured secondary value is not faint, so it reads like a number"
|
||||
|
||||
|
||||
def test_ups_glyph_state_comes_from_the_api_fields(client) -> None:
|
||||
"""`glyphState()` maps `on_battery` + `charger_status` — and `full` is **not** charging.
|
||||
|
||||
The mapping is presentation, so it lives in the page (task 02's assumption), but it is
|
||||
still a mapping the real drivers exercise: neither UPS here exports
|
||||
`battery.charger.status`, so the fallback that reads a state the API rated `ok` is what
|
||||
draws a solid shell on both of them. Nothing here re-reads `ups.status` —
|
||||
`test_the_page_does_not_re_parse_nut_status_flags` holds that line.
|
||||
"""
|
||||
js = _code(client.get("/static/page-ups.js").text)
|
||||
glyph = js[js.index("function glyphState("):js.index("function drawGlyph(")]
|
||||
assert "item.on_battery" in glyph and "'onbattery'" in glyph
|
||||
assert "charger === 'chg'" in glyph and "'charging'" in glyph
|
||||
assert "charger === 'dchg'" in glyph, "a discharging charger token is on battery too"
|
||||
assert "charger === 'full'" in glyph and "'online'" in glyph, \
|
||||
"a full, online battery must not be drawn as charging"
|
||||
assert "item.states" in glyph, "a driver with no rich status lost its state entirely"
|
||||
assert "'unknown'" in glyph, "an unknown state now degrades to a confident 'on line'"
|
||||
|
||||
|
||||
def test_ups_glyph_repaints_only_when_something_visible_changes(client) -> None:
|
||||
"""Charge jitters by a tenth of a percent every poll; the `<svg>` does not rebuild on it.
|
||||
|
||||
Rebuilding would re-resolve the theme tokens and re-announce the accessible name every
|
||||
`NUT_INTERVAL_SECONDS` for a difference no eye can see. `decorative: true` is the
|
||||
component's documented opt-out — the charge hero and the state chips already say
|
||||
"Charge 100 percent, on line", and a screen reader should not hear it twice.
|
||||
"""
|
||||
js = _code(client.get("/static/page-ups.js").text)
|
||||
draw = js[js.index("function drawGlyph("):js.index("function upsCard(")]
|
||||
assert "DashCharts.battery(" in draw
|
||||
assert "cell.dataset.drawn === drawn" in draw, "the glyph rebuilds on every poll"
|
||||
assert "decorative: true" in draw, "the glyph duplicates the hero number in a11y"
|
||||
assert "glyphState(item)" in draw
|
||||
|
||||
|
||||
def test_ups_hero_row_makes_room_for_the_glyph(client) -> None:
|
||||
"""The glyph gets a real column, and the row stacks rather than clipping on a phone.
|
||||
|
||||
The track floor is the widest runtime prose the page prints (`9h 59m`) — a number that
|
||||
cannot wrap needs the floor to be its own width, or the row grows a clipped `1h 28`.
|
||||
"""
|
||||
css = _code(_css(client))
|
||||
row = re.search(r"\.ups-hero \{([^}]*)\}", css)
|
||||
assert row, "dash.css lost the UPS hero row"
|
||||
assert "minmax(112px, 156px) minmax(0, 1fr)" in row.group(1), "the glyph has no column"
|
||||
assert "align-items: center" in row.group(1)
|
||||
heroes = re.search(r"\.ups-hero \.heroes \{([^}]*)\}", css)
|
||||
assert heroes and "minmax(min(140px, 100%), 1fr)" in heroes.group(1), \
|
||||
"a hero track can now clip the widest runtime prose"
|
||||
at = css.index("@media (max-width: 620px)")
|
||||
block = css[at:at + 240]
|
||||
assert ".ups-hero" in block and "minmax(0, 1fr)" in block, "the row never stacks"
|
||||
|
||||
|
||||
# ---- the `/llm` layout (phase 04 task 03) ---------------------------------
|
||||
#
|
||||
# The drift this phase fixes was a 14-cell grid where everything was the same size and
|
||||
|
||||
@@ -30,8 +30,10 @@ from card_contract import assert_card_contract
|
||||
from dashboards import collectors
|
||||
from dashboards.config import Settings, load
|
||||
from dashboards.rates import RateEstimator
|
||||
from dashboards.severity import classify
|
||||
from dashboards.sources import nut
|
||||
from dashboards.sources.host import HostCollector
|
||||
from ups_contract import assert_ups_hero_contract
|
||||
|
||||
pytestmark = pytest.mark.live
|
||||
|
||||
@@ -151,6 +153,29 @@ async def test_upsd_reports_both_upses_with_sane_units(settings: Settings) -> No
|
||||
assert temp is None or -20.0 <= temp <= 80.0, f"{name}: battery temp {temp} is not °C"
|
||||
|
||||
|
||||
async def test_ups_hero_fields_are_present_on_the_real_batteries(settings: Settings) -> None:
|
||||
"""Phase 05's hero row and battery glyph read these paths on the real UPSes.
|
||||
|
||||
Hermetic twin: `tests/test_ups_contract.py`, driven from the verbatim captures in
|
||||
`tests/fixtures/nut_*.vars`. This is the half that catches a driver update renaming a
|
||||
variable, adding `battery.charger.status`, or starting to report a temperature in a unit
|
||||
the page has never seen — the assertion is on the shape and the vocabulary, never on a
|
||||
number that moves between polls.
|
||||
"""
|
||||
require_upsd(settings)
|
||||
result = await collectors.NutMonitor(settings).collect()
|
||||
cards = result.payload["ups"]
|
||||
assert cards, "upsd answered but exported no UPS — nothing for the hero row to render"
|
||||
for card in cards:
|
||||
assert_ups_hero_contract(card) # includes: charger_status is None or a token we label
|
||||
runtime = card["numbers"]["battery_runtime"]
|
||||
assert card["severity"]["battery_runtime"] == (None if runtime is None else
|
||||
classify("ups.battery_runtime", runtime))
|
||||
assert not (card["on_battery"] and card["status"] and "OL" in card["status"].split()), (
|
||||
f"{card['name']}: on_battery and OL at once — the glyph would show the wrong state"
|
||||
)
|
||||
|
||||
|
||||
async def test_ups_command_gating_is_honest_without_sending_anything(settings: Settings) -> None:
|
||||
"""Every advertised command is either enabled or carries the reason it is not.
|
||||
|
||||
|
||||
+137
-1
@@ -8,13 +8,14 @@ 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
|
||||
from dashboards.sources.nut_sim import NutSimulator, SimUps, from_env
|
||||
|
||||
VAR_ROWS = {
|
||||
"battery.charge": "100",
|
||||
@@ -548,7 +549,142 @@ async def test_simulator_discharges_while_on_battery() -> None:
|
||||
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"
|
||||
|
||||
@@ -13,6 +13,7 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import sqlite3
|
||||
import time
|
||||
from collections.abc import Iterator
|
||||
from typing import Any
|
||||
@@ -215,3 +216,110 @@ async def test_events_written_by_a_job_land_in_the_store(store) -> None:
|
||||
lambda: asyncio.sleep(0, Result(payload={}, events=[event])))
|
||||
await job.run_once(store)
|
||||
assert [e.message for e in store.recent_events(limit=5)] == ["OL → OB"]
|
||||
|
||||
|
||||
class RefusingStore(SampleStore):
|
||||
"""A store that turns good polls away for a while — what SQLite does when it is locked.
|
||||
|
||||
This is not hypothetical: the dev server lost four sources at once this way, and nothing
|
||||
in the UI, the health record or the event log said so.
|
||||
"""
|
||||
|
||||
def __init__(self, path: Any, refusals: int) -> None:
|
||||
super().__init__(path)
|
||||
self.refusals = refusals
|
||||
|
||||
def add_many(self, samples: Any) -> int:
|
||||
if self.refusals > 0:
|
||||
self.refusals -= 1
|
||||
raise sqlite3.OperationalError("database is locked")
|
||||
return super().add_many(samples)
|
||||
|
||||
|
||||
async def test_a_store_that_refuses_a_good_poll_is_reported_not_survived_silently(
|
||||
tmp_path: Any, caplog: pytest.LogCaptureFixture) -> None:
|
||||
"""A source whose history is not landing is broken, and must read that way.
|
||||
|
||||
The upstream answered, so the freshest payload stays on the card; everything else says the
|
||||
opposite of healthy — `ok: false`, the real reason in `error`, and `last_ok` frozen so the
|
||||
page ages the numbers instead of calling a write failure a live source.
|
||||
"""
|
||||
store = RefusingStore(tmp_path / "locked.sqlite3", refusals=0)
|
||||
try:
|
||||
job = Job("nut", "ups", "NUT server", 0.5, lambda: asyncio.sleep(0, ok_result(42.0)))
|
||||
assert await job.run_once(store) is True, "the first poll writes before the store turns nasty"
|
||||
store.refusals = 1
|
||||
with caplog.at_level(logging.ERROR, logger="dashboards.poller"):
|
||||
assert await job.run_once(store) is False
|
||||
assert job.state.ok is False
|
||||
assert job.state.error and "database is locked" in job.state.error
|
||||
assert job.state.payload == {"value": 42.0}, "a refused write must not blank the card"
|
||||
assert job.state.last_ok is not None and job.state.last_attempt is not None
|
||||
assert job.state.last_attempt >= job.state.last_ok
|
||||
assert "could not record a good poll" in caplog.text
|
||||
finally:
|
||||
store.close()
|
||||
|
||||
|
||||
async def test_a_refusing_store_keeps_the_loop_polling_and_says_so_once(
|
||||
tmp_path: Any, caplog: pytest.LogCaptureFixture) -> None:
|
||||
store = RefusingStore(tmp_path / "locked.sqlite3", refusals=2)
|
||||
try:
|
||||
job = Job("nut", "ups", "NUT server", 0.5, lambda: asyncio.sleep(0, ok_result()))
|
||||
with caplog.at_level(logging.ERROR, logger="dashboards.poller"):
|
||||
for _ in range(3):
|
||||
await job.run_once(store)
|
||||
assert job.state.ok is True, "the third poll got through, so the source is ok again"
|
||||
assert job.state.last_ok is not None
|
||||
kinds = [e.kind for e in store.recent_events(limit=10)]
|
||||
assert kinds.count("store-error") == 1, "one line per outage, not one per refused poll"
|
||||
assert "recording" in kinds, "the recovery must be as visible as the failure"
|
||||
assert "could not record a good poll" in caplog.text
|
||||
assert "database is locked" in caplog.text, "the log must carry the real reason"
|
||||
finally:
|
||||
store.close()
|
||||
|
||||
|
||||
async def test_last_good_payload_survives_a_refused_write(tmp_path: Any) -> None:
|
||||
"""Invariant 4 with the store, not the upstream, at fault: keep the numbers, age them."""
|
||||
store = RefusingStore(tmp_path / "locked.sqlite3", refusals=99)
|
||||
try:
|
||||
job = Job("nut", "ups", "NUT server", 0.5, lambda: asyncio.sleep(0, ok_result(42.0)))
|
||||
await job.run_once(store)
|
||||
assert job.state.payload == {"value": 42.0}, "a refused write must not blank the card"
|
||||
assert job.state.last_ok is None, "and it must not be called current either"
|
||||
assert job.state.error and "the store refused it" in job.state.error
|
||||
assert job.state.attempts == 1
|
||||
finally:
|
||||
store.close()
|
||||
|
||||
|
||||
async def test_an_unexpected_error_cannot_kill_the_poll_loop(
|
||||
store, caplog: pytest.LogCaptureFixture) -> None:
|
||||
"""The task *is* a source's polling; an escaping bug must not retire it in silence.
|
||||
|
||||
Four dev-server sources once stopped advancing `attempts` while `/api/health` went on
|
||||
reporting them `ok` — an exception that escaped the loop body, retrieved by nobody. This
|
||||
guard is the reason that class of failure can only ever be loud.
|
||||
"""
|
||||
job = Job("host:gw", "system", "gw", 0.5, lambda: asyncio.sleep(0, ok_result()))
|
||||
real = job.run_once
|
||||
calls = 0
|
||||
|
||||
async def exploding(incoming: SampleStore) -> bool:
|
||||
nonlocal calls
|
||||
calls += 1
|
||||
if calls == 1:
|
||||
raise RuntimeError("a bug in the bookkeeping")
|
||||
return await real(incoming)
|
||||
|
||||
job.run_once = exploding # type: ignore[method-assign]
|
||||
with caplog.at_level(logging.ERROR, logger="dashboards.poller"):
|
||||
job.start(store)
|
||||
try:
|
||||
await asyncio.sleep(1.4) # the interval floor is 0.5s, so this is ~3 polls
|
||||
finally:
|
||||
await job.stop()
|
||||
assert calls >= 2, "the loop must still be calling the poll body after the escape"
|
||||
assert job.state.attempts >= 1 and job.state.ok is True, "and the next poll must have landed"
|
||||
assert "raised outside its own error handling" in caplog.text
|
||||
|
||||
@@ -0,0 +1,425 @@
|
||||
"""The payload contract behind the UPS hero row and battery glyph (phase 05, task 01).
|
||||
|
||||
Phase 05 puts a battery glyph and three big colour-coded numbers — charge %, load %, runtime
|
||||
left — at the top of every UPS card. Nothing is rendered here until the data is proven
|
||||
present, for the same reason phase 03 pinned `/system` first: a card that shows `n/a` gets
|
||||
"fixed" by inventing a field, and an invented field is a number the driver never reported
|
||||
(`AGENTS.md` invariant 2, rule 7).
|
||||
|
||||
What makes this file different from `tests/test_nut.py` is *where the variables come from*.
|
||||
That file's `VAR_ROWS` is a hand-written table; this one drives
|
||||
`tests/fixtures/nut_apc.vars` and `nut_cyberpower.vars` — verbatim `LIST UPS` / `LIST VAR` /
|
||||
`LIST CMD` replies recorded from the real upsd by `./scripts/capture-fixtures.sh` — because
|
||||
the two batteries disagree in ways a typed table never would:
|
||||
|
||||
* the APC's `battery.date` is the factory placeholder `2001/09/25` while its
|
||||
`battery.mfr.date` is the real one (2021/06/24 → a battery past its window), and the
|
||||
separator is `/`, not the `-` every other date in this app uses;
|
||||
* the CyberPower answers `battery.mfr.date: CPS`, which is not a date and must stay "unknown"
|
||||
rather than become a 0-year-old or 56-year-old battery;
|
||||
* the driver's own `identity.model` is `GX1500U` while the operator's label — the thing on
|
||||
the card header — is `PR1500LCDRT2U`;
|
||||
* **neither** driver exports `battery.temperature` **or** `battery.charger.status`, so the
|
||||
glyph's on-battery state has to come from `on_battery` and `charger_status` must be
|
||||
optional;
|
||||
* only the CyberPower exports `output.voltage`.
|
||||
|
||||
The path table itself lives in `tests/ups_contract.py` so the live suite
|
||||
(`tests/test_live_fleet.py::test_ups_hero_fields_are_present_on_the_real_batteries`) asserts
|
||||
the same list against real hardware and the two cannot drift apart.
|
||||
|
||||
Two decisions the task asked to be written down, both asserted below:
|
||||
|
||||
* **The glyph derives its state from `charger_status` + `on_battery`.** The NUT flags are
|
||||
already parsed server-side into `states[]` (`UPS_STATE_LABELS` / `CHARGER_STATUS`);
|
||||
re-parsing `ups.status` in JavaScript duplicates that map and lets it disagree with the API.
|
||||
* **Runtime prose stays `Dash.duration()`-shaped** (`10m 21s`, `1h 32m`) through the page's
|
||||
existing `runtimeText()`, and a runtime nobody measured is `Dash.naText` — `n/a`, never `0s`.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from dashboards.app import create_app
|
||||
from dashboards.config import Settings
|
||||
from dashboards.severity import classify
|
||||
from dashboards.sources import nut
|
||||
from dashboards.store import SampleStore
|
||||
from ups_contract import (
|
||||
HERO_FIELDS,
|
||||
UPS_SHAPES,
|
||||
assert_ups_hero_contract,
|
||||
parse_capture,
|
||||
with_vars,
|
||||
)
|
||||
|
||||
#: Deterministic "now" for the age maths (same instant `tests/test_nut.py` uses).
|
||||
NOW = 1_767_225_600.0
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Fixtures
|
||||
# --------------------------------------------------------------------------- #
|
||||
@pytest.fixture(scope="module")
|
||||
def captures() -> dict[str, nut.UpsReading]:
|
||||
"""Each real UPS as the driver reported it — the shape the hero row must tolerate."""
|
||||
return {name: parse_capture(name) for name in UPS_SHAPES}
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def hero_cards(captures: dict[str, nut.UpsReading]) -> dict[str, dict[str, Any]]:
|
||||
"""`payload()` of each capture: exactly what `/api/ups/snapshot` serves per UPS."""
|
||||
return {name: nut.payload(reading, now=NOW) for name, reading in captures.items()}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(tmp_path: Path) -> Iterator[TestClient]:
|
||||
"""A running app, so the page-side assertions run on the bytes a browser would get.
|
||||
|
||||
There is no JS test runner (a locked anchor), so what can be checked is the served
|
||||
script: which helper formats the runtime, and what it returns when there is no runtime.
|
||||
"""
|
||||
db = str(tmp_path / "ups-contract.sqlite3")
|
||||
store = SampleStore(db)
|
||||
app = create_app(Settings(db_path=db), store)
|
||||
test_client = TestClient(app)
|
||||
test_client.__enter__()
|
||||
try:
|
||||
yield test_client
|
||||
finally:
|
||||
test_client.__exit__(None, None, None)
|
||||
store.close()
|
||||
|
||||
|
||||
def _code(body: str) -> str:
|
||||
"""An asset with its comments stripped — prose may *name* a flag the code must not parse."""
|
||||
body = re.sub(r"/\*.*?\*/", " ", body, flags=re.S)
|
||||
return re.sub(r"^\s*//.*$", " ", body, flags=re.M)
|
||||
|
||||
|
||||
def _function(js: str, name: str) -> str:
|
||||
"""One top-level function's source, from its signature to its closing brace at column 0."""
|
||||
start = js.index(f"function {name}(")
|
||||
return js[start:js.index("\n}", start) + 2]
|
||||
|
||||
|
||||
def _na_text(client: TestClient) -> str:
|
||||
"""The app's own `n/a`, read out of `app.js` rather than repeated here."""
|
||||
match = re.search(r"const naText = '([^']*)';", client.get("/static/app.js").text)
|
||||
assert match is not None, "app.js no longer defines the shared naText"
|
||||
return match.group(1)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# The captures themselves are real, and they are the ones the contract names
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_every_captured_ups_is_in_the_contract() -> None:
|
||||
"""A third UPS on the wire must join `UPS_SHAPES`, or its driver goes uncontracted."""
|
||||
on_disk = {path.stem.removeprefix("nut_")
|
||||
for path in (Path(__file__).parent / "fixtures").glob("nut_*.vars")}
|
||||
assert on_disk == set(UPS_SHAPES)
|
||||
|
||||
|
||||
def test_each_capture_is_a_real_upsd_reply_not_an_empty_stub(captures: dict[str, nut.UpsReading],
|
||||
hero_cards: dict[str, dict[str, Any]]) -> None:
|
||||
"""50+ variables, a status, a description, advertised commands.
|
||||
|
||||
Without this, a truncated fixture would satisfy "the key exists" and the contract would
|
||||
prove nothing: the whole point of capturing the wire is that reality has 50 variables in
|
||||
it, not the 15 anybody would type from memory.
|
||||
"""
|
||||
for name, reading in captures.items():
|
||||
card = hero_cards[name]
|
||||
assert len(reading.vars) > 40, f"{name}: {len(reading.vars)} vars — this is not a capture"
|
||||
assert reading.commands, f"{name}: no CMD lines recorded"
|
||||
assert card["description"], f"{name}: LIST UPS gave no description"
|
||||
assert card["status"], f"{name}: ups.status is missing — every state would be unknown"
|
||||
assert card["identity"]["model"], f"{name}: no model to put beside the numbers"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Every field the hero row and the glyph read
|
||||
# --------------------------------------------------------------------------- #
|
||||
@pytest.mark.parametrize("name", UPS_SHAPES)
|
||||
def test_every_hero_and_glyph_field_is_served_by_a_real_driver(name: str,
|
||||
hero_cards: dict[str, dict[str, Any]]) -> None:
|
||||
assert_ups_hero_contract(hero_cards[name])
|
||||
|
||||
|
||||
def test_the_hero_paths_come_from_the_table_not_from_memory(hero_cards: dict[str, dict[str, Any]]) -> None:
|
||||
"""The table is the contract: every listed path is under `numbers`/`severity`/`health` or
|
||||
one of the three glyph fields, and the three hero numbers each travel with a verdict."""
|
||||
assert set(HERO_FIELDS) >= {"charge %", "load %", "runtime left", "glyph state", "card header"}
|
||||
for label, paths in HERO_FIELDS.items():
|
||||
assert paths, f"{label} lists no payload path"
|
||||
for card in hero_cards.values():
|
||||
for metric in ("charge_pct", "load_pct", "battery_runtime"):
|
||||
assert card["severity"][metric] == classify(f"ups.{metric}", card["numbers"][metric]), (
|
||||
f"{metric}: the card colour does not match severity.py"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", UPS_SHAPES)
|
||||
def test_the_hero_numbers_are_plausible_on_the_real_batteries(name: str,
|
||||
hero_cards: dict[str, dict[str, Any]]) -> None:
|
||||
"""Not just present: sane. 273 V on the battery line is the classic unit bug here."""
|
||||
numbers = hero_cards[name]["numbers"]
|
||||
assert 0.0 <= numbers["charge_pct"] <= 100.0
|
||||
assert 0.0 <= numbers["load_pct"] <= 200.0, "over 100 % is legal (overloaded), 0-200 is the honest band"
|
||||
assert numbers["battery_runtime"] > 0, "a runtime of 0 while online means the estimate is not reported"
|
||||
assert 12.0 <= numbers["battery_voltage"] <= 60.0, "these are 24 V battery packs, not line voltage"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# The real-world shape: what the drivers do *not* report
|
||||
# --------------------------------------------------------------------------- #
|
||||
@pytest.mark.parametrize("name", UPS_SHAPES)
|
||||
def test_neither_real_driver_reports_a_battery_temperature(name: str,
|
||||
hero_cards: dict[str, dict[str, Any]]) -> None:
|
||||
"""`None`, never `0 °C` — and never a key the page has to guess at (invariant 2)."""
|
||||
card = hero_cards[name]
|
||||
assert "battery_temperature" in card["numbers"]
|
||||
assert card["numbers"]["battery_temperature"] is None
|
||||
check = next(c for c in card["health"]["checks"] if c["name"] == "Battery temperature")
|
||||
assert check["status"] == "unknown"
|
||||
assert check["value"] is None
|
||||
|
||||
|
||||
def test_the_apc_date_is_a_placeholder_and_its_mfr_date_is_the_real_one(
|
||||
captures: dict[str, nut.UpsReading], hero_cards: dict[str, dict[str, Any]]) -> None:
|
||||
"""`battery.date` means nothing on this APC; `battery.mfr.date` is the fact (AGENTS.md)."""
|
||||
apc = captures["apc"]
|
||||
# the driver's own separator (`2001/09/25`), normalised here so the assertion is about the
|
||||
# date and not about whether this NUT build used slashes
|
||||
assert apc.vars["battery.date"].replace("/", "-") == "2001-09-25", \
|
||||
"the factory placeholder this driver ships"
|
||||
assert (apc.get("battery_date") or "").replace("/", "-") == "2021-06-24", \
|
||||
"VAR_MAP must prefer battery.mfr.date over the placeholder"
|
||||
age = next(c for c in hero_cards["apc"]["health"]["checks"] if c["name"] == "Battery age")
|
||||
assert age["value"] is not None and 4.0 <= age["value"] < 6.0, age
|
||||
assert age["status"] == "warn", "a 2021 battery is past the replacement window; the card says so"
|
||||
|
||||
|
||||
def test_the_cyberpower_date_is_not_a_date_so_its_age_stays_unknown(
|
||||
captures: dict[str, nut.UpsReading], hero_cards: dict[str, dict[str, Any]]) -> None:
|
||||
"""`battery.mfr.date: CPS` is a vendor string. Unknown, never an age invented from it."""
|
||||
cyberpower = captures["cyberpower"]
|
||||
assert cyberpower.vars["battery.mfr.date"] == "CPS"
|
||||
age = next(c for c in hero_cards["cyberpower"]["health"]["checks"] if c["name"] == "Battery age")
|
||||
assert age["status"] == "unknown" and age["value"] is None
|
||||
|
||||
|
||||
def test_the_card_header_names_the_box_the_operator_labels_it(
|
||||
hero_cards: dict[str, dict[str, Any]]) -> None:
|
||||
"""The driver calls the CyberPower `GX1500U`; the UPS on the shelf is a PR1500LCDRT2U."""
|
||||
card = hero_cards["cyberpower"]
|
||||
assert card["identity"]["model"] == "GX1500U"
|
||||
assert "PR1500LCDRT2U" in card["description"], "LIST UPS' description is the operator's label"
|
||||
assert hero_cards["apc"]["identity"]["mfr"] == "American Power Conversion"
|
||||
|
||||
|
||||
def test_the_two_drivers_disagree_about_what_they_publish(hero_cards: dict[str, dict[str, Any]]) -> None:
|
||||
"""One `numbers` dict, two drivers: a missing variable is `None` on one, a number on the other."""
|
||||
assert hero_cards["cyberpower"]["numbers"]["output_voltage"] is not None
|
||||
assert hero_cards["apc"]["numbers"]["output_voltage"] is None, "no output.voltage on the apc driver"
|
||||
assert hero_cards["cyberpower"]["numbers"]["realpower"] is not None
|
||||
assert hero_cards["apc"]["numbers"]["realpower"] is None, "only ups.realpower.nominal there"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# The three verdicts the hero row paints with
|
||||
# --------------------------------------------------------------------------- #
|
||||
@pytest.mark.parametrize("charge,expected", [
|
||||
(100, "ok"),
|
||||
(45, "warn"),
|
||||
(15, "fail"),
|
||||
])
|
||||
def test_charge_severity_trips_the_inverted_threshold(captures: dict[str, nut.UpsReading],
|
||||
charge: int, expected: str) -> None:
|
||||
"""Inverted: a *smaller* number is worse, and `severity.py` owns that direction."""
|
||||
card = nut.payload(with_vars(captures["apc"], **{"battery.charge": str(charge)}), now=NOW)
|
||||
assert card["severity"]["charge_pct"] == expected
|
||||
assert card["numbers"]["charge_pct"] == float(charge), "the verdict is additive, not a replacement"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("runtime,expected", [
|
||||
(3600, "ok"),
|
||||
(400, "warn"),
|
||||
(200, "fail"),
|
||||
])
|
||||
def test_runtime_severity_trips_the_inverted_threshold(captures: dict[str, nut.UpsReading],
|
||||
runtime: int, expected: str) -> None:
|
||||
"""Seconds of runtime left: 10 min is amber, 5 min is red, an hour is fine."""
|
||||
card = nut.payload(with_vars(captures["apc"], **{"battery.runtime": str(runtime)}), now=NOW)
|
||||
assert card["severity"]["battery_runtime"] == expected
|
||||
assert card["numbers"]["battery_runtime"] == float(runtime)
|
||||
|
||||
|
||||
def test_a_runtime_the_driver_cannot_estimate_has_no_verdict(captures: dict[str, nut.UpsReading]) -> None:
|
||||
"""No number → no colour → `n/a`. `None` in, `None` out, never a green pass."""
|
||||
card = nut.payload(with_vars(captures["cyberpower"], **{"battery.runtime": None}), now=NOW)
|
||||
assert card["numbers"]["battery_runtime"] is None, (
|
||||
"`battery.runtime.low` is a shutdown threshold, not time left — borrowing it for the "
|
||||
"hero would print a confident 5m for a battery that may have 30"
|
||||
)
|
||||
assert card["severity"]["battery_runtime"] is None
|
||||
check = next(c for c in card["health"]["checks"] if c["name"] == "Runtime estimate")
|
||||
assert check["status"] == "unknown"
|
||||
# the threshold itself is still published, under its own name, for whoever wants it
|
||||
assert card["numbers"]["battery_runtime_low"] == 300.0
|
||||
|
||||
|
||||
def test_load_severity_trips_the_ordinary_threshold(captures: dict[str, nut.UpsReading]) -> None:
|
||||
"""Load is not inverted: 90 % of rated capacity is the red end, unlike charge/runtime."""
|
||||
for load, expected in (("35", "ok"), ("70", "warn"), ("95", "fail")):
|
||||
card = nut.payload(with_vars(captures["cyberpower"], **{"ups.load": load}), now=NOW)
|
||||
assert card["severity"]["load_pct"] == expected, f"load {load} %"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# The glyph's state: `charger_status` + `on_battery`, never a flag re-parse in JS
|
||||
# --------------------------------------------------------------------------- #
|
||||
@pytest.mark.parametrize("charger,status,expected_on_battery,label", [
|
||||
("ol", "OL", False, "online"),
|
||||
("full", "OL", False, "fully charged"),
|
||||
("chg", "OL CHRG", False, "charging"),
|
||||
("dchg", "OB", True, "discharging"),
|
||||
])
|
||||
def test_the_glyph_states_are_published_as_fields(captures: dict[str, nut.UpsReading], charger: str,
|
||||
status: str, expected_on_battery: bool,
|
||||
label: str) -> None:
|
||||
"""The four states a battery glyph draws, from the fields the API already sends."""
|
||||
card = nut.payload(with_vars(captures["apc"], **{"battery.charger.status": charger,
|
||||
"ups.status": status}), now=NOW)
|
||||
assert card["charger_status"] == charger
|
||||
assert card["on_battery"] is expected_on_battery
|
||||
assert label in [state["label"] for state in card["states"]]
|
||||
|
||||
|
||||
def test_a_driver_that_capitalises_the_charger_status_is_still_a_known_state(
|
||||
captures: dict[str, nut.UpsReading]) -> None:
|
||||
"""`payload()` normalises the token to the `CHARGER_STATUS` keys.
|
||||
|
||||
`states()` and `on_battery` already lowercase what they compare; if the field the glyph
|
||||
reads kept the driver's casing, `charger_status === "full"` would miss on a driver that
|
||||
answers `FULL` and the battery would stop drawing its state.
|
||||
"""
|
||||
card = nut.payload(with_vars(captures["apc"], **{"battery.charger.status": " FULL "}), now=NOW)
|
||||
assert card["charger_status"] == "full"
|
||||
assert "fully charged" in [state["label"] for state in card["states"]]
|
||||
assert_ups_hero_contract(card)
|
||||
|
||||
|
||||
def test_full_charge_on_line_never_claims_the_battery_is_charging(
|
||||
captures: dict[str, nut.UpsReading]) -> None:
|
||||
"""`charger_status == "full"` with `on_battery == False`: the glyph is full, not filling.
|
||||
|
||||
"Charging" is the word an operator reads as "it was flat recently", so a driver that
|
||||
reports a rested, floating, fully-charged battery must not produce it anywhere.
|
||||
"""
|
||||
card = nut.payload(with_vars(captures["apc"], **{"battery.charger.status": "full",
|
||||
"ups.status": "OL"}), now=NOW)
|
||||
assert card["charger_status"] == "full"
|
||||
assert card["on_battery"] is False
|
||||
words = [state["label"] for state in card["states"]] + [card["status"] or "",
|
||||
card["charger_status"] or ""]
|
||||
assert "charging" not in words, f"a full, online UPS is not charging: {words}"
|
||||
assert not any("charging" in word for word in words), words
|
||||
|
||||
|
||||
def test_on_battery_makes_the_glyphs_on_battery_state_available(captures: dict[str, nut.UpsReading],
|
||||
) -> None:
|
||||
"""Both routes to on-battery work, including the one with no `ups.status` at all."""
|
||||
from_flag = nut.payload(with_vars(captures["apc"], **{"ups.status": "OB DISCHRG",
|
||||
"battery.charger.status": None}), now=NOW)
|
||||
assert from_flag["on_battery"] is True
|
||||
assert {"id": "OB", "label": "on battery", "severity": "warn",
|
||||
"detail": "Running on battery"} in from_flag["states"]
|
||||
assert from_flag["charger_status"] is None, "the APC is proof this field is optional"
|
||||
|
||||
from_charger = nut.payload(with_vars(captures["apc"], **{"ups.status": None,
|
||||
"battery.charger.status": "dchg"}), now=NOW)
|
||||
assert from_charger["on_battery"] is True
|
||||
assert "discharging" in [state["label"] for state in from_charger["states"]]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", UPS_SHAPES)
|
||||
def test_the_real_captures_omit_charger_status_so_on_battery_carries_the_state(
|
||||
name: str, captures: dict[str, nut.UpsReading],
|
||||
hero_cards: dict[str, dict[str, Any]]) -> None:
|
||||
"""The assumption, recorded against hardware: `battery.charger.status` is not there.
|
||||
|
||||
Neither usbhid-ups driver exports the NUT 2.8 rich status, so a glyph that only looked at
|
||||
`charger_status` would render "unknown" on two healthy batteries. `on_battery` is derived
|
||||
from `ups.status` server-side and is the field that must drive the glyph here.
|
||||
"""
|
||||
assert "battery.charger.status" not in captures[name].vars
|
||||
card = hero_cards[name]
|
||||
assert card["charger_status"] is None
|
||||
assert card["on_battery"] is False
|
||||
labels = [state["label"] for state in card["states"]]
|
||||
assert "online" in labels, f"the glyph still needs a state to show: {labels}"
|
||||
|
||||
|
||||
def test_the_page_does_not_re_parse_nut_status_flags(client: TestClient) -> None:
|
||||
"""`page-ups.js` may render `states[]` and the raw `status` chip; it may not interpret flags.
|
||||
|
||||
`UPS_STATE_LABELS` and `CHARGER_STATUS` in `sources/nut.py` are the map. A copy of it in
|
||||
JavaScript is a second opinion about the wire, and one of the two will be wrong.
|
||||
"""
|
||||
js = _code(client.get("/static/page-ups.js").text)
|
||||
assert "status.split(" not in js, "the page is parsing ups.status itself"
|
||||
for flag in ("OL", "OB", "LB", "RB", "CHRG", "DISCHRG", "FUB", "CAL", "OVER", "NULL"):
|
||||
assert f'"{flag}"' not in js and f"'{flag}'" not in js, f"the page special-cases the {flag} flag"
|
||||
assert "item.states" in js, "the states the API labelled are what the card renders"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# The page-side formatting of "no runtime"
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_runtime_of_none_is_na_not_zero_seconds(client: TestClient) -> None:
|
||||
"""`runtimeText(null)` is the app's `n/a`. `0s` would be a different, alarming sentence.
|
||||
|
||||
Asserted on the served script because there is no JS test runner — the promise under test
|
||||
is that the runtime cell goes through the one helper and that the helper's unknown branch
|
||||
returns the shared text before any number can be formatted.
|
||||
"""
|
||||
js = client.get("/static/page-ups.js").text
|
||||
na = _na_text(client)
|
||||
assert na == "n/a"
|
||||
|
||||
body = _function(js, "runtimeText")
|
||||
guard = re.search(r"seconds == null.*?return\s+([^;]+);", body)
|
||||
assert guard is not None, "runtimeText() no longer guards against an absent runtime"
|
||||
token = guard.group(1).strip()
|
||||
rendered = na if token == "naText" else token.strip("'\"")
|
||||
assert rendered == na, f"runtimeText() invents its own unknown text: {token}"
|
||||
|
||||
# the unknown branch is first: nothing may format a number before it
|
||||
assert body.index("seconds == null") < body.index("s`;"), "null could reach the number formatter"
|
||||
assert "return '0s'" not in body
|
||||
|
||||
# and the card's runtime cell is written through that helper, from the hero's own number
|
||||
assert re.search(r"runtimeText\([^)]*battery_runtime\s*\)", js), (
|
||||
"the runtime cell no longer formats numbers.battery_runtime through runtimeText()"
|
||||
)
|
||||
|
||||
|
||||
def test_the_page_reads_the_keys_the_contract_lists(client: TestClient) -> None:
|
||||
"""The hero row's keys are the contract's keys, on the served script and nowhere newer.
|
||||
|
||||
Task 02/03 may restructure the card, but a rename here (`n.chargeRemaining`) means the
|
||||
page and `/api/ups/snapshot` no longer speak, and the only symptom is a card of `n/a`.
|
||||
"""
|
||||
js = client.get("/static/page-ups.js").text
|
||||
assert re.search(r"item\.numbers\b", js), "the card no longer reads the payload's numbers object"
|
||||
for key in ("charge_pct", "load_pct", "battery_runtime", "battery_voltage",
|
||||
"battery_temperature", "line_voltage"):
|
||||
assert f".{key}" in js, f"the page never reads numbers.{key} — renamed or dropped"
|
||||
assert "item.health" in js, "the card header still reads health.overall / health.counts"
|
||||
@@ -0,0 +1,148 @@
|
||||
"""The `/ups` hero-row and battery-glyph payload contract, shared by the two suites.
|
||||
|
||||
Phase 05 makes every UPS card open with a battery glyph that visibly holds its charge and
|
||||
three big colour-coded numbers — **charge %**, **load %**, **runtime left** — with the health
|
||||
checks and the command buttons below them. The paths below are the fields that row (and the
|
||||
glyph) may read; they are asserted twice on purpose:
|
||||
|
||||
* hermetically, from **verbatim upsd captures** (`tests/fixtures/nut_apc.vars`,
|
||||
`nut_cyberpower.vars`) in `tests/test_ups_contract.py`, and
|
||||
* against the **real batteries** in
|
||||
`tests/test_live_fleet.py::test_ups_hero_fields_are_present_on_the_real_batteries`.
|
||||
|
||||
One table, so the live assertions cannot drift away from the fixture assertions
|
||||
(the same reasoning as `tests/card_contract.py` for `/system`).
|
||||
|
||||
=========================== ==============================================================
|
||||
Hero element Path in one card of `/api/ups/snapshot`
|
||||
=========================== ==============================================================
|
||||
charge % ``numbers.charge_pct`` + ``severity.charge_pct``
|
||||
load % ``numbers.load_pct`` + ``severity.load_pct``
|
||||
runtime left ``numbers.battery_runtime`` + ``severity.battery_runtime``
|
||||
battery volts ``numbers.battery_voltage``
|
||||
battery °C ``numbers.battery_temperature`` (``None`` on both real UPSes)
|
||||
glyph state ``charger_status`` + ``on_battery`` (+ ``states`` for a label)
|
||||
card header ``health.overall`` + ``health.counts``
|
||||
=========================== ==============================================================
|
||||
|
||||
**The glyph's state is ``charger_status`` and ``on_battery``, never a re-parse of
|
||||
``ups.status`` in JavaScript.** ``UpsReading`` already turns the flags into ``states[]`` with
|
||||
labels and severities (`UPS_STATE_LABELS` / `CHARGER_STATUS`), and a second copy of that map
|
||||
in a page is a map that can disagree with the API. Recorded here because both captures prove
|
||||
the fallback is not theoretical: **neither the APC nor the CyberPower publishes
|
||||
``battery.charger.status`` at all**, so on this fleet ``on_battery`` is the field that says
|
||||
whether the glyph is holding charge or draining, and ``charger_status`` must stay optional.
|
||||
|
||||
Values may honestly be ``None`` (that renders `n/a`, per the "never invent a number" rule);
|
||||
**present** is the contract.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from card_contract import dig
|
||||
from dashboards.sources import nut
|
||||
|
||||
FIXTURES = Path(__file__).resolve().parent / "fixtures"
|
||||
|
||||
#: The UPS shapes the captures in `tests/fixtures/` cover. A new driver joins this list and
|
||||
#: gets captured with `./scripts/capture-fixtures.sh` — `test_every_captured_ups_is_in_the_contract`
|
||||
#: fails if it does not, the way `HOST_SHAPES` does for `/system`.
|
||||
UPS_SHAPES: tuple[str, ...] = ("apc", "cyberpower")
|
||||
|
||||
#: Hero row / glyph element -> dotted payload paths inside one UPS card. Present is the
|
||||
#: contract; a value may be ``None`` on a driver that does not report it.
|
||||
HERO_FIELDS: dict[str, tuple[str, ...]] = {
|
||||
"charge %": ("numbers.charge_pct", "severity.charge_pct"),
|
||||
"load %": ("numbers.load_pct", "severity.load_pct"),
|
||||
"runtime left": ("numbers.battery_runtime", "severity.battery_runtime"),
|
||||
"battery volts": ("numbers.battery_voltage",),
|
||||
"battery °C": ("numbers.battery_temperature",),
|
||||
# the glyph: the rich status when the driver has one, `on_battery` when it does not,
|
||||
# `states[]` for the label, and `status` only ever as the raw mono chip beside it
|
||||
"glyph state": ("charger_status", "on_battery", "states", "status"),
|
||||
# unchanged since before phase 05 — the card header keeps using these
|
||||
"card header": ("health.overall", "health.counts"),
|
||||
}
|
||||
|
||||
#: The verdicts the hero row knows. `severity.py`'s vocabulary and nothing else; `unknown`
|
||||
#: is `assess()`'s word for a condition, not a colour for a value.
|
||||
VERDICTS: tuple[str | None, ...] = ("ok", "warn", "fail", None)
|
||||
|
||||
#: The health-check rows the hero row depends on being able to explain itself with.
|
||||
CHECK_NAMES: tuple[str, ...] = ("Charge", "Runtime estimate", "Load", "Battery temperature")
|
||||
|
||||
|
||||
def parse_capture(name: str) -> nut.UpsReading:
|
||||
"""Rebuild an `UpsReading` from a verbatim upsd capture.
|
||||
|
||||
The lines are split with the client's own `_split_fields()` rather than a second parser:
|
||||
upsd quotes a field only when it must (`VAR apc ups.status "OL"` is what NUT 2.8.5
|
||||
actually sends), so a hand-rolled splitter would be a second opinion about the wire —
|
||||
exactly the thing `AGENTS.md` rule 7 refuses to have.
|
||||
"""
|
||||
path = FIXTURES / f"nut_{name}.vars"
|
||||
if not path.exists():
|
||||
raise AssertionError(
|
||||
f"missing capture {path.name} — run ./scripts/capture-fixtures.sh with the upsd "
|
||||
f"forwarded (DEV_FORWARDS); the UPS hero row is contracted against real drivers"
|
||||
)
|
||||
variables: dict[str, str] = {}
|
||||
commands: list[str] = []
|
||||
description = ""
|
||||
for line in path.read_text().splitlines():
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
fields = nut._split_fields(line)
|
||||
if fields[0] == "UPS" and len(fields) >= 2:
|
||||
description = " ".join(fields[2:])
|
||||
elif fields[0] == "VAR" and len(fields) >= 4:
|
||||
variables[fields[2]] = " ".join(fields[3:])
|
||||
elif fields[0] == "CMD" and len(fields) >= 3:
|
||||
commands.append(fields[2])
|
||||
return nut.UpsReading(name=name, description=description, vars=variables,
|
||||
commands=commands, ts=1_767_225_600.0)
|
||||
|
||||
|
||||
def with_vars(reading: nut.UpsReading, **overrides: str | None) -> nut.UpsReading:
|
||||
"""The same capture with a few variables changed (or removed with ``None``).
|
||||
|
||||
Changing one variable instead of writing a new VAR table keeps every *other* real value
|
||||
in the payload, so a test about charge severity is still a test about a real APC.
|
||||
"""
|
||||
variables = dict(reading.vars)
|
||||
for key, value in overrides.items():
|
||||
if value is None:
|
||||
variables.pop(key, None)
|
||||
else:
|
||||
variables[key] = value
|
||||
return nut.UpsReading(name=reading.name, description=reading.description, vars=variables,
|
||||
commands=list(reading.commands), ts=reading.ts)
|
||||
|
||||
|
||||
def assert_ups_hero_contract(card: dict[str, Any]) -> None:
|
||||
"""Every hero/glyph path exists on this UPS card, with the vocabulary the page renders."""
|
||||
for paths in HERO_FIELDS.values():
|
||||
for path in paths:
|
||||
dig(card, path)
|
||||
verdicts = card["severity"]
|
||||
assert isinstance(verdicts, dict), "severity must be a dict of hero verdicts"
|
||||
for key in ("charge_pct", "load_pct", "battery_runtime"):
|
||||
assert key in verdicts, f"severity.{key} is missing — the hero would colour nothing"
|
||||
assert verdicts[key] in VERDICTS, f"severity.{key}={verdicts[key]!r} is not the app vocabulary"
|
||||
assert isinstance(card["on_battery"], bool), "on_battery is a bool, not a truthy string"
|
||||
charger = card["charger_status"]
|
||||
assert charger is None or charger in nut.CHARGER_STATUS, (
|
||||
f"charger_status={charger!r} is not a CHARGER_STATUS token (and not None) — the glyph "
|
||||
f"would have no state to draw"
|
||||
)
|
||||
states = card["states"]
|
||||
assert isinstance(states, list), "states is a list of {id,label,severity} chips"
|
||||
for state in states:
|
||||
assert {"id", "label", "severity"} <= set(state), f"state chip is incomplete: {state}"
|
||||
counts = card["health"]["counts"]
|
||||
assert {"ok", "warn", "fail", "unknown"} <= set(counts), f"health.counts: {sorted(counts)}"
|
||||
names = {check["name"] for check in card["health"]["checks"]}
|
||||
missing = set(CHECK_NAMES) - names
|
||||
assert not missing, f"health.checks has no {sorted(missing)} row — the card cannot explain itself"
|
||||
Reference in New Issue
Block a user