39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
|
"""Recovery detection must emit one notification per metric, not per Redis entry."""
|
||
|
|
|
||
|
|
from server.api.agent import _alert_metric_key, _detect_recovery
|
||
|
|
|
||
|
|
|
||
|
|
def test_alert_metric_key_parses_legacy_and_new_formats():
|
||
|
|
assert _alert_metric_key("mem") == "mem"
|
||
|
|
assert _alert_metric_key("mem:92.1") == "mem"
|
||
|
|
assert _alert_metric_key("cpu:85.0") == "cpu"
|
||
|
|
assert _alert_metric_key("") is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_detect_recovery_dedupes_legacy_metric_value_entries():
|
||
|
|
thresholds = {"cpu": 80, "mem": 80, "disk": 80}
|
||
|
|
prev_alerts = {"mem:88.0", "mem:91.2", "mem:93.5"}
|
||
|
|
system_info = {"mem_usage": 55.0}
|
||
|
|
|
||
|
|
recoveries = _detect_recovery(system_info, prev_alerts, thresholds)
|
||
|
|
|
||
|
|
assert recoveries == [("mem", 55.0)]
|
||
|
|
|
||
|
|
|
||
|
|
def test_detect_recovery_one_per_metric_when_multiple_metrics():
|
||
|
|
thresholds = {"cpu": 80, "mem": 80, "disk": 80}
|
||
|
|
prev_alerts = {"cpu:90.0", "mem:88.0", "mem:92.0"}
|
||
|
|
system_info = {"cpu_usage": 40.0, "mem_usage": 50.0}
|
||
|
|
|
||
|
|
recoveries = _detect_recovery(system_info, prev_alerts, thresholds)
|
||
|
|
|
||
|
|
assert sorted(recoveries) == [("cpu", 40.0), ("mem", 50.0)]
|
||
|
|
|
||
|
|
|
||
|
|
def test_detect_recovery_skips_still_alerting_metric():
|
||
|
|
thresholds = {"cpu": 80, "mem": 80, "disk": 80}
|
||
|
|
prev_alerts = {"mem:88.0", "mem:91.0"}
|
||
|
|
system_info = {"mem_usage": 85.0}
|
||
|
|
|
||
|
|
assert _detect_recovery(system_info, prev_alerts, thresholds) == []
|