tgit/tests/alerts/test_service.py

from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta

import pytest

from app.alerts.detectors import MONITORED, Condition, load_alerts_config
from app.alerts.service import AlertService
from app.db.models import Alert, AlertKind
from app.db.session import SessionMaker
from tests.conftest import ROOT

CFG = load_alerts_config(ROOT / "config").model_copy(update={"resolve_after": 0})
SETTLING = CFG.model_copy(update={"resolve_after": 60})
T0 = datetime(2026, 10, 17, 12, 0, tzinfo=UTC)
CRASH = Condition(AlertKind.CRASH, "svc", {"exit_code": 1}, container_id="c" * 64)


@dataclass
class Recorder:
    events: list[tuple[str, int, int]] = field(default_factory=list)  # (what, alert id, count)

    async def opened(self, alert: Alert, admins_only: bool) -> None:
        self.events.append(("admins" if admins_only else "opened", alert.id, alert.count))

    async def repeated(self, alert: Alert) -> None:
        self.events.append(("repeated", alert.id, alert.count))

    async def resolved(self, alert: Alert) -> None:
        self.events.append(("resolved", alert.id, alert.count))


@pytest.fixture
def rec() -> Recorder:
    return Recorder()


@pytest.fixture
def service(sessions: SessionMaker, rec: Recorder) -> AlertService:
    async def tail(container_id: str, lines: int) -> str:
        return f"last {lines} lines of {container_id[:4]}"

    return AlertService(sessions, rec, CFG, log_source=tail)


def at(minutes: float) -> datetime:
    return T0 + timedelta(minutes=minutes)


async def test_storm_is_one_alert_then_recovery(service: AlertService, rec: Recorder) -> None:
    for minute in range(5):  # the condition holds for five rounds
        await service.sync([CRASH], MONITORED, at(minute))
    await service.sync([], MONITORED, at(6))

    assert [e[0] for e in rec.events] == ["opened", "resolved"]
    alert = await service.get(rec.events[0][1])
    assert alert is not None and alert.payload["log"] == "last 30 lines of cccc"
    assert alert.payload["container_id"] == "c" * 64


async def test_flapping_within_window_bumps_counter(service: AlertService, rec: Recorder) -> None:
    await service.sync([CRASH], MONITORED, at(0))
    await service.sync([], MONITORED, at(1))
    await service.sync([CRASH], MONITORED, at(3))  # back within 10 minutes
    await service.sync([], MONITORED, at(4))
    await service.sync([CRASH], MONITORED, at(20))  # long after: a new alert

    what = [(e[0], e[2]) for e in rec.events]
    assert what == [("opened", 1), ("resolved", 1), ("repeated", 2), ("resolved", 2), ("opened", 1)]
    assert rec.events[0][1] == rec.events[2][1] != rec.events[4][1]


async def test_scope_protects_other_kinds(service: AlertService, rec: Recorder) -> None:
    await service.sync([CRASH], MONITORED, at(0))
    # Docker down: only these kinds are judged — the crash stays open
    await service.sync([], frozenset({AlertKind.DEPENDENCY_DOWN, AlertKind.DISK_LOW}), at(1))

    assert [e[0] for e in rec.events] == ["opened"]
    assert await service.active_count() == 1


async def test_event_alerts_repeat_with_counter(service: AlertService, rec: Recorder) -> None:
    attack = Condition(AlertKind.AGENT_ATTACK, "user 7", {"who": "u7", "stage": "guard", "reason": "x"})

    first = await service.event(attack, admins_only=True, now=at(0))
    second = await service.event(attack, admins_only=True, now=at(2))
    third = await service.event(attack, admins_only=True, now=at(30))

    assert first.id == second.id != third.id
    assert [(e[0], e[2]) for e in rec.events] == [("admins", 1), ("repeated", 2), ("admins", 1)]
    _, total = await service.page(0, now=at(31))
    assert total == 0  # 🛡 never shows in the team's list


async def test_list_active_first_then_history(service: AlertService) -> None:
    other = Condition(AlertKind.UNHEALTHY, "web")
    await service.sync([CRASH], MONITORED, at(0))
    await service.sync([], MONITORED, at(1))
    await service.sync([other], MONITORED, at(2))
    old = Condition(AlertKind.OOM, "ancient")
    await service.sync([other, old], MONITORED, at(-60 * 24 * 8))  # eight days ago, resolved right after
    await service.sync([other], MONITORED, at(-60 * 24 * 8 + 1))

    rows, total = await service.page(0, now=at(3))

    assert [(a.kind, a.resolved_at is None) for a in rows] == [(AlertKind.UNHEALTHY, True), (AlertKind.CRASH, False)]
    assert total == 2


async def test_broken_log_source_still_alerts(sessions: SessionMaker, rec: Recorder) -> None:
    async def broken(container_id: str, lines: int) -> str:
        raise RuntimeError("docker gone")

    service = AlertService(sessions, rec, CFG, log_source=broken)
    await service.sync([CRASH], MONITORED, at(0))

    alert = await service.get(rec.events[0][1])
    assert alert is not None and "log" not in alert.payload


async def test_resolution_waits_for_the_condition_to_settle(sessions: SessionMaker, rec: Recorder) -> None:
    service = AlertService(sessions, rec, SETTLING)
    await service.sync([CRASH], MONITORED, at(0))
    await service.sync([], MONITORED, at(0.5))  # one good round is not recovery
    await service.sync([CRASH], MONITORED, at(0.75))
    await service.sync([], MONITORED, at(1))
    await service.sync([], MONITORED, at(1.9))

    assert [e[0] for e in rec.events] == ["opened", "resolved"]


async def test_bumps_while_open_throttle_edits(sessions: SessionMaker, rec: Recorder) -> None:
    service = AlertService(sessions, rec, SETTLING)
    bumped = Condition(CRASH.kind, CRASH.service, {}, CRASH.container_id, bump=True)
    await service.sync([CRASH], MONITORED, at(0))
    for second in (15, 30, 45, 75):
        await service.sync([bumped], MONITORED, at(second / 60))

    assert [(e[0], e[2]) for e in rec.events] == [("opened", 1), ("repeated", 2), ("repeated", 5)]
    alert = await service.get(rec.events[0][1])
    assert alert is not None and alert.count == 5