80 lines
2.8 KiB
Python
80 lines
2.8 KiB
Python
from app.chat import Target
|
|
from app.models import KIND_FLAPPING, KIND_PROBLEM, KIND_RECOVERY, Event
|
|
from app.routing import Route, resolve_targets
|
|
|
|
TG = Target("telegram", "1")
|
|
MX = Target("matrix", "!r:s")
|
|
OTHER = Target("telegram", "2")
|
|
DEFAULTS = [TG, MX]
|
|
ALL = (KIND_PROBLEM, KIND_RECOVERY, KIND_FLAPPING)
|
|
|
|
|
|
def event(host="web01", service="disk", kind=KIND_PROBLEM, state=2):
|
|
return Event(ts=1, host=host, service=service, kind=kind, state=state)
|
|
|
|
|
|
def route(position=1, targets=(TG,), host="", service="", min_state=1, kinds=ALL, stop=False,
|
|
enabled=True):
|
|
return Route(position, f"r{position}", position, enabled, host, service, min_state, tuple(kinds), stop,
|
|
tuple(targets))
|
|
|
|
|
|
def test_no_rules_uses_defaults():
|
|
assert resolve_targets([], event(), DEFAULTS) == DEFAULTS
|
|
|
|
|
|
def test_no_match_falls_back_to_defaults():
|
|
assert resolve_targets([route(host="db*")], event(), DEFAULTS) == DEFAULTS
|
|
|
|
|
|
def test_glob_match_host_and_service_case_insensitive():
|
|
rules = [route(host="WEB*", service="Dis?", targets=[OTHER])]
|
|
assert resolve_targets(rules, event(), DEFAULTS) == [OTHER]
|
|
assert resolve_targets(rules, event(service="cpu"), DEFAULTS) == DEFAULTS
|
|
|
|
|
|
def test_min_state_applies_to_problems_only():
|
|
rules = [route(min_state=2, targets=[OTHER])]
|
|
assert resolve_targets(rules, event(state=1), DEFAULTS) == DEFAULTS
|
|
assert resolve_targets(rules, event(state=2), DEFAULTS) == [OTHER]
|
|
assert resolve_targets(rules, event(kind=KIND_RECOVERY, state=0), DEFAULTS) == [OTHER]
|
|
|
|
|
|
def test_kinds_filter():
|
|
rules = [route(kinds=[KIND_PROBLEM], targets=[OTHER])]
|
|
assert resolve_targets(rules, event(kind=KIND_FLAPPING), DEFAULTS) == DEFAULTS
|
|
assert resolve_targets(rules, event(kind=KIND_PROBLEM), DEFAULTS) == [OTHER]
|
|
|
|
|
|
def test_multiple_rules_add_targets_in_order_and_dedup():
|
|
rules = [route(1, [TG]), route(2, [MX, TG]), route(3, [OTHER])]
|
|
assert resolve_targets(rules, event(), DEFAULTS) == [TG, MX, OTHER]
|
|
|
|
|
|
def test_rules_sorted_by_position():
|
|
rules = [route(2, [MX]), route(1, [TG])]
|
|
assert resolve_targets(rules, event(), DEFAULTS) == [TG, MX]
|
|
|
|
|
|
def test_stop_on_match():
|
|
rules = [route(1, [TG], stop=True), route(2, [MX])]
|
|
assert resolve_targets(rules, event(), DEFAULTS) == [TG]
|
|
|
|
|
|
def test_stop_only_when_rule_matches():
|
|
rules = [route(1, [TG], host="db*", stop=True), route(2, [MX])]
|
|
assert resolve_targets(rules, event(), DEFAULTS) == [MX]
|
|
|
|
|
|
def test_disabled_rule_ignored():
|
|
rules = [route(1, [OTHER], enabled=False)]
|
|
assert resolve_targets(rules, event(), DEFAULTS) == DEFAULTS
|
|
|
|
|
|
def test_matching_rule_without_targets_sends_nowhere():
|
|
assert resolve_targets([route(targets=[])], event(), DEFAULTS) == []
|
|
|
|
|
|
def test_defaults_deduplicated():
|
|
assert resolve_targets([], event(), [TG, TG]) == [TG]
|