bvandeusenandClaude Opus 5.5
e37cd0a873
The all-rules page reads a Postgres shared with other tests, whose Ansible-run events turn the column on. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
103 lines
5.5 KiB
Python
103 lines
5.5 KiB
Python
"""Integration: /alerts/history (#5066) against real Postgres. A viewer can
|
|
read it; it shows the transition, value against the condition, delivery
|
|
outcome with its error, and a maintenance mute; ?rule= narrows it to one
|
|
rule, an HTMX request gets only the rows, and ?before= the next page."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import time
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
if not os.environ.get("STEWARD_DATABASE_URL"):
|
|
pytest.skip("needs Postgres")
|
|
from steward.app import create_app
|
|
return create_app(testing=False)
|
|
|
|
|
|
def test_a_viewer_reads_the_alert_history(app):
|
|
from steward.core.alerts import MUTED
|
|
from steward.models.alerts import AlertEvent, AlertOperator, AlertRule
|
|
from steward.models.users import User, UserRole
|
|
|
|
async def _go():
|
|
tag = uuid.uuid4().hex[:8]
|
|
uid = str(uuid.uuid4())
|
|
now = datetime.now(timezone.utc)
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
s.add(User(id=uid, username=f"hist-{tag}", email=f"hist-{tag}@test",
|
|
password_hash="x", role=UserRole.viewer, is_active=True))
|
|
await s.flush()
|
|
hot = AlertRule(name=f"hot {tag}", source_module="host_agent", resource_name=f"box-{tag}",
|
|
metric_name="cpu_pct", operator=AlertOperator(">"), threshold=90,
|
|
created_by=uid)
|
|
cold = AlertRule(name=f"cold {tag}", source_module="host_agent", resource_name=f"box-{tag}",
|
|
metric_name="cpu_pct", operator=AlertOperator("<"), threshold=1,
|
|
created_by=uid)
|
|
s.add_all([hot, cold])
|
|
await s.flush()
|
|
s.add_all([
|
|
AlertEvent(rule_id=hot.id, from_state="inactive", to_state="firing", metric_value=97.5,
|
|
transitioned_at=now - timedelta(minutes=5), email_sent=False,
|
|
email_error=f"mx-{tag}: timed out", webhook_sent=True),
|
|
AlertEvent(rule_id=hot.id, from_state="firing", to_state="resolved", metric_value=None,
|
|
transitioned_at=now - timedelta(minutes=2), note=f"{MUTED} patch-{tag}"),
|
|
AlertEvent(rule_id=cold.id, from_state="inactive", to_state="firing", metric_value=0.5,
|
|
transitioned_at=now - timedelta(minutes=1), note=f"cold-note-{tag}"),
|
|
# Outside a 1h range.
|
|
AlertEvent(rule_id=hot.id, from_state="pending", to_state="firing", metric_value=95,
|
|
transitioned_at=now - timedelta(hours=3), note=f"old-{tag}"),
|
|
])
|
|
hot_id = hot.id
|
|
|
|
client = app.test_client()
|
|
async with client.session_transaction() as sess:
|
|
sess.update(user_id=uid, user_role="viewer", username=f"hist-{tag}", login_at=time.time())
|
|
|
|
resp = await client.get("/alerts/history?range=1h")
|
|
assert resp.status_code == 200
|
|
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
|
|
# 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 and "2 changes of state" in page
|
|
assert "<th>Rule</th>" in page and "<th>Notified</th>" 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
|
|
# Columns only for what this rule's rows hold: it told someone, it ran
|
|
# nothing. (The all-rules page shares Postgres with other tests' runs.)
|
|
assert "<th>Notified</th>" in rows and "<th>Action</th>" not in rows
|
|
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).
|
|
from steward.core.paging import cursor
|
|
resp = await client.get("/alerts/history", query_string={
|
|
"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 "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.
|
|
assert f'/alerts/history?rule={hot_id}' in await (await client.get("/alerts/")).get_data(as_text=True)
|
|
|
|
asyncio.run(_go())
|