"""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