""" Tests for incident_intelligence.py — detect_recurrence function. Builds a controlled MemoryIncidentStore dataset with known timestamps/signatures. """ import sys import os import datetime import pytest sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "services", "router")) def _ts(days_ago: float = 0.0) -> str: return (datetime.datetime.utcnow() - datetime.timedelta(days=days_ago)).isoformat() def _make_inc(store, service, kind_tag, sig=None, days_ago=0.0, status="open", severity="P2"): meta = {} if sig: meta["incident_signature"] = sig if kind_tag: meta["kind"] = kind_tag inc = store.create_incident({ "service": service, "env": "prod", "severity": severity, "title": f"{kind_tag} on {service}", "started_at": _ts(days_ago), "created_by": "test", "meta": meta, }) if status == "closed": store.close_incident(inc["id"], _ts(days_ago - 0.01), "resolved") return inc @pytest.fixture def store(): from incident_store import MemoryIncidentStore return MemoryIncidentStore() @pytest.fixture def policy(): import incident_intelligence incident_intelligence._POLICY_CACHE = None return { "correlation": {"lookback_days": 30, "max_related": 10, "min_score": 20, "rules": []}, "recurrence": { "windows_days": [7, 30], "thresholds": { "signature": {"warn": 2, "high": 4}, "kind": {"warn": 3, "high": 6}, }, "top_n": 15, "recommendations": { "signature_high": "Fix signature {sig}", "signature_warn": "Review signature {sig}", "kind_high": "Fix kind {kind}", "kind_warn": "Review kind {kind}", }, }, "digest": {"markdown_max_chars": 8000, "top_incidents": 20, "output_dir": "/tmp/test_intel_reports", "include_closed": True, "include_open": True}, } # ─── Tests ──────────────────────────────────────────────────────────────────── class TestDetectRecurrence: def test_counts_correct_for_7d(self, store, policy): from incident_intelligence import detect_recurrence SIG = "aaabbbccc111222" # 5 incidents in last 7d with same sig for i in range(5): _make_inc(store, "gateway", "error_rate", sig=SIG, days_ago=float(i) / 7.0) # 1 incident older than 7d — should NOT be counted in 7d window _make_inc(store, "gateway", "error_rate", sig=SIG, days_ago=8.0) stats = detect_recurrence(window_days=7, policy=policy, store=store) assert stats["window_days"] == 7 assert stats["total_incidents"] == 5 sigs = {s["signature"]: s for s in stats["top_signatures"]} assert SIG in sigs assert sigs[SIG]["count"] == 5 def test_counts_correct_for_30d(self, store, policy): from incident_intelligence import detect_recurrence SIG = "sig30dtest00001111" for i in range(10): _make_inc(store, "router", "latency", sig=SIG, days_ago=float(i) * 2.5) # 1 older than 30d _make_inc(store, "router", "latency", sig=SIG, days_ago=31.0) stats = detect_recurrence(window_days=30, policy=policy, store=store) sigs = {s["signature"]: s for s in stats["top_signatures"]} assert SIG in sigs assert sigs[SIG]["count"] == 10 def test_threshold_classify_warn(self, store, policy): from incident_intelligence import detect_recurrence SIG = "warnsigaabb1122" # 3 incidents → should hit warn threshold (warn=2) for i in range(3): _make_inc(store, "svc", "latency", sig=SIG, days_ago=float(i) * 0.5) stats = detect_recurrence(window_days=7, policy=policy, store=store) warn_sigs = [s["signature"] for s in stats["warn_recurrence"]["signatures"]] high_sigs = [s["signature"] for s in stats["high_recurrence"]["signatures"]] assert SIG in warn_sigs, "3 incidents should appear in warn (warn_threshold=2, high=4)" assert SIG not in high_sigs, "3 < 4 should NOT be in high" def test_threshold_classify_high(self, store, policy): from incident_intelligence import detect_recurrence SIG = "highsig11223344556677" # 5 incidents → should hit high threshold (high=4) for i in range(5): _make_inc(store, "svc", "latency", sig=SIG, days_ago=float(i) * 0.3) stats = detect_recurrence(window_days=7, policy=policy, store=store) high_sigs = [s["signature"] for s in stats["high_recurrence"]["signatures"]] assert SIG in high_sigs, "5 incidents >= 4 → high" def test_kind_frequency_counted(self, store, policy): from incident_intelligence import detect_recurrence for i in range(6): _make_inc(store, f"svc_{i}", "error_rate", days_ago=float(i) * 0.5) stats = detect_recurrence(window_days=7, policy=policy, store=store) kinds = {k["kind"]: k for k in stats["top_kinds"]} assert "error_rate" in kinds assert kinds["error_rate"]["count"] == 6 def test_kind_threshold_high(self, store, policy): from incident_intelligence import detect_recurrence for i in range(7): _make_inc(store, f"svc_{i}", "latency", days_ago=float(i) * 0.5) stats = detect_recurrence(window_days=7, policy=policy, store=store) high_kinds = [k["kind"] for k in stats["high_recurrence"]["kinds"]] assert "latency" in high_kinds, "7 >= 6 → high kind" def test_services_cross_counted(self, store, policy): from incident_intelligence import detect_recurrence SIG = "crosssvc1234abcd" for svc in ["gateway", "router", "sofiia"]: _make_inc(store, svc, "oom", sig=SIG, days_ago=1.0) stats = detect_recurrence(window_days=7, policy=policy, store=store) sigs = {s["signature"]: s for s in stats["top_signatures"]} if SIG in sigs: services = set(sigs[SIG].get("services", [])) assert "gateway" in services and "router" in services def test_open_closed_counts(self, store, policy): from incident_intelligence import detect_recurrence _make_inc(store, "svc", "latency", days_ago=1.0, status="open") _make_inc(store, "svc", "latency", days_ago=1.5, status="closed") _make_inc(store, "svc", "latency", days_ago=2.0, status="open") stats = detect_recurrence(window_days=7, policy=policy, store=store) assert stats["open_count"] == 2 assert stats["closed_count"] == 1 assert stats["total_incidents"] == 3 def test_empty_store(self, store, policy): from incident_intelligence import detect_recurrence stats = detect_recurrence(window_days=7, policy=policy, store=store) assert stats["total_incidents"] == 0 assert stats["top_signatures"] == [] assert stats["top_kinds"] == [] def test_top_n_enforced(self, store, policy): from incident_intelligence import detect_recurrence # Create 20 different signatures for i in range(20): _make_inc(store, "svc", "latency", sig=f"sig{i:032d}", days_ago=float(i) * 0.2) policy["recurrence"]["top_n"] = 5 stats = detect_recurrence(window_days=7, policy=policy, store=store) assert len(stats["top_signatures"]) <= 5