86 lines
3.5 KiB
Python
86 lines
3.5 KiB
Python
from datetime import timedelta
|
|
|
|
import pytest
|
|
|
|
from app.alerts import AlertRepository
|
|
from app.db import Database
|
|
from app.models import KIND_PROBLEM, KIND_RECOVERY, SUPPRESS_DOWNTIME, SUPPRESS_MUTE, Event
|
|
from app.mutes import SOURCE_CHAT, SOURCE_UI, Mute, MuteError, MuteService, apply_mutes, find_active_mute
|
|
|
|
NOW = 10_000
|
|
|
|
|
|
def mute(host="web*", service="*", expires=NOW + 100, id=1):
|
|
return Mute(id, host, service, expires, "", "admin", NOW - 10, SOURCE_UI)
|
|
|
|
|
|
def event(host="web01", service="disk", kind=KIND_PROBLEM):
|
|
return Event(ts=NOW, host=host, service=service, kind=kind, state=2)
|
|
|
|
|
|
def test_find_active_mute_glob_match():
|
|
assert find_active_mute([mute()], "web01", "disk", NOW)
|
|
assert find_active_mute([mute(host="web01", service="Dis*")], "WEB01", "disk", NOW)
|
|
assert find_active_mute([mute()], "db01", "disk", NOW) is None
|
|
assert find_active_mute([mute(service="cpu")], "web01", "disk", NOW) is None
|
|
|
|
|
|
def test_expired_mute_ignored():
|
|
assert find_active_mute([mute(expires=NOW)], "web01", "disk", NOW) is None
|
|
assert find_active_mute([mute(expires=NOW - 1)], "web01", "disk", NOW) is None
|
|
|
|
|
|
def test_apply_mutes_suppresses_problem_and_recovery():
|
|
result = apply_mutes([event(), event(kind=KIND_RECOVERY), event(host="db01")], [mute()], NOW)
|
|
assert [e.suppressed for e in result] == [True, True, False]
|
|
assert result[0].suppress_reason == SUPPRESS_MUTE
|
|
|
|
|
|
def test_apply_mutes_keeps_downtime_reason():
|
|
downtime = Event(ts=NOW, host="web01", service="disk", kind=KIND_PROBLEM, state=2, suppressed=True,
|
|
suppress_reason=SUPPRESS_DOWNTIME)
|
|
assert apply_mutes([downtime], [mute()], NOW)[0].suppress_reason == SUPPRESS_DOWNTIME
|
|
|
|
|
|
@pytest.fixture
|
|
def service():
|
|
clock = {"now": float(NOW)}
|
|
svc = MuteService(Database(":memory:"), clock=lambda: clock["now"])
|
|
svc.clock = clock
|
|
return svc
|
|
|
|
|
|
def test_service_create_list_delete_and_audit(service):
|
|
created = service.create("web*", "", timedelta(hours=2), "Wartung", "admin", SOURCE_CHAT)
|
|
assert created.service_glob == "*" and created.expires_at == NOW + 7200 and created.source == SOURCE_CHAT
|
|
assert [m.id for m in service.active()] == [created.id]
|
|
assert service.delete(created.id, "admin") is True
|
|
assert service.delete(created.id, "admin") is False
|
|
actions = [r["action"] for r in service._db.query("SELECT action FROM audit ORDER BY id")]
|
|
assert actions == ["mute.create", "mute.delete"]
|
|
|
|
|
|
def test_service_validation(service):
|
|
for host, duration, source in (("", timedelta(hours=1), SOURCE_UI),
|
|
("web", timedelta(seconds=10), SOURCE_UI),
|
|
("web", timedelta(days=365), SOURCE_UI),
|
|
("web", timedelta(hours=1), "mail")):
|
|
with pytest.raises(MuteError):
|
|
service.create(host, "", duration, "", "admin", source)
|
|
|
|
|
|
def test_expired_mutes_not_listed_and_purged(service):
|
|
service.create("web*", "", timedelta(minutes=30), "", "admin", SOURCE_UI)
|
|
service.clock["now"] += 1801
|
|
assert service.active() == []
|
|
assert service.purge_expired() == 1
|
|
assert service._db.query("SELECT COUNT(*) AS n FROM mutes")[0]["n"] == 0
|
|
|
|
|
|
def test_muted_event_is_stored_with_reason_mute():
|
|
db = Database(":memory:")
|
|
repo = AlertRepository(db)
|
|
repo.insert_alerts(apply_mutes([event()], [mute()], NOW), "g")
|
|
row = repo.recent()[0]
|
|
assert row["suppressed"] == 1 and row["suppress_reason"] == "mute" and row["ai_status"] == "skipped"
|