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>
194 lines
8.1 KiB
Python
194 lines
8.1 KiB
Python
from dataclasses import replace
|
|
|
|
from app.checkmk import HostState, ServiceState
|
|
from app.models import (HOST_SERVICE_NAME, KIND_PROBLEM, KIND_RECOVERY, SUPPRESS_ACK, SUPPRESS_DOWNTIME,
|
|
SUPPRESS_HOST_DOWN)
|
|
from app.poller import compute_events
|
|
from app.routing import ALL_KINDS, Route, delay_provider, matching_rules, route_matches
|
|
|
|
NOW = 100_000
|
|
MINUTE = 60
|
|
|
|
|
|
def svc(state=2, host="h1", service="cpu", changed=NOW - 3600, downtime=False, ack=False, output="boom"):
|
|
return ServiceState(host, service, state, True, output, changed, downtime, False, ack)
|
|
|
|
|
|
def host(state=2, name="h1", changed=NOW - 3600, downtime=False, ack=False, output="Host down"):
|
|
return HostState(name, state, output, changed, downtime, ack).as_service()
|
|
|
|
|
|
def make_run(provider=None):
|
|
def run(previous, current, now=NOW):
|
|
events, rows = compute_events(previous, current, now, provider or (lambda *_: 0))
|
|
return events, {row.key: row for row in rows}
|
|
return run
|
|
|
|
|
|
run = make_run()
|
|
|
|
|
|
def test_acknowledged_problem_is_suppressed_with_reason_and_marked_reported():
|
|
events, rows = run({}, [svc(ack=True)])
|
|
assert [(e.kind, e.suppressed, e.suppress_reason) for e in events] == [(KIND_PROBLEM, True, SUPPRESS_ACK)]
|
|
row = rows[("h1", "cpu")]
|
|
assert row.reported_state == 2 and row.ack_state == 2
|
|
|
|
|
|
def test_ack_on_already_reported_problem_stays_silent_and_remembers_state():
|
|
_, rows = run({}, [svc()])
|
|
events, rows = run(rows, [svc(ack=True)], now=NOW + 30)
|
|
assert events == [] and rows[("h1", "cpu")].ack_state == 2
|
|
|
|
|
|
def test_worsening_after_ack_is_reported_again():
|
|
_, rows = run({}, [svc(state=1, ack=True)])
|
|
assert rows[("h1", "cpu")].ack_state == 1
|
|
events, rows = run(rows, [svc(state=2, ack=True, changed=NOW + 30)], now=NOW + 30)
|
|
assert [(e.state, e.suppressed) for e in events] == [(2, False)]
|
|
assert events[0].previous_state == 1 and rows[("h1", "cpu")].ack_state == 1
|
|
|
|
|
|
def test_unknown_is_less_severe_than_crit_but_worse_than_warn():
|
|
_, rows = run({}, [svc(state=2, ack=True)])
|
|
events, _ = run(rows, [svc(state=3, ack=True, changed=NOW + 30)], now=NOW + 30)
|
|
assert [(e.state, e.suppress_reason) for e in events] == [(3, SUPPRESS_ACK)]
|
|
_, rows = run({}, [svc(state=1, ack=True)])
|
|
events, _ = run(rows, [svc(state=3, ack=True, changed=NOW + 30)], now=NOW + 30)
|
|
assert [(e.state, e.suppressed) for e in events] == [(3, False)]
|
|
|
|
|
|
def test_removed_ack_clears_state_and_does_not_repeat_the_problem():
|
|
_, rows = run({}, [svc(ack=True)])
|
|
events, rows = run(rows, [svc()], now=NOW + 30)
|
|
assert events == [] and rows[("h1", "cpu")].ack_state is None
|
|
|
|
|
|
def test_after_removed_ack_a_worsening_is_reported_normally():
|
|
_, rows = run({}, [svc(state=1, ack=True)])
|
|
_, rows = run(rows, [svc(state=1)], now=NOW + 30)
|
|
events, _ = run(rows, [svc(state=2, changed=NOW + 60)], now=NOW + 60)
|
|
assert [(e.state, e.suppressed) for e in events] == [(2, False)]
|
|
|
|
|
|
def test_recovery_of_reported_then_acknowledged_problem_is_not_suppressed():
|
|
_, rows = run({}, [svc()])
|
|
_, rows = run(rows, [svc(ack=True)], now=NOW + 30)
|
|
events, _ = run(rows, [], now=NOW + 60)
|
|
assert [(e.kind, e.suppressed) for e in events] == [(KIND_RECOVERY, False)]
|
|
|
|
|
|
def test_downtime_has_priority_over_ack():
|
|
events, _ = run({}, [svc(ack=True, downtime=True)])
|
|
assert events[0].suppress_reason == SUPPRESS_DOWNTIME
|
|
|
|
|
|
def test_ack_during_delay_wait_suppresses_at_expiry():
|
|
run15 = make_run(delay_provider([], 15))
|
|
_, rows = run15({}, [svc(changed=NOW, ack=True)], NOW + 60)
|
|
events, _ = run15(rows, [svc(changed=NOW, ack=True)], NOW + 900)
|
|
assert [(e.suppressed, e.suppress_reason) for e in events] == [(True, SUPPRESS_ACK)]
|
|
|
|
|
|
def test_host_down_is_an_event_with_host_service_name():
|
|
events, rows = run({}, [host(state=2)])
|
|
assert [(e.service, e.kind, e.state, e.output) for e in events] == [
|
|
(HOST_SERVICE_NAME, KIND_PROBLEM, 2, "Host down")]
|
|
assert rows[("h1", HOST_SERVICE_NAME)].reported_state == 2
|
|
|
|
|
|
def test_host_unreach_uses_state_three():
|
|
events, _ = run({}, [host(state=3)])
|
|
assert events[0].state == 3
|
|
|
|
|
|
def test_host_recovery_event_when_host_is_up_again():
|
|
_, rows = run({}, [host()])
|
|
events, new_rows = run(rows, [], now=NOW + 600)
|
|
assert [(e.kind, e.service, e.previous_state) for e in events] == [(KIND_RECOVERY, HOST_SERVICE_NAME, 2)]
|
|
assert events[0].problem_since_ts == NOW - 3600 and new_rows == {}
|
|
|
|
|
|
def test_host_delay_is_separate_and_ignores_rule_delays():
|
|
rule = Route(1, "alle", 1, True, "*", "*", 1, ALL_KINDS, False, (), delay_minutes=0)
|
|
run_h = make_run(delay_provider([rule], 15, host_minutes=5))
|
|
events, rows = run_h({}, [host(changed=NOW), svc(changed=NOW, service="a")], NOW + 4 * MINUTE)
|
|
assert [e.service for e in events] == ["a"] and events[0].suppress_reason == SUPPRESS_HOST_DOWN
|
|
events, _ = run_h(rows, [host(changed=NOW), svc(changed=NOW, service="a")], NOW + 5 * MINUTE)
|
|
assert [e.service for e in events] == [HOST_SERVICE_NAME]
|
|
|
|
|
|
def test_host_acknowledged_and_downtime_work_like_services():
|
|
events, _ = run({}, [host(ack=True)])
|
|
assert events[0].suppress_reason == SUPPRESS_ACK
|
|
events, _ = run({}, [host(downtime=True)])
|
|
assert events[0].suppress_reason == SUPPRESS_DOWNTIME
|
|
|
|
|
|
def test_service_alarm_of_down_host_is_suppressed_and_not_marked_reported():
|
|
events, rows = run({}, [host(), svc()])
|
|
by_service = {e.service: e for e in events}
|
|
assert by_service[HOST_SERVICE_NAME].suppressed is False
|
|
assert by_service["cpu"].suppressed and by_service["cpu"].suppress_reason == SUPPRESS_HOST_DOWN
|
|
assert rows[("h1", "cpu")].reported_state == 0 and rows[("h1", "cpu")].host_held
|
|
|
|
|
|
def test_held_service_alarm_is_emitted_only_once_while_host_stays_down():
|
|
_, rows = run({}, [host(), svc()])
|
|
events, rows = run(rows, [host(), svc()], now=NOW + 30)
|
|
assert events == [] and rows[("h1", "cpu")].host_held
|
|
|
|
|
|
def test_service_alarm_of_other_host_is_not_affected():
|
|
events, _ = run({}, [host(name="h2"), svc()])
|
|
assert next(e for e in events if e.host == "h1").suppressed is False
|
|
|
|
|
|
def test_service_problem_persisting_after_host_returns_is_reported_normally():
|
|
_, rows = run({}, [host(), svc()])
|
|
events, rows = run(rows, [svc()], now=NOW + 600)
|
|
kinds = {(e.service, e.kind, e.suppressed) for e in events}
|
|
assert kinds == {("cpu", KIND_PROBLEM, False), (HOST_SERVICE_NAME, KIND_RECOVERY, False)}
|
|
assert not rows[("h1", "cpu")].host_held and rows[("h1", "cpu")].reported_state == 2
|
|
|
|
|
|
def test_service_that_vanishes_during_host_outage_gives_no_recovery():
|
|
_, rows = run({}, [host(), svc()])
|
|
events, _ = run(rows, [host()], now=NOW + 30)
|
|
assert events == []
|
|
|
|
|
|
def test_service_alarms_are_suppressed_while_host_alarm_is_still_waiting():
|
|
run_h = make_run(delay_provider([], 0, host_minutes=10))
|
|
events, rows = run_h({}, [host(changed=NOW), svc(changed=NOW)], NOW + MINUTE)
|
|
assert [(e.service, e.suppress_reason) for e in events] == [("cpu", SUPPRESS_HOST_DOWN)]
|
|
|
|
|
|
def test_worsening_of_held_service_is_held_again_with_new_state():
|
|
_, rows = run({}, [host(), svc(state=1)])
|
|
events, rows = run(rows, [host(), svc(state=2, changed=NOW + 30)], now=NOW + 30)
|
|
assert [(e.state, e.suppress_reason) for e in events] == [(2, SUPPRESS_HOST_DOWN)]
|
|
|
|
|
|
def test_min_state_roles_distinguish_down_and_unreach():
|
|
def rule(min_state, service=""):
|
|
return Route(1, "r", 1, True, "", service, min_state, ALL_KINDS, False, (), None)
|
|
|
|
down = run({}, [host(state=2)])[0][0]
|
|
unreach = run({}, [host(state=3)])[0][0]
|
|
assert route_matches(rule(1), down) and route_matches(rule(2), down) and not route_matches(rule(3), down)
|
|
assert route_matches(rule(3), unreach)
|
|
|
|
|
|
def test_service_glob_host_matches_only_host_alarms():
|
|
rule = Route(1, "hosts", 1, True, "", "Host", 1, ALL_KINDS, False, ())
|
|
host_event = run({}, [host()])[0][0]
|
|
service_event = run({}, [svc()])[0][0]
|
|
assert matching_rules([rule], host_event) == [rule]
|
|
assert matching_rules([rule], service_event) == []
|
|
|
|
|
|
def test_state_rows_keep_new_fields_roundtrip():
|
|
row = run({}, [svc(ack=True)])[1][("h1", "cpu")]
|
|
assert replace(row, ack_state=None).ack_state is None and row.host_held is False
|