130 lines
6.1 KiB
Python
130 lines
6.1 KiB
Python
import time
|
|
from datetime import timedelta
|
|
|
|
from tests.test_web import ADMIN, VIEWER, admin, app, client, csrf_from, login_ok, post, users, viewer # noqa: F401
|
|
from tests.test_web_routing import enable_telegram, route_form, routes
|
|
|
|
HINT = "Geheimer Hinweis zum Pool"
|
|
|
|
|
|
def rule_form(**overrides):
|
|
data = route_form(delay_minutes="0", output_contains="degraded", ai_enabled="1", ai_hint=HINT,
|
|
reminder_hours="4", loud="1", use_default_targets="1")
|
|
data.update(overrides)
|
|
return data
|
|
|
|
|
|
def test_page_is_called_rules_and_shows_new_fields(client, admin):
|
|
login_ok(client)
|
|
page = client.get("/routing")
|
|
assert "<h1>Regeln</h1>" in page.text and ">Regeln</a>" in page.text
|
|
for label in ("Verzögerung in Minuten", "Ausgabe enthält", "KI-Analyse für diese Alarme", "KI-Hinweis",
|
|
"Erinnerung alle N Stunden", "Laut", "Standardziele aller aktivierten Backends"):
|
|
assert label in page.text, label
|
|
|
|
|
|
def test_create_stores_all_new_fields(client, app, admin):
|
|
enable_telegram(app)
|
|
login_ok(client)
|
|
assert post(client, "/routing", "/routing", **rule_form()).status_code == 303
|
|
rule = routes(app)[0]
|
|
assert rule.delay_minutes == 0 and rule.output_contains == "degraded" and rule.ai_enabled
|
|
assert rule.ai_hint == HINT and rule.reminder_hours == 4 and rule.loud and rule.use_default_targets
|
|
|
|
|
|
def test_unchecked_boxes_and_empty_numbers_mean_off_and_default(client, app, admin):
|
|
login_ok(client)
|
|
form = rule_form(delay_minutes="", reminder_hours="", output_contains="", ai_hint="")
|
|
for flag in ("ai_enabled", "loud", "use_default_targets"):
|
|
form.pop(flag)
|
|
post(client, "/routing", "/routing", **form)
|
|
rule = routes(app)[0]
|
|
assert rule.delay_minutes is None and rule.reminder_hours is None
|
|
assert not rule.ai_enabled and not rule.loud and not rule.use_default_targets
|
|
|
|
|
|
def test_edit_page_prefills_values(client, app, admin):
|
|
login_ok(client)
|
|
post(client, "/routing", "/routing", **rule_form())
|
|
page = client.get(f"/routing/{routes(app)[0].id}")
|
|
assert 'name="delay_minutes"' in page.text and 'value="0"' in page.text
|
|
assert HINT in page.text and 'value="degraded"' in page.text and 'value="4"' in page.text
|
|
|
|
|
|
def test_new_fields_are_validated(client, app, admin):
|
|
login_ok(client)
|
|
for override in ({"delay_minutes": "-1"}, {"delay_minutes": "1441"}, {"delay_minutes": "abc"},
|
|
{"delay_minutes": "1.5"}, {"reminder_hours": "169"}, {"reminder_hours": "-2"},
|
|
{"output_contains": "x" * 201}, {"ai_hint": "y" * 801}, {"ai_hint": "a\x00b"}):
|
|
response = post(client, "/routing", "/routing", **rule_form(**override))
|
|
assert response.status_code == 422, override
|
|
assert routes(app) == []
|
|
|
|
|
|
def test_boundary_values_accepted(client, app, admin):
|
|
login_ok(client)
|
|
form = rule_form(delay_minutes="1440", reminder_hours="168", output_contains="x" * 200, ai_hint="y" * 800)
|
|
assert post(client, "/routing", "/routing", **form).status_code == 303
|
|
|
|
|
|
def test_audit_has_old_and_new_values_but_never_the_hint_text(client, app, admin):
|
|
login_ok(client)
|
|
post(client, "/routing", "/routing", **rule_form())
|
|
rule = routes(app)[0]
|
|
post(client, f"/routing/{rule.id}", f"/routing/{rule.id}", **rule_form(delay_minutes="5", ai_hint="Neuer geheimer Text"))
|
|
entries = app.state.db.query("SELECT action, old_value, new_value FROM audit WHERE action LIKE 'route.%'")
|
|
update = next(e for e in entries if e["action"] == "route.update")
|
|
assert "delay=0" in update["old_value"] and "delay=5" in update["new_value"]
|
|
assert f"hint_len={len(HINT)}" in update["old_value"] and "hint_len=19" in update["new_value"]
|
|
assert all(HINT not in str(e[k]) and "Neuer geheimer" not in str(e[k]) for e in entries
|
|
for k in ("old_value", "new_value"))
|
|
|
|
|
|
def test_alerting_section_shows_new_fields_with_defaults(client, admin):
|
|
login_ok(client)
|
|
page = client.get("/settings/alerting")
|
|
assert "Standard-Verzögerung in Minuten" in page.text and 'value="15"' in page.text
|
|
assert "Recovery-Meldungen senden" in page.text and "checked" in page.text
|
|
assert "Mindestdauer" not in page.text
|
|
|
|
|
|
def test_alerting_section_saves_recovery_switch_and_reminder(client, app, admin):
|
|
login_ok(client)
|
|
form = {"poll_interval": "30", "alerts_default_delay_minutes": "0", "alerts_reminder_hours": "12",
|
|
"alerts_bundle_seconds": "60", "alerts_ai_per_minute": "10"}
|
|
assert post(client, "/settings/alerting", "/settings/alerting", **form).status_code == 303
|
|
store = app.state.store
|
|
assert store.get("alerts.recovery_enabled") == "0"
|
|
assert store.get_int("alerts.default_delay_minutes", 15) == 0
|
|
assert store.get_int("alerts.reminder_hours", 0) == 12
|
|
post(client, "/settings/alerting", "/settings/alerting", alerts_recovery_enabled="1", **form)
|
|
assert store.get("alerts.recovery_enabled") == "1"
|
|
|
|
|
|
def add_state(app, service, state=2, reported=0, age=60, downtime=0):
|
|
app.state.db.execute(
|
|
"INSERT INTO service_states(host, service, state, output, last_state_change, in_downtime, "
|
|
"reported_state) VALUES ('web01', ?, ?, 'x', ?, ?, ?)",
|
|
(service, state, int(time.time()) - age, downtime, reported))
|
|
|
|
|
|
def test_dashboard_shows_reporting_status(client, app, viewer):
|
|
add_state(app, "wartend", age=60)
|
|
add_state(app, "gemeldet", reported=2, age=7200)
|
|
add_state(app, "downtime", downtime=1, age=7200)
|
|
add_state(app, "stumm", age=7200)
|
|
app.state.runtime.mutes.create("web01", "stumm", timedelta(hours=1), "test", "admin", "ui")
|
|
login_ok(client, VIEWER)
|
|
text = client.get("/").text
|
|
assert "wartet noch 14 min" in text
|
|
assert ">gemeldet<" in text
|
|
assert "unterdrückt (Downtime)" in text and "unterdrückt (Stummschaltung)" in text
|
|
|
|
|
|
def test_dashboard_waiting_time_uses_rule_delay(client, app, admin):
|
|
login_ok(client)
|
|
post(client, "/routing", "/routing", **rule_form(delay_minutes="30", match_host="web*", output_contains="",
|
|
min_state="1"))
|
|
add_state(app, "wartend", age=60)
|
|
assert "wartet noch 29 min" in client.get("/").text
|