File size: 1,468 Bytes
7e9a520 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 | """Tests for incident correlator and dedup behavior."""
from agents.correlator import IncidentCorrelator
def _alert(alertname: str, namespace: str = "default", service: str = "frontend") -> dict:
return {
"commonLabels": {"alertname": alertname, "namespace": namespace, "service": service},
"alerts": [{"labels": {"alertname": alertname, "service": service}}],
}
def test_ingest_creates_new_incident():
c = IncidentCorrelator()
incident_id, is_new, should_dispatch = c.ingest(_alert("HighCPU"))
assert incident_id.startswith("inc-")
assert is_new is True
assert should_dispatch is True
def test_ingest_correlates_similar_alerts():
c = IncidentCorrelator()
inc1, _, _ = c.ingest(_alert("HighCPU", service="frontend"))
inc2, is_new, _ = c.ingest(_alert("HighLatency", service="frontend"))
assert inc1 == inc2
assert is_new is False
def test_dedup_skips_dispatch_while_processing():
c = IncidentCorrelator()
inc, _, _ = c.ingest(_alert("HighCPU", service="frontend"))
c.mark_processing(inc, True)
_inc2, _is_new, should_dispatch = c.ingest(_alert("High5xx", service="frontend"))
assert should_dispatch is False
def test_different_namespace_does_not_correlate():
c = IncidentCorrelator()
inc1, _, _ = c.ingest(_alert("HighCPU", namespace="prod-a"))
inc2, is_new, _ = c.ingest(_alert("HighCPU", namespace="prod-b"))
assert inc1 != inc2
assert is_new is True
|