bvandeusenandClaude Opus 5.5
715ddfd72f
Under an hour |ago says "12m 3s ago", so the test looks for the minutes alone; the reading cell drops an empty class. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
105 lines
5.2 KiB
Python
105 lines
5.2 KiB
Python
"""The dashboard Alerts widget (#5510): a verdict, what is firing or silent
|
|
with its reading and since when, then what someone has acknowledged, a line
|
|
when alert email is failing, and the quiet rules counted."""
|
|
import re
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
|
|
from quart import render_template, session
|
|
|
|
from steward.alerts.service import widget_view
|
|
from steward.core.widgets import WIDGET_REGISTRY
|
|
from steward.models.alerts import AlertOperator
|
|
from tests.test_style_tokens import hex_colours
|
|
|
|
NOW = datetime.now(timezone.utc)
|
|
QUIET_EMAIL = {"failed": 0, "retrying": 0, "said": ""}
|
|
|
|
|
|
def _rule(rid, name, resource="web-1", metric="cpu_pct", op=">", threshold=90.0,
|
|
parent_id=None, enabled=True):
|
|
return SimpleNamespace(id=rid, name=name, resource_name=resource, metric_name=metric,
|
|
operator=AlertOperator(op), threshold=threshold,
|
|
parent_id=parent_id, enabled=enabled)
|
|
|
|
|
|
def _row(rule, label, value=None, since=None, acked_by=None, acked_at=None):
|
|
return {"rule": rule, "label": label, "value": value, "since": since, "acked_by": acked_by,
|
|
"state": SimpleNamespace(acknowledged_at=acked_at)}
|
|
|
|
|
|
async def _render(app, view, email=QUIET_EMAIL, role="admin"):
|
|
async with app.test_request_context("/alerts/widget"):
|
|
if role:
|
|
session["user_id"] = "u1"
|
|
session["user_role"] = role
|
|
return await render_template("alerts/widget.html", **view, email=email)
|
|
|
|
|
|
def test_the_widget_is_registered_and_points_at_alerts():
|
|
w = WIDGET_REGISTRY["alerts"]
|
|
assert w["hx_url"] == "/alerts/widget" and w["detail_url"] == "/alerts/" and w["plugin"] is None
|
|
|
|
|
|
def test_a_covering_rule_counts_once_and_is_quiet_only_when_all_its_hosts_are():
|
|
cover = _rule("c", "disk on every host", resource="*")
|
|
kid_a = _rule("c-a", "disk on every host", resource="nas", parent_id="c")
|
|
kid_b = _rule("c-b", "disk on every host", resource="web-1", parent_id="c")
|
|
lone = _rule("l", "web up", metric="is_up", op="<", threshold=0.5)
|
|
off = _rule("o", "old rule", enabled=False)
|
|
view = widget_view([_row(kid_a, "firing", 91.0, NOW)], [cover, kid_a, kid_b, lone, off])
|
|
assert view["total"] == 2 and view["quiet"] == 1 # "web up"; the covering rule is busy
|
|
assert view["rows"][0]["condition"] == "CPU > 90%"
|
|
assert widget_view([], [cover, kid_a, kid_b, lone, off])["quiet"] == 2
|
|
|
|
|
|
async def test_firing_and_silent_come_first_then_who_acknowledged(app):
|
|
hot = _rule("r1", "web-1 CPU high")
|
|
gone = _rule("r2", "nas agent", resource="nas")
|
|
taken = _rule("r3", "db disk", resource="db-1:/var", metric="disk_used_pct", op=">=", threshold=85.0)
|
|
rows = [_row(hot, "firing", 94.0, NOW - timedelta(minutes=12)),
|
|
_row(gone, "no_data", None, NOW - timedelta(hours=2)),
|
|
_row(taken, "acknowledged", 88.0, NOW - timedelta(hours=3), acked_by="bryan",
|
|
acked_at=NOW - timedelta(minutes=40))]
|
|
page = await _render(app, widget_view(rows, [hot, gone, taken, _rule("r4", "quiet one")]))
|
|
assert re.search(r'w-verdict tone-crit">\s*1 firing, 1 silent, 1 acknowledged\.', page)
|
|
assert page.index("web-1 CPU high") < page.index("nas agent") < page.index("db disk")
|
|
assert 'href="/alerts/history?rule=r1"' in page
|
|
assert "web-1 · fires on CPU > 90%" in page and ">94%<" in page and ">12m " in page
|
|
assert "nas · no reading" in page
|
|
assert "db-1:/var · acknowledged by bryan <time" in page and ">40m " in page
|
|
assert "+1 rule quiet" in page and 'href="/alerts/"' in page
|
|
assert len(re.findall(r"<th[ >]", page)) == 3
|
|
assert not hex_colours(page), hex_colours(page)
|
|
|
|
|
|
async def test_all_quiet_is_the_verdict_alone(app):
|
|
rules = [_rule("r1", "a"), _rule("r2", "b")]
|
|
page = await _render(app, widget_view([], rules))
|
|
assert re.search(r'w-verdict tone-ok">\s*Nothing firing: all 2 rules are quiet\.', page)
|
|
assert "<table" not in page and "rule quiet" not in page and "rules quiet" not in page
|
|
page = await _render(app, widget_view([], rules[:1]))
|
|
assert "Nothing firing: the rule is quiet." in page
|
|
|
|
|
|
async def test_no_rules_offers_to_add_one(app):
|
|
page = await _render(app, widget_view([], []), role="operator")
|
|
assert "No alert rules yet." in page and 'href="/alerts/rules/new"' in page
|
|
page = await _render(app, widget_view([], []), role="viewer")
|
|
assert "No alert rules yet." in page and "/alerts/rules/new" not in page
|
|
|
|
|
|
async def test_failing_alert_email_gets_a_line(app):
|
|
email = {"failed": 2, "retrying": 1, "said": "2 alert emails failed in the last 7 days; 1 still retrying"}
|
|
page = await _render(app, widget_view([], [_rule("r1", "a")]), email=email)
|
|
assert "2 alert emails failed in the last 7 days; 1 still retrying" in page
|
|
assert 'href="/settings/system/"' in page and "tone-crit" in page
|
|
page = await _render(app, widget_view([], [_rule("r1", "a")]), email=email, role="viewer")
|
|
assert "1 still retrying" in page and "/settings/system/" not in page
|
|
|
|
|
|
async def test_the_share_view_shows_the_alerts_without_links(app):
|
|
hot = _rule("r1", "web-1 CPU high")
|
|
page = await _render(app, widget_view([_row(hot, "firing", 94.0, NOW)], [hot]), role=None)
|
|
assert "web-1 CPU high" in page and "<a " not in page
|