feat(alerts): history reads as episodes; tests follow the confirm dialog (#5316, #5314)
CI / lint (push) Successful in 3s
CI / integration (push) Failing after 54s
CI / unit (push) Successful in 1m34s
CI / publish (push) Skipped

History (#5316):
- Transitions fold into episodes (steward/alerts/episodes.py): each stretch
  out of quiet, the worst state it reached, how long it lasted, its worst
  reading against the rule ('3m 21s (fires above 3m)'), who was told across
  its changes, and its Ansible runs. The changes themselves fold under it.
- An episode cut by paging says so ('cleared later (above)', 'earlier').
- For one rule: a verdict ('Fired 4 times in the last 7d, 1h 2m in all') and
  no Rule column. Notified and Action show only when the range has any.

Tests:
- The confirm guard looks for a confirm( call, not the word in the comment
  that says it is gone (lesson #4365).
- The Hosts viewer test looks for the add-host dialog by id, since every page
  now carries the confirm dialog.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-07 23:02:01 -04:00
co-authored by Claude Opus 5.5
parent 0c66c1e7d3
commit 531a193bd1
11 changed files with 346 additions and 61 deletions
+106
View File
@@ -0,0 +1,106 @@
"""An alert's history read as episodes (#5316): each time a rule left quiet,
what it went through, how long it lasted, its worst reading and who was told.
The history page lists transitions a page at a time (core.paging), newest
first. `episodes` folds one page of them, so an episode can straddle two
pages: its end on a newer page (`continues`) or its start on an older one
(`began_here` false). Model-free, so it is unit-tested without a database.
"""
from __future__ import annotations
from datetime import datetime
# The states an alert rests in; leaving one opens an episode, reaching one ends it.
QUIET = frozenset({"inactive", "resolved"})
# The worst thing an episode went through, most severe first.
_SEVERITY = ("firing", "acknowledged", "no_data", "pending")
_WORDS = {">": "above", ">=": "at or above", "<": "below", "<=": "at or below",
"==": "at", "!=": "other than"}
def fires_when(rule) -> str:
"""A rule's condition as a reading's aside: "fires above 3m", "fires
when down"."""
from steward.alerts.service import rule_condition
from steward.core.formatting import metric_value
said = rule_condition(rule)
if said.startswith("is "):
return f"fires when {said[3:]}"
op = rule.operator.value
return f"fires {_WORDS.get(op, op)} {metric_value(rule.metric_name, rule.threshold)}"
def _worst_value(rule, values: list[float]) -> float | None:
"""The reading furthest into breach: the highest for a rule that fires
above, the lowest for one that fires below, else the last."""
if not values:
return None
op = rule.operator.value
if op in (">", ">="):
return max(values)
if op in ("<", "<="):
return min(values)
return values[-1]
def episodes(rows: list[dict], *, newest_page: bool, now: datetime, since: datetime) -> list[dict]:
"""Fold transitions (alert history rows, newest first, each with its
`rule`, `at`, `from_state`, `to_state`, `raw` value, `deliveries`,
`muted_by`, `released`, `ansible_run_id` and `notes`) into episodes,
newest first.
Each episode: rule, steps (oldest first), start, end (None while open),
began_here (its first step left quiet on this page), ongoing (open on the
newest page: it is still going), continues (open on an older page: it
ends on a newer one), worst (the most severe state it reached), duration
(seconds, counted from `since` when it began before the range), worst
reading, deliveries, muted_by, released and runs."""
from steward.core.formatting import metric_value
open_by_rule: dict[str, dict] = {}
out: list[dict] = []
for r in reversed(rows):
rid = r["rule"].id
ep = open_by_rule.get(rid)
if ep is None:
ep = {"rule": r["rule"], "steps": [], "start": r["at"], "end": None,
"began_here": r["from_state"] in QUIET}
out.append(ep)
ep["steps"].append(r)
if r["to_state"] in QUIET:
ep["end"] = r["at"]
open_by_rule.pop(rid, None)
else:
open_by_rule[rid] = ep
for ep in out:
steps, rule = ep["steps"], ep["rule"]
reached = {s["to_state"] for s in steps} | {s["from_state"] for s in steps}
ep["worst"] = next((s for s in _SEVERITY if s in reached), steps[-1]["to_state"])
ep["ongoing"] = ep["end"] is None and newest_page
ep["continues"] = ep["end"] is None and not newest_page
began = ep["start"] if ep["began_here"] else since
ended = ep["end"] or (now if ep["ongoing"] else None)
ep["duration"] = (ended - began).total_seconds() if ended else None
worst = _worst_value(rule, [s["raw"] for s in steps if s["raw"] is not None])
ep["reading"] = None if worst is None else metric_value(rule.metric_name, worst)
ep["fires_when"] = fires_when(rule)
seen, deliveries = set(), []
for s in steps:
for d in s["deliveries"]:
if (d[0], d[1]) not in seen:
seen.add((d[0], d[1]))
deliveries.append(d)
ep["deliveries"] = deliveries
ep["muted_by"] = next((s["muted_by"] for s in steps if s["muted_by"]), None)
ep["released"] = any(s["released"] for s in steps)
ep["runs"] = [s["ansible_run_id"] for s in steps if s["ansible_run_id"]]
ep["notes"] = [n for s in steps for n in s["notes"]]
return list(reversed(out))
def summary(eps: list[dict]) -> dict:
"""How often a rule fired in a range and for how long in all: {fired,
seconds, ongoing}."""
fired = [e for e in eps if e["worst"] in ("firing", "acknowledged")]
return {"fired": len(fired), "seconds": sum(e["duration"] or 0 for e in fired),
"ongoing": any(e["ongoing"] for e in fired)}
+26 -7
View File
@@ -161,7 +161,7 @@ def _history_row(event, rule) -> dict:
return {
"at": event.transitioned_at, "rule": rule,
"from_state": event.from_state, "to_state": event.to_state,
"value": value, "condition": alert_svc.rule_condition(rule),
"value": value, "raw": event.metric_value, "condition": alert_svc.rule_condition(rule),
"notes": notes, "muted_by": muted_by, "released": released,
"deliveries": [(label, sent, error) for label, sent, error in (
("email", event.email_sent, event.email_error),
@@ -173,18 +173,31 @@ def _history_row(event, rule) -> dict:
@alerts_bp.get("/history")
@require_role(UserRole.viewer)
async def alert_history():
"""Every alert transition in a range, newest first; one rule's with ?rule=.
A page at a time as the reader scrolls: ?before=<cursor> is the next
page's rows alone (core.paging)."""
"""Every alert episode in a range, newest first; one rule's with ?rule=,
which also says how often it fired and for how long in all (#5316). A
page of transitions at a time as the reader scrolls: ?before=<cursor> is
the next page's episodes alone (core.paging). The Notified and Action
columns show only when something in the range was notified or acted on."""
from steward.alerts.episodes import episodes, summary
now = datetime.now(timezone.utc)
since, range_key = parse_range(request.args.get("range") or "7d")
before = request.args.get("before")
async with current_app.db_sessionmaker() as db:
rules = await alert_svc.list_rules(db)
chosen = next((r for r in rules if r.id == request.args.get("rule")), None)
rows = await alert_svc.history(db, since, chosen.id if chosen else None, PAGE_SIZE, before)
rule_id = chosen.id if chosen else None
rows = await alert_svc.history(db, since, rule_id, PAGE_SIZE, before)
show_notified, show_action = await alert_svc.history_columns(db, since, rule_id)
verdict = None
if chosen and not before:
every = await alert_svc.history(db, since, rule_id, HISTORY_SUMMARY_ROWS)
verdict = summary(episodes([_history_row(e, r) for e, r in every], newest_page=True,
now=now, since=since))
rows, after = next_page(rows, lambda row: (row[0].transitioned_at, row[0].id))
ctx = {"events": [_history_row(e, r) for e, r in rows], "after": after,
"rules": rules, "chosen": chosen, "current_range": range_key}
ctx = {"episodes": episodes([_history_row(e, r) for e, r in rows], newest_page=not before,
now=now, since=since),
"after": after, "rules": rules, "chosen": chosen, "current_range": range_key,
"show_notified": show_notified, "show_action": show_action, "verdict": verdict}
if before:
return await render_template("alerts/_history_page.html", **ctx)
if request.headers.get("HX-Request"):
@@ -192,6 +205,12 @@ async def alert_history():
return await render_template("alerts/history.html", **ctx)
# How many of one rule's transitions the history verdict reads ("fired N
# times, X in all"); a rule that flapped more often than this in the range
# is summed over its newest ones.
HISTORY_SUMMARY_ROWS = 2000
@alerts_bp.get("/api/rule_fields")
@require_role(UserRole.viewer)
async def rule_fields_api():
+15
View File
@@ -299,6 +299,21 @@ async def history(db, since: datetime, rule_id: str | None = None,
return [tuple(row) for row in (await db.execute(stmt)).all()]
async def history_columns(db, since: datetime, rule_id: str | None = None) -> tuple[bool, bool]:
"""(whether any transition in the range notified someone or was muted,
whether any started an Ansible run): the history page shows its Notified
and Action columns only then (#5316)."""
from steward.core.alerts import MUTED
base = select(AlertEvent.id).where(AlertEvent.transitioned_at >= since)
if rule_id:
base = base.where(AlertEvent.rule_id == rule_id)
notified = await db.scalar(select(base.where(or_(
AlertEvent.email_sent.is_not(None), AlertEvent.webhook_sent.is_not(None),
AlertEvent.note.like(f"%{MUTED}%"))).exists()))
acted = await db.scalar(select(base.where(AlertEvent.ansible_run_id.is_not(None)).exists()))
return bool(notified), bool(acted)
async def list_maintenance_windows(db) -> list[MaintenanceWindow]:
"""Every window, latest start first."""
return list((await db.execute(
+6
View File
@@ -210,6 +210,12 @@ strong, b, th { font-weight: 500; }
/* The Docker page's one-line answer, above its tiles (#5298). */
.page-verdict { margin: 0 0 0.75rem; font-size: 0.95rem; }
.table tr.row-off > td { opacity: 0.55; }
.table td.nowrap { white-space: nowrap; }
.cell-note { font-size: 0.78rem; color: var(--text-muted); margin-top: 0.2rem; }
/* An alert episode's own transitions, folded under it (#5316). */
.episode-steps { margin-top: 0.3rem; font-size: 0.8rem; }
.episode-steps summary { cursor: pointer; color: var(--text-muted); }
.episode-steps ol { margin: 0.35rem 0 0; padding-left: 1.1rem; display: grid; gap: 0.25rem; }
/* A problem section's heading carries its own count, so no tile repeats it (#5214). */
.section-count { margin-bottom: 0.6rem; }
.section-count > span:first-child { margin-left: 0.25rem; }
+60 -37
View File
@@ -1,51 +1,74 @@
{# alerts/_history_page.html — one page of alert transitions, then the sentinel
that loads the next page as the reader nears it (core.paging, pref 172). #}
{# alerts/_history_page.html — one page of alert episodes (#5316), then the
sentinel that loads the next page as the reader nears it (core.paging,
pref 172). An episode is one stretch out of quiet: what it reached, for how
long, its worst reading, who was told; its transitions fold under it. #}
{% from "_macros.html" import status_badge, scroll_more, scroll_end %}
{% for ev in events %}
{% set cols = 3 + (0 if chosen else 1) + (1 if show_notified else 0) + (1 if show_action else 0) %}
{% for ep in episodes %}
<tr>
<td style="font-size:0.82rem;white-space:nowrap;">{{ ev.at|ago }}</td>
<td class="nowrap">
{%- if ep.began_here %}{{ ep.start|ago }}{% else %}<span class="tone-dim" title="It began before this part of the list">earlier</span>{% endif -%}
</td>
{% if not chosen %}
<td class="cell-name">
<a href="/alerts/history?rule={{ ep.rule.id }}&range={{ current_range }}" title="{{ ep.rule.name }}">{{ ep.rule.name }}</a>
<div class="cell-sub">{{ source_label(ep.rule.source_module) }} · {{ ep.rule.resource_name }}</div>
</td>
{% endif %}
<td>
<a href="/alerts/history?rule={{ ev.rule.id }}&range={{ current_range }}">{{ ev.rule.name }}</a>
<div style="font-size:0.78rem;color:var(--text-dim);">{{ source_label(ev.rule.source_module) }} · {{ ev.rule.resource_name }}</div>
</td>
<td style="white-space:nowrap;">
{{ status_badge(ev.from_state) }}
<span style="color:var(--text-dim);">→</span>
{{ status_badge(ev.to_state) }}
{% for n in ev.notes %}
<div style="font-size:0.78rem;color:var(--text-muted);margin-top:0.25rem;white-space:normal;">{{ n }}</div>
{% endfor %}
</td>
<td style="font-size:0.85rem;white-space:nowrap;">
{{ ev.value }}
<div style="font-size:0.75rem;color:var(--text-dim);">fires when {{ ev.condition }}</div>
</td>
<td style="font-size:0.82rem;">
{% if ev.muted_by %}
<span class="badge badge-yellow" title="A maintenance window muted this notification">muted · {{ ev.muted_by }}</span>
{% if ev.released %}<div style="color:var(--text-muted);margin-top:0.25rem;">notified when the window ended</div>{% endif %}
{% if not ep.began_here and ep.steps|length == 1 %}
{%- set s = ep.steps[0] %}{{ status_badge(s.from_state) }} <span class="tone-dim">→</span> {{ status_badge(s.to_state) }}
{% else %}
{{ status_badge(ep.worst) }}
{%- if ep.ongoing %} <span class="tone-crit">{% if ep.began_here %}for {{ ep.duration|duration }}, {% endif %}still going</span>
{%- elif ep.continues %} <span class="tone-muted">cleared later (above)</span>
{%- elif ep.began_here %} for {{ ep.duration|duration }}<span class="tone-dim"> · cleared {{ ep.end|ago }}</span>
{%- else %} <span class="tone-dim">cleared {{ ep.end|ago }}</span>{% endif %}
{% endif %}
{% for label, sent, error in ev.deliveries %}
<div style="margin-top:0.15rem;">
{% if sent %}
<span class="badge badge-green">{{ label }} sent</span>
{% else %}
<span class="badge badge-red">{{ label }} failed</span>
{% if error %}<div style="color:var(--text-muted);font-size:0.78rem;margin-top:0.15rem;max-width:28rem;word-break:break-word;">{{ error }}</div>{% endif %}
{% for n in ep.notes %}<div class="cell-note">{{ n }}</div>{% endfor %}
{% if ep.steps|length > 1 %}
<details class="episode-steps">
<summary>{{ ep.steps|length }} changes of state</summary>
<ol>
{% for s in ep.steps %}
<li>{{ status_badge(s.from_state) }} <span class="tone-dim">→</span> {{ status_badge(s.to_state) }}
<span class="tone-muted">{{ s.value }}</span> <span class="tone-dim">· {{ s.at|ago }}</span></li>
{% endfor %}
</ol>
</details>
{% endif %}
</td>
<td class="nowrap">
{%- if ep.reading %}{{ ep.reading }} <span class="tone-dim">({{ ep.fires_when }})</span>
{%- else %}<span class="tone-dim">no reading</span> <span class="tone-dim">({{ ep.fires_when }})</span>{% endif -%}
</td>
{% if show_notified %}
<td>
{% if ep.muted_by %}
<span class="badge badge-yellow" title="A maintenance window muted this notification">muted · {{ ep.muted_by }}</span>
{% if ep.released %}<div class="cell-note">notified when the window ended</div>{% endif %}
{% endif %}
{% for label, sent, error in ep.deliveries %}
<div>
{% if sent %}<span class="badge badge-green">{{ label }} sent</span>
{% else %}<span class="badge badge-red">{{ label }} failed</span>
{% if error %}<div class="cell-note" style="max-width:28rem;overflow-wrap:anywhere;">{{ error }}</div>{% endif %}
{% endif %}
</div>
{% endfor %}
{% if not ev.muted_by and not ev.deliveries %}<span style="color:var(--text-dim);" title="No notification was sent for this change">—</span>{% endif %}
{% if not ep.muted_by and not ep.deliveries %}<span class="tone-dim" title="No notification was sent for this">—</span>{% endif %}
</td>
<td style="font-size:0.82rem;">
{% if ev.ansible_run_id %}
<a href="/ansible/runs/{{ ev.ansible_run_id }}">Ansible run</a>
{% else %}<span style="color:var(--text-dim);">—</span>{% endif %}
{% endif %}
{% if show_action %}
<td>
{% for run_id in ep.runs %}<div><a href="/ansible/runs/{{ run_id }}">Ansible run</a></div>
{% else %}<span class="tone-dim">—</span>{% endfor %}
</td>
{% endif %}
</tr>
{% endfor %}
{% if after %}
{{ scroll_more("/alerts/history?before=" ~ (after|urlencode) ~ "&range=" ~ (current_range|urlencode) ~ (("&rule=" ~ (chosen.id|urlencode)) if chosen else ""), cols=6) }}
{{ scroll_more("/alerts/history?before=" ~ (after|urlencode) ~ "&range=" ~ (current_range|urlencode) ~ (("&rule=" ~ (chosen.id|urlencode)) if chosen else ""), cols=cols) }}
{% else %}
{{ scroll_end(cols=6) }}
{{ scroll_end(cols=cols) }}
{% endif %}
+23 -11
View File
@@ -1,17 +1,29 @@
{# alerts/_history_rows.html — the history table; swapped in on a range or rule
change. Its rows come a page at a time (alerts/_history_page.html). #}
{% from "_macros.html" import status_badge, empty_state %}
{% if events %}
<div class="card-flush" style="overflow-x:auto;">
<table class="table">
{# alerts/_history_rows.html — the history: for one rule, how often it fired
and for how long in all; then its episodes (#5316), swapped in on a range
or rule change. The rule column shows only across rules, Notified and
Action only when the range has something to put in them. Episodes come a
page of transitions at a time (alerts/_history_page.html). #}
{% from "_macros.html" import empty_state %}
{% if chosen and verdict %}
<p class="page-verdict">
{%- if verdict.fired %}<strong class="tone-{{ 'crit' if verdict.ongoing else 'warn' }}">Fired {{ verdict.fired }} time{{ "s" if verdict.fired != 1 }}</strong> in the last {{ current_range }}
{%- if verdict.seconds %}, {{ verdict.seconds|duration }} in all{% endif %}
{%- if verdict.ongoing %} · <span class="tone-crit">firing now</span>{% endif %}
{%- else %}<strong class="tone-ok">Never fired</strong> in the last {{ current_range }}{% endif %}
<span class="tone-dim">· fires when {{ rule_condition(chosen) }} on {{ chosen.resource_name }}</span>
</p>
{% endif %}
{% if episodes %}
<div class="card-flush">
<table class="table alert-episodes">
<thead>
<tr>
<th>When</th>
<th>Rule</th>
<th>Change</th>
<th>Reading</th>
<th>Notified</th>
<th>Action</th>
{% if not chosen %}<th>Rule</th>{% endif %}
<th>What happened</th>
<th title="The reading furthest past the rule's threshold">Worst reading</th>
{% if show_notified %}<th>Notified</th>{% endif %}
{% if show_action %}<th>Action</th>{% endif %}
</tr>
</thead>
<tbody>
+1 -1
View File
@@ -3,7 +3,7 @@
{% block title %}Alert History — Steward{% endblock %}
{% block breadcrumb %}{{ crumbs([("Alerts", "/alerts/"), ("History", "")]) }}{% endblock %}
{% block content %}
{% call page_header("Alert history", "Every change of state, newest first: the value against the rule's condition, whether the email and webhook went out, and the Ansible run an action started.") %}
{% call page_header("Alert history", "Each time a rule left quiet, newest first: what it reached and for how long, its worst reading, and who was told.") %}
<select id="history-rule" name="rule" aria-label="Rule" class="input" style="width:auto;"
hx-get="/alerts/history" hx-target="#history-rows" hx-include="#time-range"
hx-trigger="change" hx-push-url="false"
+98
View File
@@ -0,0 +1,98 @@
"""Alert history read as episodes (#5316): each stretch out of quiet, how
long it lasted, its worst reading, who was told; an episode that straddles a
page of the list says so instead of guessing."""
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace as NS
from steward.alerts.episodes import episodes, fires_when, summary
from steward.models.alerts import AlertOperator
NOW = datetime(2026, 10, 8, 12, 0, tzinfo=timezone.utc)
SINCE = NOW - timedelta(days=7)
def _rule(rid="r1", metric="seconds_since_seen", op=AlertOperator.gt, threshold=180.0):
return NS(id=rid, name=f"rule {rid}", metric_name=metric, operator=op, threshold=threshold,
resource_name="VDNT-Docker02", source_module="host_agent")
def _t(rule, mins_ago, frm, to, raw=None, **kw):
return {"rule": rule, "at": NOW - timedelta(minutes=mins_ago), "from_state": frm, "to_state": to,
"raw": raw, "value": f"{raw}", "notes": kw.get("notes", []), "deliveries": kw.get("deliveries", []),
"muted_by": kw.get("muted_by"), "released": False, "ansible_run_id": kw.get("run")}
def _newest_first(*rows):
return sorted(rows, key=lambda r: r["at"], reverse=True)
def test_a_firing_and_its_resolve_are_one_episode():
r = _rule()
eps = episodes(_newest_first(_t(r, 402, "inactive", "firing", 201.0), _t(r, 389, "firing", "resolved", 19.0)),
newest_page=True, now=NOW, since=SINCE)
assert len(eps) == 1
ep = eps[0]
assert ep["worst"] == "firing" and ep["began_here"] and not ep["ongoing"]
assert ep["duration"] == 13 * 60 and ep["reading"] == "3m 21s" # the highest: it fires above
assert ep["fires_when"] == "fires above 3m"
def test_silence_then_firing_is_one_episode_and_the_worst_is_firing():
"""The live rule's shape: inactive → no data, inactive → firing (an
alert at rest records no_data as a label, not a state), no data → firing,
firing → resolved."""
r = _rule()
eps = episodes(_newest_first(
_t(r, 1397, "inactive", "no_data", notes=["no sample for 7m 54s (usually every 1m)"]),
_t(r, 1396, "inactive", "firing", 473.0),
_t(r, 1395, "no_data", "firing", 533.0, notes=["data resumed after 3m 54s silent"]),
_t(r, 1394, "firing", "resolved", 42.0)), newest_page=True, now=NOW, since=SINCE)
assert len(eps) == 1 and len(eps[0]["steps"]) == 4
assert eps[0]["worst"] == "firing" and eps[0]["duration"] == 3 * 60
assert eps[0]["notes"] == ["no sample for 7m 54s (usually every 1m)", "data resumed after 3m 54s silent"]
def test_episodes_are_newest_first_and_kept_apart_per_rule():
a, b = _rule("a"), _rule("b", "cpu_pct", AlertOperator.lt, 5.0)
eps = episodes(_newest_first(
_t(a, 60, "inactive", "firing", 200.0), _t(b, 50, "inactive", "firing", 2.0),
_t(a, 40, "firing", "resolved", 10.0), _t(b, 30, "firing", "resolved", 9.0),
_t(b, 20, "resolved", "firing", 1.0)), newest_page=True, now=NOW, since=SINCE)
assert [(e["rule"].id, e["start"]) for e in eps] == [
("b", NOW - timedelta(minutes=20)), ("b", NOW - timedelta(minutes=50)), ("a", NOW - timedelta(minutes=60))]
assert eps[0]["ongoing"] and eps[0]["duration"] == 20 * 60
assert eps[1]["reading"] == "2.0%" and eps[1]["fires_when"] == "fires below 5.0%" # the lowest: it fires below
def test_an_episode_split_by_paging_says_so():
r = _rule()
# Its end is on a newer page: this older page cannot know when it cleared.
older = episodes([_t(r, 60, "inactive", "firing", 200.0)], newest_page=False, now=NOW, since=SINCE)
assert older[0]["continues"] and not older[0]["ongoing"] and older[0]["duration"] is None
# Its start is on an older page: it began before what this page holds.
newer = episodes([_t(r, 10, "firing", "resolved", 5.0)], newest_page=True, now=NOW, since=SINCE)
assert not newer[0]["began_here"] and newer[0]["end"] == NOW - timedelta(minutes=10)
def test_who_was_told_is_gathered_once_per_episode():
r = _rule()
eps = episodes(_newest_first(
_t(r, 30, "inactive", "firing", 200.0, deliveries=[("email", False, "timed out"), ("webhook", True, None)],
run="run-1"),
_t(r, 20, "firing", "resolved", 5.0, deliveries=[("webhook", True, None)], muted_by="patch")),
newest_page=True, now=NOW, since=SINCE)
assert eps[0]["deliveries"] == [("email", False, "timed out"), ("webhook", True, None)]
assert eps[0]["muted_by"] == "patch" and eps[0]["runs"] == ["run-1"]
def test_the_verdict_counts_what_fired_and_for_how_long():
r = _rule()
eps = episodes(_newest_first(
_t(r, 100, "inactive", "firing", 200.0), _t(r, 90, "firing", "resolved", 5.0),
_t(r, 50, "inactive", "no_data"), _t(r, 45, "inactive", "resolved"),
_t(r, 10, "inactive", "firing", 300.0)), newest_page=True, now=NOW, since=SINCE)
assert summary(eps) == {"fired": 2, "seconds": 10 * 60 + 10 * 60, "ongoing": True}
def test_an_up_or_down_rule_says_the_word():
assert fires_when(_rule(metric="is_up", op=AlertOperator.lt, threshold=0.5)) == "fires when down"
+1 -1
View File
@@ -127,7 +127,7 @@ async def test_add_host_is_a_dialog_on_the_hosts_list(app):
async with app.test_request_context("/hosts/"):
session["user_role"] = "viewer"
viewer = await render_template("hosts/list.html", **_list_ctx(add={}))
assert "<dialog" not in viewer and ">Add host<" not in viewer
assert 'id="add-host"' not in viewer and ">Add host<" not in viewer
async def test_add_host_sends_a_clash_back_with_what_was_typed(app):
+9 -3
View File
@@ -68,14 +68,19 @@ def test_a_viewer_reads_the_alert_history(app):
page = await resp.get_data(as_text=True)
assert "Alert History" in page and 'href="/alerts/"' in page # crumbs back to Alerts
assert f"hot {tag}" in page and f"cold-note-{tag}" in page and f"old-{tag}" not in page
assert "CPU 98%" in page and "fires when CPU &gt; 90%" in page
# One episode per stretch out of quiet (#5316): its worst reading against
# the rule, who was told across its changes, the changes folded under it.
assert '98% <span class="tone-dim">(fires above 90%)</span>' in page
assert "email failed" in page and f"mx-{tag}: timed out" in page and "webhook sent" in page
assert f"muted · patch-{tag}" in page and "No CPU reading" in page
assert f"muted · patch-{tag}" in page and "No CPU reading" in page and "2 changes of state" in page
assert "<th>Rule</th>" in page and "<th>Notified</th>" in page and "<th>Action</th>" not in page
resp = await client.get(f"/alerts/history?range=24h&rule={hot_id}", headers={"HX-Request": "true"})
rows = await resp.get_data(as_text=True)
assert "<html" not in rows and "<table" in rows
assert f"old-{tag}" in rows and f"cold-note-{tag}" not in rows
assert "<th>Rule</th>" not in rows # one rule: no column for it
assert "Fired 1 time</strong> in the last 24h" in rows # its verdict
# The next page as the reader scrolls: only older rows, without the table
# around them, and the list says where it ends (#5211).
@@ -84,7 +89,8 @@ def test_a_viewer_reads_the_alert_history(app):
"range": "24h", "rule": hot_id, "before": cursor(now - timedelta(minutes=3), "")})
older = await resp.get_data(as_text=True)
assert resp.status_code == 200 and "<table" not in older
assert "CPU 98%" in older and f"old-{tag}" in older and f"patch-{tag}" not in older
assert "98%" in older and f"old-{tag}" in older and f"patch-{tag}" not in older
assert "cleared later (above)" in older # its end is on the newer page
assert "End of the list." in older and "data-scroll-more" not in older
# The rules list links each rule to its history.
+1 -1
View File
@@ -47,5 +47,5 @@ def test_base_asks_before_a_data_confirm_form_submits():
assert re.search(r'<button type="submit"[^>]*btn-danger[^>]*value="go"', base)
assert "dataset.confirm" in js and "getElementById('confirm-dialog')" in js and "requestSubmit" in js
assert "textContent = title" in js and "innerHTML" not in js # text, never markup
assert "window.confirm" not in js and "window.confirm" not in base
assert not re.search(r"\bconfirm\s*\(", js + base) # a call, not the comment saying it is gone
assert any("data-confirm=" in p.read_text() for p in _templates())