checkbot/tests/test_rule_decisions.py

from app.chat import Target
from app.models import KIND_FLAPPING, KIND_PROBLEM, KIND_RECOVERY, KIND_REMINDER, Event
from app.routing import (ALL_KINDS, Route, ai_allowed, ai_hints, delay_provider, resolve_reminder_hours,
                         resolve_delay, resolve_loud, resolve_targets, route_matches)

TG = Target("telegram", "1")
MX = Target("matrix", "!r:s")
DEFAULTS = [TG, MX]


def rule(position=1, **overrides):
    fields = dict(id=position, name=f"r{position}", position=position, enabled=True, match_host="",
                  match_service="", min_state=1, kinds=ALL_KINDS, stop_on_match=False, targets=())
    fields.update(overrides)
    return Route(**fields)


def event(kind=KIND_PROBLEM, state=2, output="boom", host="web01", service="disk"):
    return Event(ts=1, host=host, service=service, kind=kind, state=state, output=output)


def test_output_contains_is_case_insensitive_substring():
    r = rule(output_contains="DEGRADED")
    assert route_matches(r, event(output="pool is degraded now"))
    assert not route_matches(r, event(output="pool online"))


def test_output_contains_is_not_a_regex():
    r = rule(output_contains="a.c")
    assert route_matches(r, event(output="xa.cx"))
    assert not route_matches(r, event(output="abc"))


def test_empty_output_filter_matches_everything():
    assert route_matches(rule(), event(output=""))


def test_reminder_follows_problem_rules():
    only_problem = rule(kinds=(KIND_PROBLEM,), min_state=2)
    assert route_matches(only_problem, event(kind=KIND_REMINDER, state=2))
    assert not route_matches(only_problem, event(kind=KIND_REMINDER, state=1))
    assert not route_matches(rule(kinds=(KIND_RECOVERY,)), event(kind=KIND_REMINDER))


def test_kinds_and_min_state_still_apply():
    r = rule(min_state=2, kinds=(KIND_PROBLEM, KIND_FLAPPING))
    assert not route_matches(r, event(state=1))
    assert not route_matches(r, event(kind=KIND_RECOVERY, state=0))
    assert route_matches(r, event(kind=KIND_FLAPPING, state=1))


def test_delay_without_match_uses_default():
    assert resolve_delay([], "h", "s", 2, "", 15) == 15
    assert resolve_delay([rule(match_host="db*", delay_minutes=1)], "web01", "s", 2, "", 15) == 15


def test_delay_rule_without_value_uses_default():
    assert resolve_delay([rule()], "h", "s", 2, "", 15) == 15


def test_delay_smallest_wins_and_zero_overrides_default():
    rules = [rule(1, delay_minutes=10), rule(2, delay_minutes=0), rule(3, delay_minutes=5)]
    assert resolve_delay(rules, "h", "s", 2, "", 15) == 0
    assert resolve_delay([rule(1, delay_minutes=10), rule(2, delay_minutes=5)], "h", "s", 2, "", 15) == 5


def test_delay_respects_service_glob_state_and_output():
    r = rule(match_service="*zpool*", min_state=1, output_contains="error", delay_minutes=0)
    assert resolve_delay([r], "h", "ZPool status", 1, "Some Error", 15) == 0
    assert resolve_delay([r], "h", "cpu", 1, "Some Error", 15) == 15
    assert resolve_delay([r], "h", "zpool", 1, "fine", 15) == 15
    assert resolve_delay([rule(min_state=2, delay_minutes=0)], "h", "s", 1, "", 15) == 15


def test_delay_ignores_disabled_rules_and_honours_stop():
    assert resolve_delay([rule(enabled=False, delay_minutes=0)], "h", "s", 2, "", 15) == 15
    rules = [rule(1, delay_minutes=7, stop_on_match=True), rule(2, delay_minutes=0)]
    assert resolve_delay(rules, "h", "s", 2, "", 15) == 7


def test_delay_provider_returns_seconds():
    provider = delay_provider([rule(match_service="zpool*", delay_minutes=0)], 15)
    assert provider("h", "zpool status", 1, "") == 0
    assert provider("h", "cpu", 1, "") == 900


def test_ai_allowed_off_as_soon_as_one_matching_rule_disables():
    assert ai_allowed([], event())
    assert ai_allowed([rule(1), rule(2, match_host="db*", ai_enabled=False)], event())
    assert not ai_allowed([rule(1), rule(2, ai_enabled=False)], event())


def test_resolve_loud():
    assert resolve_loud([], event())
    assert not resolve_loud([rule(loud=False)], event())
    assert resolve_loud([rule(1, loud=False), rule(2, loud=True)], event())
    assert resolve_loud([rule(1, loud=False, match_host="db*")], event())


def test_ai_hints_in_rule_order_and_only_matching():
    rules = [rule(2, ai_hint="zweiter"), rule(1, ai_hint="erster"), rule(3, ai_hint="x", match_host="db*"),
             rule(4, ai_hint="  ")]
    assert ai_hints(rules, event()) == "erster\nzweiter"
    assert ai_hints([], event()) == ""


def test_ai_hints_clipped_to_1200_characters():
    rules = [rule(1, ai_hint="a" * 800), rule(2, ai_hint="b" * 800)]
    text = ai_hints(rules, event())
    assert len(text) == 1200 and text.startswith("a" * 800) and text.endswith("b" * 399)


def test_reminder_hours_null_uses_default_zero_turns_off_positive_takes_smallest():
    assert resolve_reminder_hours([], event(), 3) == 3
    assert resolve_reminder_hours([rule(1, reminder_hours=None)], event(), 3) == 3
    assert resolve_reminder_hours([rule(1, reminder_hours=0)], event(), 3) == 0
    assert resolve_reminder_hours([rule(1, reminder_hours=8), rule(2, reminder_hours=4), rule(3)], event(), 3) == 4


def test_reminder_hours_explicit_zero_beats_other_matching_rules():
    rules = [rule(1, reminder_hours=0), rule(2, reminder_hours=4), rule(3, reminder_hours=None)]
    assert resolve_reminder_hours(rules, event(), 6) == 0


def test_reminder_hours_ignore_non_matching_rules():
    rules = [rule(1, reminder_hours=0, match_host="db*"), rule(2, reminder_hours=5)]
    assert resolve_reminder_hours(rules, event(), 6) == 5


def test_use_default_targets_adds_defaults_to_explicit_targets():
    other = Target("telegram", "9")
    assert resolve_targets([rule(use_default_targets=True)], event(), DEFAULTS) == DEFAULTS
    assert resolve_targets([rule(targets=(other,), use_default_targets=True)], event(), DEFAULTS) == [other, TG, MX]
    assert resolve_targets([rule(targets=(other,))], event(), DEFAULTS) == [other]
    assert resolve_targets([rule()], event(), DEFAULTS) == []