monlet/server/unit_tests/test_notifiers.py

from __future__ import annotations

import json
from uuid import uuid4

import httpx
import pytest

from monlet_server.services.notifier_worker import _backoff
from monlet_server.services.notifiers.alertmanager import AlertmanagerNotifier
from monlet_server.services.notifiers.debug import DebugNotifier
from monlet_server.services.notifiers.telegram import TelegramNotifier
from monlet_server.services.notifiers.webhook import WebhookNotifier


def _payload() -> dict:
    return {
        "incident_id": str(uuid4()),
        "agent_id": "agent-1",
        "check_id": "chk-1",
        "status": "critical",
        "severity": "critical",
        "incident_key": "agent-1:chk-1",
        "observed_at": "2026-05-27T05:00:00+00:00",
        "opened_at": "2026-05-27T05:00:00+00:00",
        "summary": "boom",
        "output": "stderr: boom token=abc123",
    }


class _StubTransport(httpx.AsyncBaseTransport):
    def __init__(self, handler):
        self._handler = handler
        self.requests: list[httpx.Request] = []

    async def handle_async_request(self, request):
        self.requests.append(request)
        return self._handler(request)


def _client(handler) -> tuple[httpx.AsyncClient, _StubTransport]:
    transport = _StubTransport(handler)
    return httpx.AsyncClient(transport=transport, base_url="http://x"), transport


@pytest.mark.asyncio
async def test_debug_notifier_always_ok():
    res = await DebugNotifier().deliver("firing", _payload())
    assert res.ok and not res.permanent


@pytest.mark.asyncio
async def test_telegram_success_and_redaction():
    def handler(req):
        body = req.content.decode()
        assert "token=abc123" not in body
        assert "2026-05-27 09:00:00 +04" in body
        assert "***" in body
        return httpx.Response(200, json={"ok": True})

    client, transport = _client(handler)
    try:
        notifier = TelegramNotifier(client, "TOK", "123", "Asia/Tbilisi")
        res = await notifier.deliver("firing", _payload())
        assert res.ok
        assert "/botTOK/sendMessage" in str(transport.requests[0].url)
    finally:
        await client.aclose()


@pytest.mark.asyncio
async def test_telegram_4xx_permanent():
    client, _ = _client(lambda r: httpx.Response(400, json={}))
    try:
        notifier = TelegramNotifier(client, "TOK", "123", "UTC")
        res = await notifier.deliver("firing", _payload())
        assert not res.ok and res.permanent
    finally:
        await client.aclose()


@pytest.mark.asyncio
async def test_telegram_5xx_transient():
    client, _ = _client(lambda r: httpx.Response(503, json={}))
    try:
        notifier = TelegramNotifier(client, "TOK", "123", "UTC")
        res = await notifier.deliver("firing", _payload())
        assert not res.ok and not res.permanent
    finally:
        await client.aclose()


@pytest.mark.asyncio
async def test_telegram_network_error_transient():
    def handler(req):
        raise httpx.ConnectError("nope")

    client, _ = _client(handler)
    try:
        notifier = TelegramNotifier(client, "TOK", "123", "UTC")
        res = await notifier.deliver("firing", _payload())
        assert not res.ok and not res.permanent
    finally:
        await client.aclose()


@pytest.mark.asyncio
async def test_webhook_sends_bearer_and_redacts_payload():
    captured: dict = {}

    def handler(req):
        captured["auth"] = req.headers.get("authorization")
        captured["body"] = req.content.decode()
        return httpx.Response(202)

    client, _ = _client(handler)
    try:
        notifier = WebhookNotifier(client, "http://x/hook", token="WTOK")
        res = await notifier.deliver("firing", _payload())
        assert res.ok
        assert captured["auth"] == "Bearer WTOK"
        assert "token=abc123" not in captured["body"]
    finally:
        await client.aclose()


@pytest.mark.asyncio
async def test_alertmanager_payload_shape():
    captured: dict = {}

    def handler(req):
        captured["url"] = str(req.url)
        captured["body"] = req.content.decode()
        return httpx.Response(200)

    client, _ = _client(handler)
    try:
        notifier = AlertmanagerNotifier(client, "http://am/", "Asia/Tbilisi")
        res = await notifier.deliver("firing", _payload())
        assert res.ok
        assert captured["url"].endswith("/api/v2/alerts")
        body = json.loads(captured["body"])
        alert = body[0]
        assert alert["labels"]["alertname"] == "monlet_incident"
        assert alert["startsAt"] == "2026-05-27T05:00:00+00:00"
        assert alert["annotations"]["observed_at"] == "2026-05-27 09:00:00 +04"
    finally:
        await client.aclose()


def test_backoff_schedule():
    expected = [5, 15, 60, 300, 1800, 3600, 7200, 14400]
    for i, sec in enumerate(expected, start=1):
        assert _backoff(i).total_seconds() == sec
    assert _backoff(99).total_seconds() == 14400


@pytest.mark.asyncio
async def test_alertmanager_severity_not_in_labels():
    captured: dict = {}

    def handler(req):
        captured["body"] = req.content.decode()
        return httpx.Response(200)

    client, _ = _client(handler)
    try:
        notifier = AlertmanagerNotifier(client, "http://am", "UTC")
        await notifier.deliver("firing", _payload())
        alerts = json.loads(captured["body"])
        assert "severity" not in alerts[0]["labels"]
        assert alerts[0]["annotations"]["severity"] == "critical"
    finally:
        await client.aclose()


@pytest.mark.asyncio
async def test_telegram_redacts_secret_in_incident_key():
    captured: dict = {}

    def handler(req):
        captured["body"] = req.content.decode()
        return httpx.Response(200)

    client, _ = _client(handler)
    try:
        payload = _payload()
        payload["incident_key"] = "agent-1:chk-1:token=leakedsecret"
        notifier = TelegramNotifier(client, "TOK", "1", "UTC")
        await notifier.deliver("firing", payload)
        assert "leakedsecret" not in captured["body"]
    finally:
        await client.aclose()