patrickandClaude Sonnet 5.5
347f55cf7c
Includes atomic migrations, per-rule reminder semantics, first-run helper script and the note on the undecided UI chat idea. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
175 lines
8.6 KiB
Python
175 lines
8.6 KiB
Python
import re
|
|
import time
|
|
|
|
from app import audit
|
|
from app.models import KIND_SYSTEM, Event
|
|
from tests.test_web import (ADMIN, VIEWER, admin, app, client, csrf_from, login_ok, post, # noqa: F401
|
|
users, viewer)
|
|
|
|
ALERTING = "/settings/alerting"
|
|
|
|
|
|
def alerting_form(**overrides):
|
|
data = {"poll_interval": "30", "alerts_default_delay_minutes": "15", "alerts_reminder_hours": "0",
|
|
"alerts_recovery_enabled": "1", "alerts_bundle_seconds": "60", "alerts_ai_per_minute": "10",
|
|
"alerts_host_delay_minutes": "0", "alerts_send_limit_per_minute": "15", "ui_base_url": ""}
|
|
data.update(overrides)
|
|
return data
|
|
|
|
|
|
def insert_state(app, host, service, state, output="x", **columns):
|
|
values = {"host": host, "service": service, "state": state, "output": output, "last_state_change": 1000,
|
|
"reported_state": state, **columns}
|
|
names = ", ".join(values)
|
|
app.state.db.execute(f"INSERT INTO service_states({names}) VALUES ({', '.join('?' for _ in values)})",
|
|
tuple(values.values()))
|
|
|
|
|
|
def test_alerting_page_shows_new_fields_and_baseline_button(client, admin):
|
|
login_ok(client)
|
|
page = client.get(ALERTING).text
|
|
for text in ("Host-Verzögerung in Minuten", "Sendelimit: Nachrichten pro Minute und Ziel",
|
|
"Basis-URL dieser Oberfläche", "Baseline neu setzen", 'name="alerts_host_delay_minutes"',
|
|
'name="alerts_send_limit_per_minute"', 'name="ui_base_url"'):
|
|
assert text in page, text
|
|
assert f'data-test-url="{ALERTING}/baseline"' in page
|
|
assert "Regel-Verzögerungen gelten nicht für Host-Alarme" in page
|
|
|
|
|
|
def test_alerting_defaults_are_shown(client, admin):
|
|
login_ok(client)
|
|
page = client.get(ALERTING).text
|
|
assert re.search(r'name="alerts_send_limit_per_minute" type="text" value="15"', page)
|
|
assert re.search(r'name="alerts_host_delay_minutes" type="text" value="0"', page)
|
|
|
|
|
|
def test_alerting_saves_new_values_and_strips_trailing_slash(client, app, admin):
|
|
login_ok(client)
|
|
form = alerting_form(alerts_host_delay_minutes="5", alerts_send_limit_per_minute="20",
|
|
ui_base_url="https://bot.example.de/")
|
|
assert post(client, ALERTING, ALERTING, **form).status_code == 303
|
|
store = app.state.store
|
|
assert store.get_int("alerts.host_delay_minutes", 0) == 5
|
|
assert store.get_int("alerts.send_limit_per_minute", 15) == 20
|
|
assert store.get("ui.base_url") == "https://bot.example.de"
|
|
|
|
|
|
def test_alerting_rejects_invalid_new_values(client, app, admin):
|
|
login_ok(client)
|
|
for override in ({"alerts_host_delay_minutes": "1441"}, {"alerts_host_delay_minutes": "-1"},
|
|
{"alerts_host_delay_minutes": "abc"}, {"alerts_send_limit_per_minute": "0"},
|
|
{"alerts_send_limit_per_minute": "61"}, {"alerts_send_limit_per_minute": "1.5"},
|
|
{"ui_base_url": "ftp://bot.example.de"}, {"ui_base_url": "bot.example.de"},
|
|
{"ui_base_url": "https://bot example.de"}, {"ui_base_url": "https://bot.example.de/?a=1"},
|
|
{"ui_base_url": "https://bot.example.de/#x"}, {"ui_base_url": "javascript:alert(1)"},
|
|
{"ui_base_url": "https://" + "a" * 200}):
|
|
response = post(client, ALERTING, ALERTING, **alerting_form(**override))
|
|
assert response.status_code == 422, override
|
|
assert app.state.store.get("ui.base_url") in (None, "")
|
|
|
|
|
|
def test_empty_base_url_clears_the_link(client, app, admin):
|
|
login_ok(client)
|
|
post(client, ALERTING, ALERTING, **alerting_form(ui_base_url="https://bot.example.de"))
|
|
post(client, ALERTING, ALERTING, **alerting_form(ui_base_url=""))
|
|
assert not app.state.store.get("ui.base_url")
|
|
|
|
|
|
def test_content_mode_select_on_telegram_and_matrix(client, app, admin):
|
|
login_ok(client)
|
|
for slug, key in (("telegram", "telegram.content_mode"), ("matrix", "matrix.content_mode")):
|
|
page = client.get(f"/settings/{slug}").text
|
|
field = f"{slug}_content_mode"
|
|
assert f'<select id="f-{field}" name="{field}">' in page
|
|
assert re.search(r'value="voll"\s+selected', page) and 'value="kurz"' in page
|
|
assert post(client, f"/settings/{slug}", f"/settings/{slug}", **{field: "kurz"}).status_code == 303
|
|
assert app.state.store.get(key) == "kurz"
|
|
assert re.search(r'value="kurz"\s+selected', client.get(f"/settings/{slug}").text)
|
|
assert post(client, f"/settings/{slug}", f"/settings/{slug}", **{field: "bogus"}).status_code == 422
|
|
assert app.state.store.get(key) == "kurz"
|
|
|
|
|
|
def test_telegram_hint_mentions_external_servers(client, admin):
|
|
login_ok(client)
|
|
assert "Telegram-Server liegen außerhalb" in client.get("/settings/telegram").text
|
|
|
|
|
|
def test_content_mode_defaults_to_full_when_unset(app):
|
|
from app.alerts import content_mode_provider
|
|
mode = content_mode_provider(app.state.store)
|
|
assert mode("telegram") == "voll" and mode("matrix") == "voll" and mode("other") == "voll"
|
|
app.state.store.set("telegram.content_mode", "kurz", "t")
|
|
assert mode("telegram") == "kurz" and mode("matrix") == "voll"
|
|
|
|
|
|
def test_baseline_button_sets_request_and_audits(client, app, admin):
|
|
login_ok(client)
|
|
response = post(client, f"{ALERTING}/baseline", ALERTING)
|
|
assert response.status_code == 200 and response.json()["ok"] is True
|
|
marker = app.state.db.query("SELECT value FROM app_state WHERE key='baseline.requested'")
|
|
assert [r["value"] for r in marker] == ["1"]
|
|
assert ("baseline.request", ADMIN) in [(e["action"], e["user"]) for e in audit.recent(app.state.db)]
|
|
|
|
|
|
def test_baseline_button_needs_admin_and_csrf(client, app, viewer):
|
|
login_ok(client, VIEWER)
|
|
token = csrf_from(client, "/")
|
|
assert client.post(f"{ALERTING}/baseline", data={"csrf": token}).status_code == 403
|
|
assert client.post(f"{ALERTING}/baseline").status_code == 403
|
|
assert app.state.db.query("SELECT 1 FROM app_state WHERE key='baseline.requested'") == []
|
|
|
|
|
|
def test_baseline_button_without_csrf_is_rejected_for_admin(client, app, admin):
|
|
login_ok(client)
|
|
assert client.post(f"{ALERTING}/baseline", data={"csrf": "falsch"}).status_code == 403
|
|
assert app.state.db.query("SELECT 1 FROM app_state WHERE key='baseline.requested'") == []
|
|
|
|
|
|
def test_dashboard_lists_host_problems_first(client, app, viewer):
|
|
insert_state(app, "web01", "Disk", 2, "voll")
|
|
insert_state(app, "a-app", "CPU", 1, "last")
|
|
insert_state(app, "srv9", "Host", 2, "no route")
|
|
insert_state(app, "srv8", "Host", 3, "via parent")
|
|
login_ok(client, VIEWER)
|
|
page = client.get("/").text
|
|
positions = [page.index(name) for name in ("srv8", "srv9", "web01", "a-app")]
|
|
assert positions[0] < positions[2] and positions[1] < positions[2] and positions[1] < positions[3]
|
|
assert "DOWN" in page and "UNREACH" in page and "no route" in page
|
|
|
|
|
|
def test_dashboard_labels_host_down_and_acknowledged_services(client, app, viewer):
|
|
insert_state(app, "srv9", "Host", 2, "no route", reported_state=2)
|
|
insert_state(app, "srv9", "Disk", 2, "voll", reported_state=0, host_held=1)
|
|
insert_state(app, "web01", "Load", 2, "hoch", ack_state=2)
|
|
login_ok(client, VIEWER)
|
|
page = client.get("/").text
|
|
assert "unterdrückt (Host down)" in page and "unterdrückt (bestätigt)" in page
|
|
|
|
|
|
def test_dashboard_host_waits_with_host_delay_not_rule_delay(client, app, viewer):
|
|
app.state.store.set("alerts.host_delay_minutes", "10", "t")
|
|
insert_state(app, "srv9", "Host", 2, "no route", reported_state=0, last_state_change=int(time.time()) - 60)
|
|
login_ok(client, VIEWER)
|
|
assert "wartet noch 9 min" in client.get("/").text
|
|
|
|
|
|
def test_alerts_page_has_anchor_labels_and_host_state(client, app, viewer):
|
|
runtime = app.state.runtime
|
|
now = int(time.time())
|
|
events = [
|
|
Event(now, "web01", "Disk", "problem", 2, "voll", suppressed=True, suppress_reason="bestaetigt"),
|
|
Event(now, "web01", "Load", "problem", 2, "hoch", suppressed=True, suppress_reason="host_down"),
|
|
Event(now, "web01", "Mem", "problem", 2, "knapp", suppressed=True, suppress_reason="sendelimit"),
|
|
Event(now, "srv9", "Host", "problem", 2, "no route"),
|
|
Event(now, "checkbot", "Checkmk-Verbindung", KIND_SYSTEM, 2, "Checkmk nicht erreichbar seit 14:05"),
|
|
]
|
|
ids = runtime.alerts.insert_alerts(events, "g")
|
|
login_ok(client, VIEWER)
|
|
page = client.get("/alerts").text
|
|
for alert_id in ids:
|
|
assert f'id="alert-{alert_id}"' in page
|
|
for label in ("unterdrückt (bestätigt)", "unterdrückt (Host down)", "unterdrückt (im Sammelhinweis)",
|
|
"Checkmk nicht erreichbar seit 14:05", "system"):
|
|
assert label in page, label
|
|
assert ">DOWN<" in page
|