Wire the live-test gate and pin the toolchain to 3.14
README advertised `uv run pytest -m live --live`, but there was no --live option and no `live` marker registered — the command errored out, so the one workflow that catches an upstream renaming a series did not exist. - conftest adds --live and skips `live`-marked tests without it, so the default run stays hermetic (258 passed, 6 skipped, no network). - tests/test_live_fleet.py drives the real code paths (collectors.*, NutMonitor.collect) against the fleet, both engines, and upsd. Each skips when its source is unset, and none of them may write — a battery test on live hardware is not a unit test. The host test polls twice because a rate has no second point on the first scrape. - The agent's real-/proc test stays in the default run and is explicitly not `live`: it touches only the box running pytest. - dev-tunnels.sh grows DEV_FORWARDS for services already listening remotely, which is how upsd gets reachable without opening 3493; pidfiles are now host+port so one host can carry two tunnels. - requires-python, ruff and pyright were still on 3.13 while the venv and Containerfile are 3.14, so the lock resolved for a stack nobody deploys. LLM-Generated: true
This commit is contained in:
@@ -0,0 +1,167 @@
|
||||
"""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:
|
||||
|
||||
./scripts/dev-tunnels.sh start # agents + upsd forwarded to localhost
|
||||
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.** 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.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from dashboards import collectors
|
||||
from dashboards.config import Settings, load
|
||||
from dashboards.rates import RateEstimator
|
||||
from dashboards.sources import nut
|
||||
from dashboards.sources.host import HostCollector
|
||||
|
||||
pytestmark = pytest.mark.live
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def settings() -> Settings:
|
||||
"""Live env, not the import-time SETTINGS singleton."""
|
||||
return load()
|
||||
|
||||
|
||||
@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) -> 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 — start ./scripts/dev-tunnels.sh")
|
||||
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:
|
||||
# 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)
|
||||
await asyncio.sleep(1.2)
|
||||
result = await collectors.fetch_host_snapshot(client, target, host_collector)
|
||||
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')}"
|
||||
)
|
||||
|
||||
|
||||
@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]
|
||||
rates = RateEstimator(span=settings.rate_window)
|
||||
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"
|
||||
|
||||
|
||||
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_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}"
|
||||
Reference in New Issue
Block a user