feat(plugins): Traefik problems first, UniFi/SNMP/fleet on shared shapes, audit names its entities, templates on tokens (#5126)
CI / lint (push) Successful in 2s
CI / integration (push) Successful in 1m5s
CI / unit (push) Successful in 1m19s
CI / publish (push) Successful in 57s

Traefik: routers are graded from their latest scrape (router_state: crit on
any real 5xx share, warn on p95 past the latency critical cutoff or a high
4xx share). Failing and slow routers lead as cards that say why; healthy
ones fold into one table that stays open or shut across polls. Proxy
counts are stat tiles; certificates are graded by Settings -> Thresholds
(tls_days).

UniFi: WAN health is a stat strip (latency graded by the thresholds),
devices carry status dots, event categories and firewall actions are
badges, the alarm count reuses .count-badge (renamed from .nav-count).
SNMP: each device wears a status badge and its readings are stat tiles.

Agent fleet: the table comes first, with a Status column (Reporting /
Silent / Never reported, by the host page's stale_after_seconds), silent
agents first, hosts linked, last seen as |ago.

Audit: entries name their entity (one query per type, steward/audit/
service.py) and link it; something deleted since keeps the name its entry
recorded and says "(deleted)".

Tokens: charts read colours through chart-colors.js (stewardColor resolves
a token to the rgb the browser computed; Chart.js defaults take the page's
text and border colours), so host, history, Traefik and SNMP charts drop
their hex. Template hex colours reach 0 (HEX_CEILING = 0), the last rgba
pills go, and heavy weights (600/700/bold, 60 of them) become 500, the
house pair being 400/500; a guard keeps templates there.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-06 18:25:09 -04:00
co-authored by Claude Opus 5.5
parent c2a79c8bb0
commit 414c942cbf
60 changed files with 686 additions and 391 deletions
+2 -2
View File
@@ -121,7 +121,7 @@ async def test_the_sidebar_counts_alerts_needing_attention(app, monkeypatch):
async with app.test_request_context("/alerts/"):
session["user_id"], session["user_role"] = "u1", "viewer"
page = await render_template("alerts/list.html", active=[], rules=[], operators=[])
assert '<span class="nav-count" title="3 alerts need attention">3</span>' in page
assert '<span class="count-badge" title="3 alerts need attention">3</span>' in page
async def test_no_count_for_a_fragment_or_a_logged_out_page(app, monkeypatch):
@@ -131,4 +131,4 @@ async def test_no_count_for_a_fragment_or_a_logged_out_page(app, monkeypatch):
fragment = await render_template("alerts/list.html", active=[], rules=[], operators=[])
async with app.test_request_context("/alerts/"):
logged_out = await render_template("alerts/list.html", active=[], rules=[], operators=[])
assert "nav-count" not in fragment and "nav-count" not in logged_out
assert "count-badge" not in fragment and "count-badge" not in logged_out
+45
View File
@@ -0,0 +1,45 @@
"""Audit entries name the thing they are about, linked to its page, and keep
the recorded name of something deleted since (#5126)."""
from datetime import datetime, timezone
from types import SimpleNamespace as NS
from quart import render_template
from steward.audit.service import entity_labels
class _Empty:
"""A session whose every query finds nothing: every entity was deleted."""
async def execute(self, _stmt):
return NS(all=lambda: [], scalars=lambda: NS(all=lambda: []))
def _ev(etype, eid, action="x.y"):
return NS(entity_type=etype, entity_id=eid, action=action, username="admin",
timestamp=datetime(2026, 10, 6, 12, 0, tzinfo=timezone.utc))
async def test_a_deleted_entity_keeps_the_name_its_entry_recorded():
items = [{"event": _ev("host", "h-123456789"), "detail": {"name": "web-1"}},
{"event": _ev("monitor", "m-abcdef012"), "detail": {}},
{"event": _ev("plugin", "docker"), "detail": {}},
{"event": _ev(None, None), "detail": {}}]
labels = await entity_labels(_Empty(), items)
assert labels[("host", "h-123456789")] == {"label": "web-1", "url": None, "gone": True}
assert labels[("monitor", "m-abcdef012")] == {"label": "m-abcdef", "url": None, "gone": True}
assert labels[("plugin", "docker")] == {"label": "docker", "url": "/settings/plugins/", "gone": False}
assert len(labels) == 3
async def test_the_page_names_and_links_the_entity(app):
events = [
{"event": _ev("host", "h1", "host.renamed"), "detail": {"old": "a", "new": "b"},
"entity": {"label": "web-1", "url": "/hosts/h1", "gone": False}},
{"event": _ev("monitor", "m9", "monitor.deleted"), "detail": {},
"entity": {"label": "api check", "url": None, "gone": True}},
]
async with app.test_request_context("/audit/"):
page = await render_template("audit/list.html", events=events, limit=200, action_filter="")
assert '<a href="/hosts/h1">web-1</a>' in page
assert "api check" in page and "(deleted)" in page
assert "h1</td>" not in page # no bare id where a name is known
@@ -0,0 +1,39 @@
"""The agent fleet page opens on the fleet itself: a status column, silent
agents first (#5126)."""
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace as NS
from plugins.host_agent.routes import fleet_rows, fleet_state
from tests._templates import render_plugin_template
def _reg(host_id, seen_ago_s):
now = datetime.now(timezone.utc)
return NS(host_id=host_id, last_seen_at=None if seen_ago_s is None else now - timedelta(seconds=seen_ago_s),
host_ip="10.0.0.5", agent_version="1.4.0", distro="debian")
def test_fleet_state():
now = datetime.now(timezone.utc)
assert fleet_state(_reg("a", 30), now, 180) == "up"
assert fleet_state(_reg("a", 600), now, 180) == "stale"
assert fleet_state(_reg("a", None), now, 180) == "pending"
async def test_silent_agents_come_first_and_the_table_leads_the_page(app):
now = datetime.now(timezone.utc)
hosts = {i: NS(id=i, name=n) for i, n in [("h1", "web"), ("h2", "nas"), ("h3", "attic")]}
rows = fleet_rows([_reg("h1", 30), _reg("h2", 900), _reg("h3", None)], hosts, now, 180)
assert [r["name"] for r in rows] == ["nas", "attic", "web"]
async with app.test_request_context("/plugins/host_agent/fleet/"):
page = await render_plugin_template(app, "host_agent", "settings_list.html", registrations=rows,
new_token=None, install_url=None, ansible_available=False,
deploy_targets=[], deploy_groups=[], managed_key_set=False)
assert page.index("<th>Status</th>") < page.index('action="/plugins/host_agent/fleet/add-host"')
assert '<span class="badge badge-warn">Silent</span>' in page
assert '<span class="badge badge-warn">Never reported</span>' in page
assert '<span class="badge badge-ok">Reporting</span>' in page
assert page.index('href="/hosts/h2">nas</a>') < page.index('href="/hosts/h1">web</a>')
assert 'data-confirm="Delete the agent registration for nas?' in page
+22
View File
@@ -0,0 +1,22 @@
"""The SNMP list says each device's state with the shared badge and shows its
readings as stat tiles (#5126)."""
from tests._templates import render_plugin_template
from tests.test_style_tokens import hex_colours
async def test_devices_wear_a_status_badge_and_readings_are_tiles(app):
devices = [
{"name": "ups", "host": "10.0.0.9", "oids": [{"oid": "1.3.6", "label": "battery_charge_pct"},
{"oid": "1.3.7", "label": "in_octets"}],
"readings": {"battery_charge_pct": 97.0, "in_octets": 2_500_000.0}},
{"name": "printer", "host": "10.0.0.10", "oids": [], "readings": {}},
]
async with app.test_request_context("/plugins/snmp/"):
page = await render_plugin_template(app, "snmp", "snmp/index.html", devices=devices, poll_interval=60)
assert '<span class="badge badge-ok">Reachable</span>' in page
assert '<span class="badge badge-warn">No data yet</span>' in page
assert '<span class="stat-tile-value">97</span>' in page and '<span class="stat-tile-value">2.50M</span>' in page
assert 'href="/plugins/snmp/device/ups"' in page
main = page[page.index("<main"):]
assert "font-weight:600" not in main and not hex_colours(main), hex_colours(main)
View File
@@ -0,0 +1,65 @@
"""The Traefik page puts failing and slow routers first and folds the healthy
ones into one table; certificates are graded by Settings → Thresholds (#5126)."""
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace as NS
import pytest
from plugins.traefik.routes import router_state, split_routers
from steward.core.settings import to_thresholds_cfg
from tests._templates import render_plugin_template
from tests.test_style_tokens import hex_colours
TH = to_thresholds_cfg({}) # latency crit 200 ms
def _latest(req=1.0, p95=20.0, e4=0.0, e5=0.0):
return NS(request_rate=req, response_bytes_rate=1000.0, latency_p95_ms=p95, latency_p99_ms=p95 * 1.5,
error_rate_4xx_pct=e4, error_rate_5xx_pct=e5,
scraped_at=datetime.now(timezone.utc) - timedelta(seconds=30))
@pytest.mark.parametrize("latest, grade", [
(_latest(), "ok"),
(_latest(e5=2.5), "crit"),
(_latest(p95=450.0), "warn"),
(_latest(e4=5.0), "warn"),
(_latest(p95=120.0, e4=0.5), "ok"), # slow-ish but under the critical cutoff
])
def test_router_state(latest, grade):
assert router_state(latest, TH)[0] == grade
def _router(name, latest):
state, why = router_state(latest, TH)
return {"name": name, "latest": latest, "state": state, "why": why,
"sparkline_req": "", "sparkline_p95": "", "sparkline_5xx": ""}
def test_failing_routers_come_before_slow_ones_and_healthy_ones_split_off():
problems, healthy = split_routers([
_router("b-ok", _latest()), _router("slow", _latest(p95=450.0)),
_router("broken", _latest(e5=3.0)), _router("a-ok", _latest())])
assert [r["name"] for r in problems] == ["broken", "slow"]
assert [r["name"] for r in healthy] == ["a-ok", "b-ok"]
async def test_the_page_leads_with_problems_and_folds_the_healthy(app):
problems, healthy = split_routers([
_router("api", _latest(e5=3.0)), _router("web", _latest()), _router("docs", _latest())])
now = datetime.now(timezone.utc)
certs = [{"serial": "1", "cn": "soon.example", "sans": [], "not_after": now + timedelta(days=2),
"days_remaining": 2.0},
{"serial": "2", "cn": "fine.example", "sans": [], "not_after": now + timedelta(days=60),
"days_remaining": 60.0}]
async with app.test_request_context("/plugins/traefik/rows"):
page = await render_plugin_template(app, "traefik", "traefik/rows.html", problems=problems,
healthy=healthy, global_stat=None, certs=certs, range_key="24h",
error_4xx=1.0, error_5xx=0.1)
assert page.index(">api</h3>") < page.index('id="traefik-healthy"')
assert "3.0% of responses are 5xx" in page
assert ">2 healthy routers</summary>" in page
assert page.index(">docs</td>") < page.index(">web</td>")
assert 'class="tone-crit">2 days</td>' in page and 'class="tone-ok">60 days</td>' in page
assert "font-weight:600" not in page and not hex_colours(page), hex_colours(page)
+34
View File
@@ -0,0 +1,34 @@
"""UniFi's overview reads in the shared shapes: WAN health as stat tiles graded
by Settings → Thresholds, devices with status dots, categories as badges (#5126)."""
from datetime import datetime, timezone
from types import SimpleNamespace as NS
from tests._templates import render_plugin_template
from tests.test_style_tokens import hex_colours
async def test_wan_health_and_devices_use_the_shared_shapes(app):
wan = NS(is_up=False, wan_ip="203.0.113.7", latency_ms=250.0, rx_bytes_rate=2_000_000.0,
tx_bytes_rate=500_000.0, uptime_seconds=7200)
devices = [NS(name="switch", mac="aa", model="USW", device_type="usw", ip="10.0.0.2", state=1,
uptime_seconds=3600),
NS(name="ap-attic", mac="bb", model="U6", device_type="uap", ip="10.0.0.3", state=0,
uptime_seconds=None)]
async with app.test_request_context("/plugins/unifi/rows"):
page = await render_plugin_template(app, "unifi", "unifi/rows.html", latest_wan=wan, devices=devices,
sparkline_latency="", sparkline_rx="", sparkline_tx="")
assert '<span class="stat-tile-label">WAN</span>' in page and "stat-crit" in page
assert '<span class="stat-tile-value">250 ms</span>' in page
assert 'class="dot dot-ok"' in page and 'class="dot dot-crit"' in page
assert "font-weight:600" not in page and "font-weight:700" not in page
assert not hex_colours(page), hex_colours(page)
async def test_event_categories_are_badges(app):
events = [NS(event_time=datetime(2026, 10, 6, 12, 0, tzinfo=timezone.utc), category="ids",
event_key="EVT_IPS_Alert", message="blocked", subsystem="ids")]
async with app.test_request_context("/plugins/unifi/events"):
page = await render_plugin_template(app, "unifi", "unifi/events.html", events=events)
assert '<span class="badge badge-crit">ids</span>' in page
assert "rgba(" not in page
+13 -1
View File
@@ -22,7 +22,7 @@ _USED = re.compile(r"var\(\s*(--[\w-]+)")
# string -- but not an HTML attribute's value (href="#abc" is a link to an id).
_HEX = re.compile(r"""(?:[:(,]\s*|(?<![\w-]=)['"])#(?:[0-9a-fA-F]{8}|[0-9a-fA-F]{6}|[0-9a-fA-F]{3})(?![0-9a-zA-Z_-])""")
HEX_CEILING = 47 # 75 at #5092, 58 at #5095, 56 at #5121, 54 at #5122, 47 with Ansible runs on tokens (#5125); only lower it
HEX_CEILING = 0 # 75 at #5092, 58 at #5095, 56 at #5121, 54 at #5122, 47 at #5125, 0 with the charts on tokens (#5126)
def _sheets():
@@ -45,11 +45,17 @@ def hex_colours(text: str) -> list[str]:
return _HEX.findall(_COMMENTS.sub("", text))
def heavy_weights(text: str) -> list[str]:
"""Weights past the house pair (400, 500), outside comments."""
return re.findall(r"font-weight\s*:\s*(?:[6-9]00|bold)", _COMMENTS.sub("", text))
def test_the_guards_can_fail():
assert undefined_vars([":root { --a: 1; } .x { color: var(--a); background: var(--b); }"]) == {"--b"}
page = '<div style="color:#fff;background: rgb(0 0 0)" hx-target="#add-form"></div><script>c = "#4aa86a"</script>'
assert len(hex_colours(page)) == 2
assert hex_colours("{# color:#fff #}<a href=\"#abc\">") == []
assert heavy_weights('<b style="font-weight: 600">{# font-weight:700 #}</b>') == ["font-weight: 600"]
def test_every_css_variable_used_is_defined():
@@ -63,6 +69,12 @@ def test_the_base_stylesheet_states_no_colour_or_heavy_weight_of_its_own():
assert not re.findall(r"font-weight\s*:\s*(?:[6-9]00|bold)", css)
def test_templates_use_the_two_house_weights():
"""The house style has two weights, 400 and 500 (design system 1)."""
heavy = {p.name: len(found) for p in _templates() if (found := heavy_weights(p.read_text()))}
assert not heavy, heavy
def test_hex_colours_in_templates_only_go_down():
count = sum(len(hex_colours(p.read_text())) for p in _templates())
assert count <= HEX_CEILING, (