FabledSteward/tests/core/test_alerts_page.py

"""The Alerts page says each alert as a sentence around a link to what it
watches, offers only the actions the viewer's role can take, and the sidebar
counts what needs attention (#5124)."""
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace

import pytest
from quart import render_template, session

from steward.alerts import service as alert_svc
from steward.alerts.service import alert_sentence
from steward.models.alerts import AlertOperator, AlertStateEnum
from tests.test_style_tokens import hex_colours


def _rule(metric="cpu_pct", op=AlertOperator.gt, threshold=90.0, resource="web-1", rid="r1"):
    return SimpleNamespace(id=rid, name=f"{resource} {metric}", source_module="host_agent",
                           resource_name=resource, metric_name=metric, operator=op,
                           threshold=threshold, enabled=True,
                           consecutive_failures_required=1, consecutive_ok_required=1)


def _said(says):
    return says["lead"] + says["resource"] + says["tail"]


@pytest.mark.parametrize("op, threshold, value, reads", [
    (AlertOperator.gt, 90.0, 94.0, "cpu_pct on web-1 is 94% (> 90%), firing for 12m"),
    (AlertOperator.gte, 90.0, 90.0, "cpu_pct on web-1 is 90% (>= 90%), firing for 12m"),
    (AlertOperator.lt, 5.0, 2.5, "cpu_pct on web-1 is 2.5% (< 5.0%), firing for 12m"),
    (AlertOperator.lte, 5.0, 5.0, "cpu_pct on web-1 is 5.0% (<= 5.0%), firing for 12m"),
    (AlertOperator.eq, 100.0, 100.0, "cpu_pct on web-1 is 100% (== 100%), firing for 12m"),
    (AlertOperator.ne, 0.0, 3.0, "cpu_pct on web-1 is 3.0% (!= 0.0%), firing for 12m"),
])
def test_a_firing_alert_reads_as_one_sentence(op, threshold, value, reads):
    now = datetime.now(timezone.utc)
    says = alert_sentence(_rule(op=op, threshold=threshold), "firing", value,
                          now - timedelta(minutes=12), now)
    assert says["resource"] == "web-1"
    assert _said(says) == reads


def test_units_follow_the_metric_name():
    now = datetime.now(timezone.utc)
    says = alert_sentence(_rule("response_ms", threshold=200.0, resource="api"), "pending", 431.0,
                          now - timedelta(seconds=45), now)
    assert _said(says) == "response_ms on api is 431 ms (> 200 ms), pending for 45s"


def test_an_up_or_down_metric_says_the_word():
    now = datetime.now(timezone.utc)
    says = alert_sentence(_rule("is_up", AlertOperator.lt, 1.0, "nas"), "acknowledged", 0.0,
                          now - timedelta(hours=2, minutes=5), now)
    assert _said(says) == "nas is down, acknowledged for 2h 5m"


def test_no_reading_yet_is_said_not_shown_as_a_number():
    now = datetime.now(timezone.utc)
    says = alert_sentence(_rule(), "firing", None, now - timedelta(minutes=3), now)
    assert _said(says) == "cpu_pct on web-1 has no reading yet (> 90%), firing for 3m"


def test_a_silent_series_says_how_long_it_has_been_quiet():
    now = datetime.now(timezone.utc)
    says = alert_sentence(_rule(), "no_data", 50.0, now - timedelta(minutes=20), now)
    assert _said(says) == "No cpu_pct from web-1 for 20m"


def _row(rule, label, url, state=AlertStateEnum.firing):
    now = datetime.now(timezone.utc)
    st = SimpleNamespace(state=state, no_data_since=None, last_sample_at=None,
                         last_transition_at=now - timedelta(minutes=12))
    return {"state": st, "rule": rule, "label": label, "url": url,
            "says": alert_sentence(rule, label, 94.0, st.last_transition_at, now)}


async def test_an_operator_sees_the_sentence_linked_and_can_acknowledge(app):
    rule = _rule()
    async with app.test_request_context("/alerts/"):
        session["user_role"] = "operator"
        page = await render_template("alerts/list.html", active=[_row(rule, "firing", "/hosts/h1")],
                                     rules=[rule], operators=[])

    assert 'cpu_pct on <a href="/hosts/h1">web-1</a> is 94% (&gt; 90%), firing for 12m' in page
    assert '<span class="badge badge-crit">Firing</span>' in page
    assert 'action="/alerts/r1/acknowledge"' in page
    assert 'href="/alerts/rules/new"' in page and 'href="/alerts/rules/r1/edit"' in page
    assert 'action="/alerts/rules/r1/delete"' not in page  # deleting a rule is the admin's
    main = page[page.index("<main"):]  # the sidebar brand mark carries its own colours
    assert "onclick" not in main and not hex_colours(main), hex_colours(main)


async def test_a_viewer_reads_but_gets_no_actions(app):
    rule = _rule()
    async with app.test_request_context("/alerts/"):
        session["user_role"] = "viewer"
        page = await render_template("alerts/list.html", active=[_row(rule, "firing", None)],
                                     rules=[rule], operators=[])

    assert "cpu_pct on <strong>web-1</strong>" in page  # nothing to link to: no bare href
    for action in ("/acknowledge", "/alerts/rules/new", "/edit", "/toggle", "/delete"):
        assert action not in page, action
    assert 'href="/alerts/history?rule=r1"' in page


async def test_an_admin_can_delete_a_rule(app):
    rule = _rule()
    async with app.test_request_context("/alerts/"):
        session["user_role"] = "admin"
        page = await render_template("alerts/list.html", active=[], rules=[rule], operators=[])
    assert 'action="/alerts/rules/r1/delete"' in page
    assert 'data-confirm="Delete rule web-1 cpu_pct?"' in page


async def _active_count(_db):
    return 3


async def test_the_sidebar_counts_alerts_needing_attention(app, monkeypatch):
    monkeypatch.setattr(alert_svc, "active_count", _active_count)
    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="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):
    monkeypatch.setattr(alert_svc, "active_count", _active_count)
    async with app.test_request_context("/alerts/", headers={"HX-Request": "true"}):
        session["user_id"], session["user_role"] = "u1", "viewer"
        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 "count-badge" not in fragment and "count-badge" not in logged_out