awoooi/apps/api/tests/test_alert_converged_recurrence.py

import pytest

from src.services import converged_alert_recurrence_notifier as notifier


class _FakeRedis:
    def __init__(self, result):
        self.result = result
        self.calls = []

    async def set(self, key, value, *, ex=None, nx=None):
        self.calls.append({"key": key, "value": value, "ex": ex, "nx": nx})
        return self.result


class _FakeGateway:
    alert_chat_id = "group-chat"

    def __init__(self):
        self.primary_messages = []
        self.private_messages = []

    async def send_alert_notification(self, text):
        self.primary_messages.append(text)
        return {"ok": True}

    async def send_notification(self, text, *, chat_id=None):
        self.private_messages.append({"text": text, "chat_id": chat_id})
        return {"ok": True}


def test_converged_recurrence_message_escapes_html():
    text = notifier.format_converged_alert_recurrence_message(
        source="alertmanager",
        alertname="Disk<Full>",
        severity="critical",
        namespace="prod&ops",
        target_resource="api<script>",
        hit_count=7,
        incident_id="INC-20260611-ABCD",
        approval_id="approval-1",
        alert_category="host_resource",
        notification_type="TYPE-3",
    )

    assert "告警仍在發生" in text
    assert "累計次數：<b>7</b>" in text
    assert "Disk&lt;Full&gt;" in text
    assert "prod&amp;ops" in text
    assert "api&lt;script&gt;" in text
    assert "這不是新的自動修復授權" in text


def test_converged_recurrence_message_supports_llm_inflight_state():
    text = notifier.format_converged_alert_recurrence_message(
        source="alertmanager",
        alertname="PodCrashLoop<api>",
        severity="warning",
        namespace="awoooi-prod",
        target_resource="api",
        hit_count=2,
        incident_id=None,
        approval_id=None,
        alert_category="kubernetes",
        notification_type="TYPE-3",
        recurrence_stage="llm_inflight",
    )

    assert "告警仍在發生：AI 分析中" in text
    assert "背景 AI 分析鎖" in text
    assert "PodCrashLoop&lt;api&gt;" in text
    assert "事件：<code>-</code>" in text
    assert "簽核：<code>-</code>" in text
    assert "這不是新的自動修復授權" in text


@pytest.mark.asyncio
async def test_converged_recurrence_uses_redis_throttle(monkeypatch):
    redis = _FakeRedis(True)
    monkeypatch.setattr(notifier, "get_redis", lambda: redis)

    result = await notifier.should_notify_converged_alert_recurrence(
        fingerprint="abc",
        hit_count=2,
    )

    assert result is True
    assert redis.calls == [
        {
            "key": notifier.converged_alert_recurrence_key("abc"),
            "value": "2",
            "ex": notifier.CONVERGED_ALERT_RECURRENCE_TTL_SECONDS,
            "nx": True,
        }
    ]


@pytest.mark.asyncio
async def test_converged_recurrence_throttles_when_redis_key_exists(monkeypatch):
    redis = _FakeRedis(False)
    monkeypatch.setattr(notifier, "get_redis", lambda: redis)

    result = await notifier.should_notify_converged_alert_recurrence(
        fingerprint="abc",
        hit_count=3,
    )

    assert result is False


@pytest.mark.asyncio
async def test_converged_recurrence_falls_back_to_milestones(monkeypatch):
    def _raise_redis_error():
        raise RuntimeError("redis unavailable")

    monkeypatch.setattr(notifier, "get_redis", _raise_redis_error)

    assert await notifier.should_notify_converged_alert_recurrence(
        fingerprint="abc",
        hit_count=3,
    )
    assert not await notifier.should_notify_converged_alert_recurrence(
        fingerprint="abc",
        hit_count=4,
    )


@pytest.mark.asyncio
async def test_converged_recurrence_mirrors_to_private_chat(monkeypatch):
    gateway = _FakeGateway()

    async def _always_notify(*, fingerprint, hit_count):
        return True

    monkeypatch.setattr(notifier, "should_notify_converged_alert_recurrence", _always_notify)
    monkeypatch.setattr(notifier, "get_telegram_gateway", lambda: gateway)
    monkeypatch.setattr(notifier.settings, "OPENCLAW_TG_CHAT_ID", "private-chat")

    await notifier.notify_converged_alert_recurrence(
        source="alertmanager",
        fingerprint="abc",
        alertname="ServiceDown",
        severity="critical",
        namespace="prod",
        target_resource="api",
        hit_count=9,
        incident_id="INC-20260611-ABCD",
        approval_id="approval-1",
        alert_category="service",
        notification_type="TYPE-3",
    )

    assert len(gateway.primary_messages) == 1
    assert gateway.private_messages == [
        {
            "text": gateway.primary_messages[0],
            "chat_id": "private-chat",
        }
    ]