Files
ducoterra a08e693698 phase: 10_dev_direct_dial
**Phase 10 final pass — repo green; live criteria blocked off-repo.**

- Verified: `gateway` runs `dashboards-agent` (active/enabled, `10.1.1.27:9120` → `/health` 200, `/snapshot` 401 no-token / 200 bearer on-host, token matches `/etc/dashboards-agent.token`); `driveripper`/`deskwork` inactive, `sudo -n` → "a password is required", `:9120` empty.
- New measurement (transient runtime rule on gateway `firewall-cmd --add-port=9120/tcp`, **reverted** — runtime+permanent port lists verified back to the original set, no rich rules): laptop still gets no SYN through, while `:22`/`:80` on the same address answer in ms ⇒ the block is the lab gateway `10.2.0.1` port filter, not this app/host firewall/agent. No other live change made (no sudoers, no upsd, no UPS command).
- Tests/lint (exact): `bash .agents/validate.sh` → **EXIT 0** — `uv run pytest --cov --cov-fail-under=94` **793 passed, 11 skipped, coverage 98.33 %**; `ruff check src tests agent` clean; `pyright` 0 errors; `node --check` clean; hygiene clean; `uv pip list --outdated` empty (bumped fastapi 0.143.3→0.143.5, `uv.lock` only).
- `uv run pytest -m live --live` (`.env.dev` sourced): **4 passed, 1 skipped, 6 failed** — all six are `:9120`/`:3051` connect timeouts; zero parse/contract failures.
- Dev server restarted on current code: engines `ok`; 3×`host` + 3×`disks` + `nut` `DOWN` with real reasons (designed degradation, stub rows intact, no credential leakage).
- Completion criteria: screenshots ✓ (`data/logs/phase10-{index,system,disks,llm,ups}-live.png`); tunnel-vocabulary grep ✓ outside `.agents/` history (PLAN.md locked line, phase overviews, past reports; fixtures/`uv.lock` hits are digit substrings — `tests/test_deploy_scripts.py` guards operational files); no page-payload change ✓ (contract tests). **Not met, operator-side:** agents on all three hosts, `LIST UPS` from the laptop, `dev-server.sh status` all-ok, live suite green.
- Fixed in this pass: `sources/nut.py` unreachable-upsd message no longer prints a bare `(TimeoutError: )` (now `(timed out after 4s)` / `(ConnectionRefusedError: …)`), pinned by a new test + extended refused-port test.
- Docs: README verified consistent (every env var in `config.py` documented, routes/flags/`--with-llm`/pager sizes checked) — no changes needed. AGENTS.md fixed 4 inconsistencies: per-host sudo reality (gateway grants `NOPASSWD ALL`, which is why it installed without `agent-sudoers`; driveripper scoped; deskwork unrelated), the now-proven firewall finding, `/system` table typo ("agent per host"), gap-4 DB/WAL sizes + test count. `.env.example` `upsd.users` sample now one `instcmds` per line (it contradicted README and the in-app copy). `.env.dev` (gitignored) comment corrected to match the evidence.
- Flagged, untouched (locked): `.agents/PLAN.md` L92–95 still says upsd is "forwarded to `127.0.0.1:3494` in dev" — false after this phase; evidence + exact operator commands appended to `data/logs/NOTES-verify.md` §11.
- Next pending phase: `11_admin_login`.
2026-10-18 05:44:13 -04:00

149 lines
7.8 KiB
Python

"""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 NUT_HOST "
f"pointing at the real upsd; 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"