app/backend/tests/flow/test_alerts.py

"""Alerting: failing loudly once, not thirty-six thousand times."""

import asyncio

import pytest

from app.flow.alerts import (
    FLAP_THRESHOLD,
    RATE_LIMIT,
    Alert,
    AlertManager,
    AlertsConfig,
    Channel,
    Rule,
    describe,
)
from app.flow.events import EventBus


class Clock:
    """A hand-wound clock, so cooldowns take no real time."""

    def __init__(self) -> None:
        self.now = 0.0

    def __call__(self) -> float:
        return self.now

    def advance(self, seconds: float) -> None:
        self.now += seconds


def manager(clock: Clock) -> tuple[AlertManager, list[Alert]]:
    sent: list[Alert] = []
    config = AlertsConfig(
        channels=[Channel(name="phone", kind="ntfy", config={"topic": "t"})],
        rules=[Rule(events=[], channels=["phone"], cooldown_s=900)],
    )
    alerts = AlertManager(EventBus(), config=config, now=clock)

    async def capture(channel, alert):
        sent.append(alert)

    alerts.send = capture  # type: ignore[method-assign]
    return alerts, sent


def error_event(node: str = "heating.pump") -> dict:
    return {"type": "node_error", "flow": "heating", "node": node, "error": "boom"}


def test_a_failure_reaches_the_channel():
    clock = Clock()
    alerts, sent = manager(clock)

    asyncio.run(alerts.handle(error_event()))

    assert [a.title for a in sent] == ["heating.pump failed"]
    assert sent[0].body == "boom"


def test_ordinary_traffic_is_not_an_alert():
    clock = Clock()
    alerts, sent = manager(clock)

    asyncio.run(alerts.handle({"type": "message_value", "name": "heating.temp"}))
    asyncio.run(alerts.handle({"type": "node_executed", "node": "heating.pump"}))

    assert sent == []


def test_the_same_failure_repeating_is_one_alert():
    clock = Clock()
    alerts, sent = manager(clock)

    async def scenario():
        for _ in range(50):
            await alerts.handle(error_event())
            clock.advance(1)

    asyncio.run(scenario())

    assert len(sent) == 1


def test_the_cooldown_ends_and_says_what_was_missed():
    clock = Clock()
    alerts, sent = manager(clock)

    async def scenario():
        await alerts.handle(error_event())
        for _ in range(4):
            clock.advance(10)
            await alerts.handle(error_event())
        clock.advance(1000)
        await alerts.handle(error_event())

    asyncio.run(scenario())

    assert len(sent) == 2
    assert "4 more since the last alert" in sent[1].body


def test_different_nodes_alert_separately():
    clock = Clock()
    alerts, sent = manager(clock)

    async def scenario():
        await alerts.handle(error_event("heating.pump"))
        await alerts.handle(error_event("heating.valve"))

    asyncio.run(scenario())

    assert len(sent) == 2


def test_a_flapping_connection_goes_quiet():
    """A device dropping every few seconds is one story, not one alert each."""
    clock = Clock()
    alerts, sent = manager(clock)

    async def scenario():
        for _ in range(FLAP_THRESHOLD * 2 + 6):
            await alerts.handle(
                {"type": "node_health", "node": "heating.pump", "health": "down"}
            )
            clock.advance(5)
            await alerts.handle(
                {"type": "node_health", "node": "heating.pump", "health": "ok"}
            )
            clock.advance(5)

    asyncio.run(scenario())

    # The first drop is worth knowing about; the rest is noise.
    assert len(sent) == 1


def test_a_storm_is_capped():
    clock = Clock()
    alerts, sent = manager(clock)

    async def scenario():
        for i in range(RATE_LIMIT + 20):
            await alerts.handle(error_event(f"heating.n{i}"))
            clock.advance(1)

    asyncio.run(scenario())

    assert len(sent) == RATE_LIMIT


def test_nothing_is_sent_without_a_rule():
    clock = Clock()
    alerts, sent = manager(clock)
    alerts.config = AlertsConfig(
        channels=[Channel(name="phone", kind="ntfy", config={"topic": "t"})],
        rules=[Rule(events=["queue_unavailable"], channels=["phone"])],
    )

    asyncio.run(alerts.handle(error_event()))

    assert sent == []


def test_alerting_can_be_switched_off():
    clock = Clock()
    alerts, sent = manager(clock)
    alerts.config = alerts.config.model_copy(update={"enabled": False})

    asyncio.run(alerts.handle(error_event()))

    assert sent == []


@pytest.mark.parametrize(
    ("event", "expected"),
    [
        (
            {"type": "flow_quarantined", "flow": "heating"},
            "Flow 'heating' was quarantined",
        ),
        (
            {"type": "queue_unavailable", "error": "gone"},
            "The work queue is unreachable",
        ),
        ({"type": "engine_degraded", "reason": "lag"}, "The engine is struggling"),
        ({"type": "node_health", "health": "ok"}, None),
        # The engine publishes `health`, not `status`: reading the wrong key
        # meant a device dropping never alerted anyone.
        (
            {"type": "node_health", "node": "heating.pump", "health": "down"},
            "heating.pump lost its connection",
        ),
    ],
)
def test_every_alerting_event_reads_as_a_sentence(event, expected):
    alert = describe(event)
    assert (alert.title if alert else None) == expected