ducoterra
86c8bf4992
Phase 33 completion pass is done. Final report:
## Phase 33 — completion report
**All 8 task files complete** (`.agents/phases/complete/33_gateway_route_switching/`), full gate green, docs verified.
### Work in this pass
1. **`tests/test_routectl.py` added** (17 tests, all green) — exercises the *real* helper binary unprivileged against a fake tree (`--root`), the real agent parser, the captured gateway fixtures, and stubbed `nginx`/`haproxy`/`systemctl` on `PATH`. Pins the full protocol (locate → state → dry-run → temp-file/fsync/`os.replace` write → checker → restore on checker failure → reload-never-restart → re-read proof → sidecar record), all three route shapes (haproxy `server`, nginx `proxy_pass`, nginx stream `map` via its sink-value id), `drift` / unknown-id / `map-default` refusals, `list`/`verify`, and the `--serve` peer gate (`SO_PEERCRED` uid check, one JSON line per request, exact body shape).
2. **Real helper defect fixed** — `deploy/dashboards-routectl.py`: `main()`'s early-return paths returned `({"ok": False, …}, EXIT_*)` tuples, so `sys.exit()` printed a Python repr instead of a JSON line (usage errors, agent-import refusal). Now prints `json.dumps(…)` and returns only the code, keeping the one-JSON-line contract.
3. **Documentation consistency fixes** (task 08 criterion 6):
- `README.md`: garbled verdict-colour sentence on the switch ("green and to the position") → "in the **on** position and tinted `--gate-on`" (the existence-pair contract).
- `README.md`: audit sentence now names all three event kinds — `route-disabled` / `route-enabled` / **`route-denied`** (the code stores one per refused call; dry runs store none).
- `page-gateway.js`: replaced the stale pre-33 module docstring ("the switch is not wired") with the two-arm `GATEWAY_ALLOW_WRITES` contract.
- `AGENTS.md`: every phase-33 claim re-verified against code (the ten gates, the 35 s/30 s budget pair, `gateway.writes_enabled`, `/api/summary`'s `disabled`, the 1.4.0 agent bullet, the gap list) — no changes needed.
### Verification
- `uv run pytest --cov`: **1970 passed**, 18 skipped — coverage **98.74%** (floor 94%)
- `ruff` clean · `pyright` 0 errors · `bash .agents/validate.sh` **exit 0** (incl. `node --check` on every page script)
- Dev server running; all 10 sources `ok`; bundles exported
### Live infrastructure
**No live changes made.** The live drill remains unexecuted per `data/logs/NOTES-33-switching.md` (no operator approval; checker binaries unavailable in the podman proxy units). All routes stay enabled, config hashes unchanged, and the sandbox rehearsal + backup tree are in place for the operator's own `DANGER` call. All changes are in the working tree only.
687 lines
38 KiB
Python
687 lines
38 KiB
Python
"""The only tests that touch a network: `uv run pytest -m live --live`.
|
|
|
|
Everything else in this suite is hermetic on purpose (see conftest.py). These are the
|
|
bridge to reality — they re-run the *production* code paths (`collectors.fetch_*`,
|
|
`NutMonitor.collect`) against real targets, which is what catches an upstream that
|
|
renamed a series or a driver that started reporting a temperature in kelvin.
|
|
|
|
They read exactly the env vars the app reads, so the dev loop is direct dial — the same
|
|
addresses the deployed dashboard dials, nothing forwarded:
|
|
|
|
set -a; . .env.dev; set +a
|
|
uv run pytest -m live --live
|
|
|
|
Each test skips when its source is not configured, so running the whole set with half a
|
|
fleet reachable is a normal outcome, not a failure.
|
|
|
|
**Nothing in here may write — with exactly one documented exception.** The UPS test asserts
|
|
the command gating is honest; it never sends an instant command — a battery test on
|
|
hardware someone else is plugged into is not a unit test. The one write is the phase-33
|
|
round-trip test, gated by its own flag (`--allow-live-writes`) on top of `--live`: a
|
|
default live run never switches a route, and the skip reason names the route it would.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
from collections.abc import Iterator
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from card_contract import assert_card_contract
|
|
from dashboards import certs, collectors
|
|
from dashboards.config import Settings, load
|
|
from dashboards.rates import RateEstimator
|
|
from dashboards.severity import classify
|
|
from dashboards.sources import gateway, nut
|
|
from dashboards.sources.disks import SMART_SERIES, DisksCollector
|
|
from dashboards.sources.host import HostCollector
|
|
from slots_contract import assert_slot_shape
|
|
from ups_contract import assert_ups_hero_contract
|
|
|
|
pytestmark = pytest.mark.live
|
|
|
|
|
|
def endpoint_identity(out: Any) -> tuple[Any, ...] | None:
|
|
"""One endpoint in the kernel's currency, derived **here** rather than by the source.
|
|
|
|
An address endpoint is its own address; a name endpoint is the addresses the agent's resolver
|
|
published for it (`out.resolved`), normalised so `::ffff:10.1.2.64` and `10.1.2.64` are one
|
|
key. `None` is "nothing can name this far end", which is counted nowhere. Phase 32 made this
|
|
helper necessary: a live router can have two route lines reaching one address through two
|
|
different names, and a total keyed on the config's text would book it twice.
|
|
"""
|
|
port = out.get("port")
|
|
if not isinstance(port, int):
|
|
return None
|
|
spellings = ([out["host"]] if out.get("address") is True
|
|
else list(out.get("resolved") or []))
|
|
keys: list[str] = []
|
|
for spelling in spellings:
|
|
try:
|
|
address = ipaddress.ip_address(str(spelling))
|
|
except ValueError:
|
|
continue
|
|
mapped = getattr(address, "ipv4_mapped", None)
|
|
keys.append(str(mapped if mapped is not None else address))
|
|
return (min(keys), port) if keys else None
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def settings() -> Settings:
|
|
"""Live env, not the import-time SETTINGS singleton."""
|
|
return load()
|
|
|
|
|
|
class _StoredAuthority:
|
|
"""`certs.MetaStore` over an existing database, opened **read-only**.
|
|
|
|
The live suite has to dial the fleet the way the app does, and since the phase-24 rollout that
|
|
means presenting this dashboard's own client certificate. The authority lives in the `meta`
|
|
table of the store the dev server owns, so this is the smallest honest reader of it: `mode=ro`
|
|
on the URI, `get_meta` the only method that does anything, and the two write methods a
|
|
*refusal* rather than a no-op — this file's docstring says nothing in here may write, and
|
|
minting or repairing an authority from a test would be the worst write it could make.
|
|
"""
|
|
|
|
def __init__(self, path: str) -> None:
|
|
self._conn = sqlite3.connect(f"file:{os.path.abspath(path)}?mode=ro", uri=True)
|
|
|
|
def close(self) -> None:
|
|
self._conn.close()
|
|
|
|
def get_meta(self, key: str) -> str | None:
|
|
row = self._conn.execute("SELECT value FROM meta WHERE key = ?", (key,)).fetchone()
|
|
return None if row is None else str(row[0])
|
|
|
|
def set_meta(self, key: str, value: str) -> None:
|
|
raise AssertionError(f"live tests are read-only: refused to write {key!r}")
|
|
|
|
def delete_meta(self, key: str) -> int:
|
|
raise AssertionError(f"live tests are read-only: refused to delete {key!r}")
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def agent_tls(settings: Settings) -> Iterator[dict[str, collectors.AgentTls]]:
|
|
"""One :class:`collectors.AgentTls` per `https://` host, from the authority in the store.
|
|
|
|
Empty when no `HOST_TARGETS` entry is https — every host is then dialled on the pre-phase-15
|
|
token path, which is what an `http://` fleet still is — and skipped when the store or its
|
|
`certs.*` rows are absent, because a box running the live suite without the dev database is a
|
|
normal partial environment rather than a failure (the same rule that lets half a fleet be
|
|
reachable).
|
|
|
|
`certs.read_state` is the request-time read and never `ensure_certs`: a live test may not mint
|
|
a root CA, and the material this writes is removed on teardown.
|
|
"""
|
|
targets = settings.host_targets
|
|
if not any(target.url.startswith("https://") for target in targets):
|
|
yield {}
|
|
return
|
|
if not os.path.exists(settings.db_path):
|
|
pytest.skip(f"no store at {settings.db_path} to read the CA from")
|
|
authority = _StoredAuthority(settings.db_path)
|
|
try:
|
|
state = certs.read_state(authority)
|
|
except (RuntimeError, sqlite3.Error) as exc:
|
|
authority.close()
|
|
pytest.skip(f"no readable authority in {settings.db_path}: {type(exc).__name__}")
|
|
material = certs.write_material(state)
|
|
try:
|
|
yield collectors.agent_tls_map(material, authority, targets)
|
|
finally:
|
|
certs.remove_material(material)
|
|
authority.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def host_collector() -> HostCollector:
|
|
return HostCollector(rates=RateEstimator(span=10.0))
|
|
|
|
|
|
async def test_every_host_agent_answers_and_parses(settings: Settings,
|
|
host_collector: HostCollector,
|
|
agent_tls: dict[str, collectors.AgentTls]) -> None:
|
|
"""Each HOST_TARGETS entry must answer /snapshot and yield the metrics the page renders."""
|
|
if not settings.host_targets:
|
|
pytest.skip("HOST_TARGETS is unset")
|
|
required = {"cpu.usage_pct", "mem.used_pct", "disk.used_pct"}
|
|
async with httpx.AsyncClient(timeout=settings.host_timeout) as client:
|
|
for target in settings.host_targets:
|
|
tls = agent_tls.get(target.name)
|
|
# Poll twice: usage_pct is a Δ/jiffies rate, and a rate has no second point yet
|
|
# on the first scrape. That is the design (the agent never remembers who polled
|
|
# it), so a live test that scraped once would be asserting a false failure.
|
|
await collectors.fetch_host_snapshot(client, target, host_collector, tls=tls)
|
|
await asyncio.sleep(1.2)
|
|
result = await collectors.fetch_host_snapshot(client, target, host_collector, tls=tls)
|
|
metrics = {sample.metric for sample in result.samples}
|
|
assert result.payload["info"]["hostname"], f"{target.name}: no hostname"
|
|
missing = required - metrics
|
|
assert not missing, f"{target.name}: snapshot parsed but no {sorted(missing)}"
|
|
# a section the agent could not read is a `degraded` note, never a silent zero
|
|
assert result.payload.get("degraded") in (None, []), (
|
|
f"{target.name}: agent reported {result.payload.get('degraded')}"
|
|
)
|
|
|
|
|
|
async def test_every_host_serves_the_hero_card_contract(settings: Settings,
|
|
host_collector: HostCollector,
|
|
agent_tls: dict[str, collectors.AgentTls]) -> None:
|
|
"""The phase-03 `/system` card paths exist on the **real** agents, not just the captures.
|
|
|
|
Values may honestly be `n/a` on real hardware (gateway reports no CPU sensor and no GPU
|
|
at all); the *shape* is what phase 03 renders, so this is the test that catches an agent
|
|
upgrade renaming a field. Hermetic twin: `tests/test_system_contract.py`.
|
|
"""
|
|
if not settings.host_targets:
|
|
pytest.skip("HOST_TARGETS is unset")
|
|
async with httpx.AsyncClient(timeout=settings.host_timeout) as client:
|
|
for target in settings.host_targets:
|
|
tls = agent_tls.get(target.name)
|
|
await collectors.fetch_host_snapshot(client, target, host_collector, tls=tls)
|
|
await asyncio.sleep(1.2)
|
|
result = await collectors.fetch_host_snapshot(client, target, host_collector, tls=tls)
|
|
assert_card_contract(result.payload)
|
|
assert result.payload["cpu"]["temp_c"] == result.payload["temperatures"]["cpu_c"], (
|
|
f"{target.name}: the card and /api/summary would report different CPU temps"
|
|
)
|
|
|
|
|
|
async def test_agent_disk_rows_carry_the_smart_shape(
|
|
settings: Settings, agent_tls: dict[str, collectors.AgentTls]) -> None:
|
|
"""Phase-09: a redeployed agent answers `smart` per physical disk of `/snapshot`.
|
|
|
|
Agents are redeployed by hand, so an agent whose rows carry no `smart` key at all
|
|
SKIPS this host rather than failing it. `smart: null` is a documented state (no
|
|
smartctl binary, or the agent user lacks device read access) — what must hold is the
|
|
shape of a non-null block and that unknowns are null, never 0. Read-only: nothing
|
|
here sends `smartctl -t`; a self-test on hardware something is plugged into is the
|
|
operator's call, not a unit test.
|
|
"""
|
|
if not settings.host_targets:
|
|
pytest.skip("HOST_TARGETS is unset")
|
|
seen_smart = 0
|
|
async with httpx.AsyncClient(timeout=settings.host_timeout) as client:
|
|
for target in settings.host_targets:
|
|
try:
|
|
# through `fetch_snapshot`, not a hand-rolled `get`, because that is the function
|
|
# that owns the certificate dial: an `https://` agent needs this app's own CA and
|
|
# client leaf, and a test dialling it with the platform trust store would fail on
|
|
# a healthy fleet (`CERTIFICATE_VERIFY_FAILED: self-signed certificate in chain`).
|
|
response = await collectors.fetch_snapshot(client, target,
|
|
tls=agent_tls.get(target.name))
|
|
except collectors.UpstreamError:
|
|
continue # reachability has its own live test
|
|
devices = (response.get("disk_io") or {}).get("devices") or []
|
|
if devices and not all("smart" in device for device in devices):
|
|
continue # agent predates the smart pass — redeploy it to test this
|
|
for device in devices:
|
|
seen_smart += 1
|
|
smart = device.get("smart")
|
|
if smart is None:
|
|
continue # degraded by design; the errors[] entry says why
|
|
# the literal set is the pin: a rename on the agent must fail here too
|
|
assert set(smart) == {"model", "serial", "overall_health", "temp_c",
|
|
"power_on_hours", "reallocated_sector_ct",
|
|
"current_pending_sector_ct",
|
|
"offline_uncorrectable", "percentage_used"}, \
|
|
f"{target.name}/{device['name']}"
|
|
if smart["power_on_hours"] is not None:
|
|
assert smart["power_on_hours"] > 0
|
|
if smart["temp_c"] is not None:
|
|
assert -20.0 < smart["temp_c"] < 125.0
|
|
assert smart["overall_health"] in (None, "PASSED", "FAILED", "UNKNOWN")
|
|
if not seen_smart:
|
|
pytest.skip("no agent redeployed with the smart pass yet (deploy/install-agent.sh)")
|
|
|
|
|
|
async def test_disks_source_charts_the_physical_drives_the_agent_reports(
|
|
settings: Settings, agent_tls: dict[str, collectors.AgentTls]) -> None:
|
|
"""Phase 09 task 02: the `disks` source over a real agent snapshot.
|
|
|
|
This is the production code path — the fetch the poller runs, not a hand-built dict —
|
|
so what fails here is a rename between the agent and the page, not a typo in a fixture.
|
|
The assertions hold whether or not this fleet can read SMART yet: one row per whole
|
|
disk, `disk.*@<dev>` series only for values that exist, and no invented numbers.
|
|
"""
|
|
if not settings.host_targets:
|
|
pytest.skip("HOST_TARGETS is unset")
|
|
collector = DisksCollector()
|
|
checked = 0
|
|
async with httpx.AsyncClient(timeout=settings.disks_timeout) as client:
|
|
for target in settings.host_targets:
|
|
try:
|
|
result = await collectors.fetch_disk_snapshot(
|
|
client, target, collector, timeout=settings.disks_timeout,
|
|
tls=agent_tls.get(target.name))
|
|
except collectors.UpstreamError:
|
|
continue # reachability has its own live test
|
|
checked += 1
|
|
payload = result.payload
|
|
assert payload["name"] == target.name
|
|
assert payload["disks"], f"{target.name}: no physical disks at all?"
|
|
for row in payload["disks"]:
|
|
assert row["path"] == f"/dev/{row['device']}"
|
|
assert "/" not in row["device"], f"partition {row['device']} in a disk list"
|
|
assert row["health"] in (None, "ok", "failed")
|
|
devices = {row["device"] for row in payload["disks"]}
|
|
bases = {base for _field, base in SMART_SERIES}
|
|
for sample in result.samples:
|
|
assert sample.source == "disks", "the wrong family would chart nowhere"
|
|
base, _, device = sample.metric.partition("@")
|
|
assert base in bases and device in devices, (
|
|
f"{target.name}: {sample.metric} is not a pinned disk series")
|
|
assert sample.value == sample.value and abs(sample.value) != float("inf"), \
|
|
"NaN must never reach the store"
|
|
if any(row["smart"] for row in payload["disks"]):
|
|
assert payload["smartctl_available"] is True
|
|
if not checked:
|
|
pytest.skip("no host agent answered")
|
|
|
|
|
|
async def test_the_gateway_source_joins_the_live_ingress(settings: Settings,
|
|
agent_tls: dict[str, collectors.AgentTls]) -> None:
|
|
"""Phase 29/30: `sources/gateway.py` over the **real** gateway agent, read-only.
|
|
|
|
The production path — the same `fetch_snapshot` the poller will run, through the stored
|
|
mTLS authority — so what fails here is a drift between the agent's sections and the join,
|
|
not a typo in a fixture. The assertions are the ones that hold whatever the traffic is:
|
|
the agent answers the three sections, every configured frontend and `listen` port is a
|
|
trunk, and the inbound count **re-derived here** from the snapshot's own ESTABLISHED rows
|
|
equals the sum the payload prints — the identity of phase 30's simplified join, checked
|
|
against the kernel rather than against a fixture. The totals are an inequality by design
|
|
(`sum(listeners) + sum(endpoints) <= established`): a socket neither config file describes
|
|
is counted nowhere, and on this router that is most of the table.
|
|
|
|
Phase 32 widened **which** routes can be counted — an endpoint the config wrote as an address,
|
|
or one the gateway's own resolver published for the name it wrote — and the two assertions near
|
|
the end are stated in those terms, with the endpoint identity derived here so the outbound total
|
|
is not simply the source's own arithmetic re-added.
|
|
"""
|
|
target = next((item for item in settings.host_targets if item.name == settings.gateway_host),
|
|
None)
|
|
if target is None:
|
|
pytest.skip(f"GATEWAY_HOST={settings.gateway_host!r} is not in HOST_TARGETS")
|
|
async with httpx.AsyncClient(timeout=settings.gateway_timeout) as client:
|
|
try:
|
|
snapshot = await collectors.fetch_snapshot(client, target,
|
|
timeout=settings.gateway_timeout,
|
|
tls=agent_tls.get(target.name))
|
|
except collectors.UpstreamError as exc:
|
|
pytest.skip(f"{target.name} did not answer: {exc}")
|
|
payload = gateway.parse(snapshot, target.name)
|
|
if not payload["supported"]:
|
|
pytest.fail(f"{target.name} runs agent {payload.get('agent_version')} — the gateway page "
|
|
"needs 1.2.0's haproxy/nginx/sockets sections (AGENTS.md gap 11's rollout)")
|
|
assert payload["host"] == target.name
|
|
assert payload["sockets"]["established"] == len(snapshot["sockets"]["established"])
|
|
assert payload["listeners"], "no frontend on either config would leave the page empty"
|
|
assert payload["connections"], "the two configs declare no route at all"
|
|
for trunk in payload["listeners"]:
|
|
assert isinstance(trunk["connections"], int), "the socket table answered, so a count is known"
|
|
|
|
# the join, re-derived from the agent's own rows rather than from the payload's arithmetic
|
|
configured = {trunk["port"] for trunk in payload["listeners"]}
|
|
assert configured, "a gateway with no configured port has nothing to reconcile"
|
|
inbound = [row for row in snapshot["sockets"]["established"]
|
|
if ((row.get("local") or {}).get("port")) in configured]
|
|
assert sum(t["connections"] for t in payload["listeners"]) == len(inbound), (
|
|
"the trunk counts are the ESTABLISHED rows whose local port a config names — no more, "
|
|
"no less")
|
|
|
|
endpoints: dict[tuple[Any, Any], Any] = {}
|
|
joinable: set[str] = set()
|
|
for route in payload["connections"]:
|
|
assert route["in"]["port"] in configured, "a route hangs off no trunk the page draws"
|
|
out = route["out"]
|
|
placed = out["address"] is True or bool(out.get("resolved"))
|
|
assert (route["connections"] is None) is (not placed), (
|
|
f"{route['id']}: only an endpoint whose address is known — written, or resolved by the "
|
|
"gateway itself — can carry a count")
|
|
key = endpoint_identity(out)
|
|
if key is not None:
|
|
joinable.add(str(route["metric"]))
|
|
endpoints.setdefault(key, route["connections"])
|
|
else:
|
|
assert route["metric"] is None, "a route nobody can count publishes no series"
|
|
live_out = sum(value for value in endpoints.values() if isinstance(value, int))
|
|
assert sum(t["connections"] for t in payload["listeners"]) + live_out \
|
|
<= payload["sockets"]["established"], "the page may under-count the kernel, never exceed it"
|
|
|
|
# and nothing outside the two config files reached the payload at all — this app's own agent
|
|
# port is the case phase 30 deleted the reporting of
|
|
text = json.dumps(payload)
|
|
assert ":9120" not in text and "@9120" not in text, "the agent's own port leaked into the map"
|
|
stored = gateway.samples(payload, 0.0)
|
|
assert all(sample.source == gateway.SOURCE for sample in stored)
|
|
assert {sample.metric for sample in stored if "out_conns" in sample.metric} <= joinable, (
|
|
"only an endpoint whose address the join knows publishes an outbound gauge — written as an "
|
|
"address, or resolved by the gateway itself (phase 32); an unplaced name publishes nothing")
|
|
|
|
|
|
async def test_a_resolved_target_on_the_live_ingress_is_never_a_dash(
|
|
settings: Settings, agent_tls: dict[str, collectors.AgentTls]) -> None:
|
|
"""Phase 32 task 03 against the real ingress: `ok` in the resolver means a number on the page.
|
|
|
|
The operator's complaint was that one line out of thirty-three moved, and the acceptance line for
|
|
removing the dash is that it goes because the gateway published the address — not because a
|
|
column was hidden. So every route the resolver placed carries a **finite** `connections` (`0`
|
|
allowed: resolved and idle is a reading), a non-empty `out.resolved`, the agent's own `via`, the
|
|
timestamp of the lookup, and the gauge that goes with it. Skips when the ingress runs an agent
|
|
with no `resolver` section yet — which is every host until task 05 re-stages it.
|
|
"""
|
|
target = next((item for item in settings.host_targets if item.name == settings.gateway_host),
|
|
None)
|
|
if target is None:
|
|
pytest.skip(f"GATEWAY_HOST={settings.gateway_host!r} is not in HOST_TARGETS")
|
|
async with httpx.AsyncClient(timeout=settings.gateway_timeout) as client:
|
|
try:
|
|
snapshot = await collectors.fetch_snapshot(client, target,
|
|
timeout=settings.gateway_timeout,
|
|
tls=agent_tls.get(target.name))
|
|
except collectors.UpstreamError as exc:
|
|
pytest.skip(f"{target.name} did not answer: {exc}")
|
|
section = snapshot.get("resolver")
|
|
if not isinstance(section, dict):
|
|
pytest.skip(f"{target.name} publishes no resolver section — agent 1.3.0 is task 05's "
|
|
"re-stage; until then every name-written route is a dash by design")
|
|
payload = gateway.parse(snapshot, target.name)
|
|
if section.get("available") is not True:
|
|
pytest.skip(f"{target.name}'s resolver did not answer "
|
|
f"(available={section.get('available')!r}): nothing is placed, so nothing "
|
|
"is expected to be counted")
|
|
|
|
placed = [route for route in payload["connections"] if route["out"].get("resolved")]
|
|
assert placed, (f"{target.name} published {len(section.get('entries') or [])} resolver "
|
|
"entries and the join placed none of them")
|
|
for route in placed:
|
|
out = route["out"]
|
|
# The count lives on the **route** (`connections[]`), not on its `out` endpoint — the same
|
|
# shape `test_the_gateway_source_joins_the_live_ingress` reads. This path only ever ran once
|
|
# agent 1.3.0 was installed (before that the whole test skipped for want of a `resolver`
|
|
# section), which is how a wrong key survived task 03: a KeyError is not a dash.
|
|
assert isinstance(route["connections"], int) and not isinstance(route["connections"], bool), (
|
|
f"{route['id']}: resolved to {out['resolved']} and the line is still a dash")
|
|
assert out["via"] == section.get("via") == "libc", (
|
|
"the vantage is the agent's string, copied, never one this module invented")
|
|
assert isinstance(out["resolved_ts"], int | float), (
|
|
f"{route['id']}: a count obtained through a lookup must say when the lookup happened")
|
|
assert route["metric"] is not None, "a counted route always has its gauge"
|
|
stored = {sample.metric for sample in gateway.samples(payload, 0.0)}
|
|
assert {route["metric"] for route in placed} <= stored, (
|
|
"a resolved route's count must reach the store, or its chart stays empty forever")
|
|
|
|
|
|
@pytest.mark.parametrize("engine", ["gufo", "vllm"])
|
|
async def test_engine_still_publishes_the_series_we_parse(engine: str,
|
|
settings: Settings) -> None:
|
|
"""Metric names are verified against the deployed engine, never guessed.
|
|
|
|
The assertion is that at least one cumulative counter per family *resolves* — that is
|
|
what the Δ rates, the throughput cells, and the history rows all hang off. If an
|
|
upgrade renames it, this fails here instead of printing 0 tok/s on the dashboard.
|
|
"""
|
|
base_url = settings.gufo_url if engine == "gufo" else settings.vllm_url
|
|
api_key = settings.gufo_api_key if engine == "gufo" else settings.vllm_api_key
|
|
if not base_url:
|
|
pytest.skip(f"{engine.upper()}_BASE_URL is unset — that engine is not deployed here")
|
|
|
|
counters = {
|
|
"gufo": ("tokens_generated_total", "tokens_prompt_total"),
|
|
"vllm": ("generation_tokens_total", "prompt_tokens_total"),
|
|
}[engine]
|
|
# The span the LLM jobs really get: phase 23 spans the LLM estimator by the poll
|
|
# interval, not by the host's `RATE_WINDOW_SECONDS`.
|
|
rates = RateEstimator(span=settings.llm_interval)
|
|
async with httpx.AsyncClient(timeout=settings.llm_timeout) as client:
|
|
result = await collectors.fetch_engine(client, engine, base_url, api_key, rates,
|
|
with_slots=engine == "gufo")
|
|
metrics = result.payload["metrics"]
|
|
resolved = [key for key in counters if metrics[key]["source"] is not None]
|
|
assert resolved, (
|
|
f"{engine}: none of {counters} resolved at {base_url} — upstream renamed its "
|
|
f"counters, update {engine.upper()}_SERIES in sources/llm.py"
|
|
)
|
|
sessions = result.payload["sessions"]
|
|
assert sessions["active"] is None or sessions["active"] >= 0
|
|
assert result.samples, f"{engine}: nothing to store, so the charts would stay empty"
|
|
|
|
|
|
async def test_gufo_slots_answer_the_pinned_shape(settings: Settings) -> None:
|
|
"""The `/slots` rows behind the phase-07 slot grid, asserted against the running gufo.
|
|
|
|
Hermetic twin: `tests/test_llm_slots_contract.py`, driven from
|
|
`tests/fixtures/slots_gufo.json`. This is the half that catches gufo renaming a slot
|
|
field — the old table rendered dashes because nobody had ever pinned the shape, and the
|
|
assertion here goes through the production path (`collectors.fetch_engine`, which owns
|
|
the `/slots` → `/v1/slots` fallback), not a hand-rolled request.
|
|
"""
|
|
if not settings.gufo_url:
|
|
pytest.skip("GUFO_BASE_URL is unset — no gufo to ask for /slots")
|
|
# The span the LLM jobs really get: phase 23 spans the LLM estimator by the poll
|
|
# interval, not by the host's `RATE_WINDOW_SECONDS`.
|
|
rates = RateEstimator(span=settings.llm_interval)
|
|
async with httpx.AsyncClient(timeout=settings.llm_timeout) as client:
|
|
result = await collectors.fetch_engine(client, "gufo", settings.gufo_url,
|
|
settings.gufo_api_key, rates, with_slots=True)
|
|
rows = result.payload["sessions"]["list"]
|
|
assert rows, (
|
|
"gufo answered /slots with no rows — even an idle engine reports one row per "
|
|
"parallel slot (the idle shape pinned in tests/fixtures/slots_gufo.json); "
|
|
f"detail: {result.payload['sessions']['detail']!r}"
|
|
)
|
|
assert_slot_shape(rows)
|
|
|
|
|
|
def require_upsd(settings: Settings) -> None:
|
|
"""Skip, not fail, when there is no real upsd to talk to."""
|
|
if not settings.nut_enabled:
|
|
pytest.skip("NUT_HOST is unset — no upsd to talk to")
|
|
if os.environ.get("NUT_SIMULATE", "").lower() in ("1", "true", "yes", "on"):
|
|
pytest.skip("NUT_SIMULATE=1 — point NUT_HOST at the real upsd for this run")
|
|
|
|
|
|
async def test_upsd_reports_both_upses_with_sane_units(settings: Settings) -> None:
|
|
"""Real batteries: plausible volts/percent, and the health checks are never silent."""
|
|
require_upsd(settings)
|
|
result = await collectors.NutMonitor(settings).collect()
|
|
upses = result.payload["ups"]
|
|
assert upses, "upsd answered but exported no UPS (check ups.conf driver instances)"
|
|
for card in upses:
|
|
name = card["name"]
|
|
assert card["status"], f"{name}: ups.status is empty"
|
|
assert card["health"]["checks"], f"{name}: assess() produced no health checks"
|
|
numbers = card["numbers"]
|
|
charge = numbers.get("charge_pct")
|
|
assert charge is None or 0.0 <= charge <= 100.0, f"{name}: charge_pct={charge}"
|
|
# kelvin-as-celsius is the classic unit bug here; 273 would sail past a bare >0
|
|
temp = numbers.get("battery_temperature")
|
|
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.
|
|
|
|
Deliberately does not POST anything: `POST /api/ups/command` behaviour is covered by
|
|
tests/test_api.py against a fake client, because running a deep battery test on live
|
|
hardware to prove a button works is exactly what this app is careful about.
|
|
"""
|
|
require_upsd(settings)
|
|
result = await collectors.NutMonitor(settings).collect()
|
|
seen = False
|
|
for card in result.payload["ups"]:
|
|
for entry in card["commands"]:
|
|
seen = True
|
|
assert isinstance(entry["enabled"], bool), f"{card['name']}: {entry} has no gate state"
|
|
if not entry["enabled"]:
|
|
assert entry.get("reason"), (
|
|
f"{card['name']}: {entry['command']} disabled with no reason — the UI "
|
|
f"would show a dead button with no explanation"
|
|
)
|
|
if entry.get("risk") == nut.RISK_DANGEROUS:
|
|
assert entry.get("confirm_text"), (
|
|
f"{card['name']}: {entry['command']} can cut power without a typed confirmation"
|
|
)
|
|
if not seen:
|
|
pytest.skip("the UPS advertises no commands we model (see ACTION_CATALOG)")
|
|
|
|
|
|
def test_target_config_is_clean_and_leak_free(settings: Settings) -> None:
|
|
"""Every configured target parses, and none of them leak a userinfo token."""
|
|
names = [target.name for target in settings.host_targets]
|
|
assert len(names) == len(set(names)), f"duplicate target names: {names}"
|
|
for target in settings.host_targets:
|
|
assert target.url.startswith(("http://", "https://")), f"{target.name}: {target.url}"
|
|
assert "@" not in target.display_url, f"{target.name}: display_url leaks credentials"
|
|
for problem in settings.host_target_problems:
|
|
assert isinstance(problem, str) and problem, f"empty problem: {problem!r}"
|
|
|
|
|
|
#: The one write target of the opt-in round trip (phase 33 task 07): a benchmark service
|
|
#: with no human dependency. Never the auth, git, alerts or file routes — task 07's
|
|
#: precondition names them, and a test that would touch one of them is a test that does not
|
|
#: run.
|
|
SWITCH_DRILL_ROUTE = "haproxy:5201->web.reeselink.com:5201"
|
|
|
|
|
|
async def test_the_switch_round_trip_on_one_named_route(
|
|
request: pytest.FixtureRequest, settings: Settings,
|
|
agent_tls: dict[str, collectors.AgentTls]) -> None:
|
|
"""Phase 33's one live write: the named route off and back on, through the app's own
|
|
`POST /route` dial.
|
|
|
|
Gated twice so a default live run stays read-only: the file's `live` marker (skip
|
|
without `--live`) **and** the separate `--allow-live-writes` flag, whose skip reason
|
|
names the route it would touch. When it does run — the operator's drill, task 07 step
|
|
5 — it dials `collectors.post_route` (the app's second and last upstream write, through
|
|
the stored mTLS authority, no hand-rolled client) and proves the round trip on the
|
|
production read (`fetch_snapshot` → `gateway.parse`): the helper's own words for the
|
|
write, the row's `enabled` flipping both ways, the agent's own section carrying the
|
|
marked line in between, and the haproxy section the agent serves byte-identical between
|
|
the pre-drill and post-drill reads. The restore lives in a `finally`: even a failed
|
|
proof turns the route back on before the test ends, and a re-enable that itself fails
|
|
propagates the helper's own word — a marked line left behind must be the operator's
|
|
first line, never a silent one.
|
|
"""
|
|
if not bool(request.config.getoption("--allow-live-writes")):
|
|
pytest.skip(f"opt-in live write: skip without --allow-live-writes (this test would "
|
|
f"switch {SWITCH_DRILL_ROUTE} off and back on — the default live suite "
|
|
"stays read-only")
|
|
target = next((item for item in settings.host_targets if item.name == settings.gateway_host),
|
|
None)
|
|
if target is None:
|
|
pytest.skip(f"GATEWAY_HOST={settings.gateway_host!r} is not in HOST_TARGETS")
|
|
|
|
async def read_ingress(client: httpx.AsyncClient) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
snapshot = await collectors.fetch_snapshot(client, target,
|
|
timeout=settings.gateway_timeout,
|
|
tls=agent_tls.get(target.name))
|
|
return snapshot, gateway.parse(snapshot, target.name)
|
|
|
|
async def post(body: dict[str, Any]) -> dict[str, Any]:
|
|
# One client per call: one write, one connection — the same discipline the agent
|
|
# keeps for its one helper dial (a retry of a write is a second reload).
|
|
async with httpx.AsyncClient(timeout=settings.gateway_timeout) as client:
|
|
return await collectors.post_route(client, target, body,
|
|
timeout=settings.gateway_timeout,
|
|
tls=agent_tls.get(target.name))
|
|
|
|
disabled = False
|
|
try:
|
|
async with httpx.AsyncClient(timeout=settings.gateway_timeout) as client:
|
|
try:
|
|
before, payload = await read_ingress(client)
|
|
except collectors.UpstreamError as exc:
|
|
pytest.skip(f"{target.name} did not answer: {exc}")
|
|
# Precondition 1: the agent on this host is the one with the write path.
|
|
version = str(before.get("agent_version") or "")
|
|
if not version.startswith("1.4."):
|
|
pytest.skip(f"{target.name} runs agent {version or 'unknown'} — POST /route is "
|
|
"1.4.0's; re-stage with deploy/install-agent.sh --allow-writes "
|
|
"--with-routectl")
|
|
# Precondition 2: a clean pre-state — the route is on and nothing else is off;
|
|
# a half-finished earlier attempt is the state task 07 step 4 refuses to drill.
|
|
route = next((row for row in payload["connections"]
|
|
if row["id"] == SWITCH_DRILL_ROUTE), None)
|
|
if route is None:
|
|
pytest.skip(f"{SWITCH_DRILL_ROUTE} is not on the ingress the agent published — "
|
|
"the drill target moved; update this test's target and re-verify")
|
|
already_off = [row["id"] for row in payload["connections"]
|
|
if row["enabled"] is False]
|
|
assert not already_off, (
|
|
f"a route is already disabled before the drill {already_off} — a previous "
|
|
"attempt left a marked line; restore it before re-running (task 07 step 6)")
|
|
# 1. dry_run: the diff, and no write (task 07 step 5.1).
|
|
preview = await post({"id": SWITCH_DRILL_ROUTE, "action": "disable",
|
|
"dry_run": True})
|
|
assert preview.get("ok") is True and preview.get("dry_run") is True, preview
|
|
assert preview.get("applied") is False and preview.get("changed") is True, preview
|
|
assert any("#off" in str(line) for line in preview.get("diff") or []), (
|
|
f"the dry-run diff must show the marker landing: {preview.get('diff')!r}")
|
|
# 2. apply: the helper's own word for the write (step 5.2). A checker that refuses
|
|
# (e.g. no `nginx -t` on the host's PATH) answers `applied: false, restored: true`
|
|
# and this test fails on the helper's word instead of printing a false green.
|
|
applied = await post({"id": SWITCH_DRILL_ROUTE, "action": "disable"})
|
|
assert applied.get("ok") is True and applied.get("applied") is True, (
|
|
f"the helper refused the write instead of applying it: {applied!r}")
|
|
assert applied.get("validated") is True and applied.get("reloaded") is True, applied
|
|
disabled = True
|
|
# 3. the off state, proven on the production read (step 5.3/5.4): the row and the
|
|
# agent's own section both report the marker.
|
|
async with httpx.AsyncClient(timeout=settings.gateway_timeout) as client:
|
|
off_snapshot, off_payload = await read_ingress(client)
|
|
off_route = next(row for row in off_payload["connections"]
|
|
if row["id"] == SWITCH_DRILL_ROUTE)
|
|
assert off_route["enabled"] is False, "the row must report disabled within one read"
|
|
marked = [server for backend in (off_snapshot.get("haproxy") or {}).get("backends", ())
|
|
for server in backend.get("servers", ()) if server.get("disabled") is True]
|
|
assert marked, "the agent's haproxy section shows no marked server line"
|
|
finally:
|
|
# The restore lives here, not after the assertions: a failed proof still leaves the
|
|
# route on, and a failed re-enable propagates the helper's own word.
|
|
if disabled:
|
|
await post({"id": SWITCH_DRILL_ROUTE, "action": "enable"})
|
|
# 4. the round trip closed: the row is on, nothing is off, and the haproxy section the
|
|
# agent serves is byte-identical to the pre-drill read (step 5.5/5.6).
|
|
async with httpx.AsyncClient(timeout=settings.gateway_timeout) as client:
|
|
after, after_payload = await read_ingress(client)
|
|
after_route = next(row for row in after_payload["connections"]
|
|
if row["id"] == SWITCH_DRILL_ROUTE)
|
|
assert after_route["enabled"] is True, (
|
|
f"the route is still off after the re-enable — the drill left a marked line: "
|
|
f"{after_route!r}")
|
|
left_off = [row["id"] for row in after_payload["connections"]
|
|
if row["enabled"] is False]
|
|
assert not left_off, f"marked lines left behind after the round trip: {left_off}"
|
|
assert json.dumps(after.get("haproxy")) == json.dumps(before.get("haproxy")), (
|
|
"the haproxy section is not byte-identical after the round trip")
|