bvandeusenandClaude Opus 5.5
a6c24299a2
#5059 "Alert state machine: acknowledgement holds until resolve; N consecutive OK to resolve; editing, disabling or deleting a rule resets it". - The state machine is now next_step(), a pure function of one sample. An acknowledged alert stays acknowledged while breached; before, the next breaching sample re-fired it. Firing and acknowledged alerts resolve after consecutive_ok_required OK checks in a row (new rule column, default 1; rule form, rules list and MCP create/update_alert_rule clear_after_checks). - Changing what a rule watches (source, resource, metric, operator, threshold) or disabling it resets its alert to inactive; a firing or acknowledged one records a resolved event whose new note says why, and nobody is notified. Web and MCP edits share alerts.service.apply_rule_change. - Migration 0028: consecutive_ok_required, consecutive_ok_count, alert_events.note, and metric_value nullable (a reset has no sample). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
58 lines
2.6 KiB
Python
58 lines
2.6 KiB
Python
"""The alert state machine, one sample at a time (core/alerts.next_step)."""
|
|
import pytest
|
|
|
|
from steward.core.alerts import Step, next_step
|
|
from steward.models.alerts import AlertStateEnum as S
|
|
|
|
# (state, fail, ok, breached, fail_required, ok_required) -> Step
|
|
CASES = [
|
|
# inactive
|
|
((S.inactive, 0, 0, False, 1, 1), Step(S.inactive, 0, 0)),
|
|
((S.inactive, 0, 0, True, 1, 1), Step(S.firing, 1, 0, "firing")),
|
|
((S.inactive, 0, 0, True, 3, 1), Step(S.pending, 1, 0)),
|
|
# pending
|
|
((S.pending, 1, 0, True, 3, 1), Step(S.pending, 2, 0)),
|
|
((S.pending, 2, 0, True, 3, 1), Step(S.firing, 3, 0, "firing")),
|
|
((S.pending, 2, 0, False, 3, 1), Step(S.inactive, 0, 0)),
|
|
# firing
|
|
((S.firing, 1, 0, True, 1, 1), Step(S.firing, 2, 0)),
|
|
((S.firing, 1, 0, False, 1, 1), Step(S.inactive, 0, 0, "resolved")),
|
|
((S.firing, 1, 0, False, 1, 3), Step(S.firing, 1, 1)),
|
|
((S.firing, 1, 2, False, 1, 3), Step(S.inactive, 0, 0, "resolved")),
|
|
((S.firing, 1, 2, True, 1, 3), Step(S.firing, 2, 0)), # a breach resets the OK run
|
|
# acknowledged: holds while breached, however long
|
|
((S.acknowledged, 1, 0, True, 1, 1), Step(S.acknowledged, 2, 0)),
|
|
((S.acknowledged, 50, 0, True, 2, 1), Step(S.acknowledged, 51, 0)),
|
|
((S.acknowledged, 5, 0, False, 1, 1), Step(S.inactive, 0, 0, "resolved")),
|
|
((S.acknowledged, 5, 0, False, 1, 2), Step(S.acknowledged, 5, 1)),
|
|
((S.acknowledged, 5, 1, True, 1, 2), Step(S.acknowledged, 6, 0)),
|
|
((S.acknowledged, 5, 1, False, 1, 2), Step(S.inactive, 0, 0, "resolved")),
|
|
# the transient resolved state reads as inactive
|
|
((S.resolved, 0, 0, True, 2, 1), Step(S.pending, 1, 0)),
|
|
((S.resolved, 0, 0, False, 1, 1), Step(S.inactive, 0, 0)),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("args,want", CASES)
|
|
def test_next_step(args, want):
|
|
assert next_step(*args) == want
|
|
|
|
|
|
def test_every_state_and_outcome_is_covered():
|
|
seen = {(a[0], a[3]) for a, _ in CASES}
|
|
assert seen == {(s, b) for s in S for b in (True, False)}
|
|
assert {w.event for _, w in CASES} == {None, "firing", "resolved"}
|
|
|
|
|
|
def test_an_acknowledged_alert_never_fires_again_while_breached():
|
|
state, fail, ok = S.acknowledged, 1, 0
|
|
for _ in range(100):
|
|
step = next_step(state, fail, ok, True, 1, 1)
|
|
assert step.state == S.acknowledged and step.event is None
|
|
state, fail, ok = step.state, step.fail_count, step.ok_count
|
|
|
|
|
|
def test_requirements_below_one_count_as_one():
|
|
assert next_step(S.inactive, 0, 0, True, 0, 0) == Step(S.firing, 1, 0, "firing")
|
|
assert next_step(S.firing, 1, 0, False, 0, 0) == Step(S.inactive, 0, 0, "resolved")
|