249 lines
8.8 KiB
Python
249 lines
8.8 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from datetime import UTC, datetime, timedelta
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
|
from monlet_server.models import Incident, NotificationOutbox
|
|
from monlet_server.services.notifier_worker import tick
|
|
from monlet_server.services.notifiers.base import DeliveryResult
|
|
from monlet_server.settings import Settings
|
|
|
|
|
|
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",
|
|
}
|
|
|
|
|
|
# ---- Worker integration with real PG ----
|
|
|
|
|
|
@pytest.fixture
|
|
def settings_full() -> Settings:
|
|
s = Settings()
|
|
s.notifier_max_attempts = 3
|
|
s.notifier_batch_size = 50
|
|
return s
|
|
|
|
|
|
async def _seed_incident_and_outbox(
|
|
sm: async_sessionmaker, notifier: str
|
|
) -> tuple[NotificationOutbox, Incident]:
|
|
async with sm() as session:
|
|
# Need agent for FK on incidents.agent_id? No FK on incidents.agent_id per schema.
|
|
inc = Incident(
|
|
id=uuid4(),
|
|
incident_key=f"k-{uuid4()}",
|
|
agent_id="agent-1",
|
|
check_id="chk-1",
|
|
state="open",
|
|
severity="critical",
|
|
opened_at=datetime.now(UTC),
|
|
last_event_id=uuid4(),
|
|
summary="boom",
|
|
)
|
|
session.add(inc)
|
|
await session.flush()
|
|
row = NotificationOutbox(
|
|
id=uuid4(),
|
|
incident_id=inc.id,
|
|
notifier=notifier,
|
|
event_type="firing",
|
|
state="pending",
|
|
attempts=0,
|
|
next_attempt_at=datetime.now(UTC) - timedelta(seconds=1),
|
|
payload=_payload(),
|
|
)
|
|
session.add(row)
|
|
await session.commit()
|
|
return row, inc
|
|
|
|
|
|
class _StubNotifier:
|
|
def __init__(self, name: str, results: list[DeliveryResult]) -> None:
|
|
self.name = name
|
|
self._results = list(results)
|
|
self.calls = 0
|
|
|
|
async def deliver(self, event_type: str, payload: dict) -> DeliveryResult:
|
|
self.calls += 1
|
|
if self._results:
|
|
return self._results.pop(0)
|
|
return DeliveryResult(ok=True)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_worker_marks_sent(engine, settings_full):
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
|
row, _ = await _seed_incident_and_outbox(sm, "stub")
|
|
registry = {"stub": _StubNotifier("stub", [DeliveryResult(ok=True)])}
|
|
n = await tick(sm, registry, settings_full)
|
|
assert n == 1
|
|
async with sm() as session:
|
|
fresh = (
|
|
await session.execute(select(NotificationOutbox).where(NotificationOutbox.id == row.id))
|
|
).scalar_one()
|
|
assert fresh.state == "sent"
|
|
assert fresh.attempts == 1
|
|
assert fresh.last_error is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_worker_retries_then_succeeds(engine, settings_full):
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
|
row, _ = await _seed_incident_and_outbox(sm, "stub")
|
|
notifier = _StubNotifier(
|
|
"stub",
|
|
[DeliveryResult(ok=False, permanent=False, error="net"), DeliveryResult(ok=True)],
|
|
)
|
|
registry = {"stub": notifier}
|
|
await tick(sm, registry, settings_full)
|
|
async with sm() as session:
|
|
fresh = (
|
|
await session.execute(select(NotificationOutbox).where(NotificationOutbox.id == row.id))
|
|
).scalar_one()
|
|
assert fresh.state == "retry"
|
|
assert fresh.attempts == 1
|
|
assert fresh.next_attempt_at is not None
|
|
# Force eligibility for second tick.
|
|
fresh.next_attempt_at = datetime.now(UTC) - timedelta(seconds=1)
|
|
await session.commit()
|
|
await tick(sm, registry, settings_full)
|
|
async with sm() as session:
|
|
fresh = (
|
|
await session.execute(select(NotificationOutbox).where(NotificationOutbox.id == row.id))
|
|
).scalar_one()
|
|
assert fresh.state == "sent"
|
|
assert fresh.attempts == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_worker_permanent_failure_marked_failed(engine, settings_full):
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
|
row, _ = await _seed_incident_and_outbox(sm, "stub")
|
|
registry = {
|
|
"stub": _StubNotifier("stub", [DeliveryResult(ok=False, permanent=True, error="http_400")])
|
|
}
|
|
await tick(sm, registry, settings_full)
|
|
async with sm() as session:
|
|
fresh = (
|
|
await session.execute(select(NotificationOutbox).where(NotificationOutbox.id == row.id))
|
|
).scalar_one()
|
|
assert fresh.state == "failed"
|
|
assert fresh.last_error == "http_400"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_worker_unknown_notifier_discarded(engine, settings_full):
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
|
row, _ = await _seed_incident_and_outbox(sm, "ghost")
|
|
await tick(sm, {}, settings_full)
|
|
async with sm() as session:
|
|
fresh = (
|
|
await session.execute(select(NotificationOutbox).where(NotificationOutbox.id == row.id))
|
|
).scalar_one()
|
|
assert fresh.state == "discarded"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_worker_exhausts_attempts(engine, settings_full):
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
|
row, _ = await _seed_incident_and_outbox(sm, "stub")
|
|
notifier = _StubNotifier(
|
|
"stub",
|
|
[DeliveryResult(ok=False, permanent=False, error="boom")] * 10,
|
|
)
|
|
registry = {"stub": notifier}
|
|
for _ in range(settings_full.notifier_max_attempts):
|
|
async with sm() as session:
|
|
await session.execute(select(NotificationOutbox).where(NotificationOutbox.id == row.id))
|
|
# Force ready.
|
|
await session.execute(
|
|
NotificationOutbox.__table__.update()
|
|
.where(NotificationOutbox.id == row.id)
|
|
.values(next_attempt_at=datetime.now(UTC) - timedelta(seconds=1))
|
|
)
|
|
await session.commit()
|
|
await tick(sm, registry, settings_full)
|
|
async with sm() as session:
|
|
fresh = (
|
|
await session.execute(select(NotificationOutbox).where(NotificationOutbox.id == row.id))
|
|
).scalar_one()
|
|
assert fresh.state == "failed"
|
|
assert fresh.attempts == settings_full.notifier_max_attempts
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_worker_recovers_stuck_sending(engine, settings_full):
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
|
row, _ = await _seed_incident_and_outbox(sm, "stub")
|
|
async with sm() as session:
|
|
await session.execute(
|
|
NotificationOutbox.__table__.update()
|
|
.where(NotificationOutbox.id == row.id)
|
|
.values(state="sending", updated_at=datetime.now(UTC) - timedelta(hours=1))
|
|
)
|
|
await session.commit()
|
|
settings_full.notifier_lease_sec = 60
|
|
registry = {"stub": _StubNotifier("stub", [DeliveryResult(ok=True)])}
|
|
n = await tick(sm, registry, settings_full)
|
|
assert n == 1
|
|
async with sm() as session:
|
|
fresh = (
|
|
await session.execute(select(NotificationOutbox).where(NotificationOutbox.id == row.id))
|
|
).scalar_one()
|
|
assert fresh.state == "sent"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fanout_creates_row_per_enabled_notifier(
|
|
app_client, ui_auth_headers, session, monkeypatch
|
|
):
|
|
# Enable telegram (token+chat_id satisfied) and webhook in addition to debug.
|
|
os.environ["MONLET_NOTIFIER_TELEGRAM_ENABLED"] = "true"
|
|
os.environ["MONLET_NOTIFIER_TELEGRAM_TOKEN"] = "tok"
|
|
os.environ["MONLET_NOTIFIER_TELEGRAM_CHAT_ID"] = "1"
|
|
os.environ["MONLET_NOTIFIER_WEBHOOK_ENABLED"] = "true"
|
|
os.environ["MONLET_NOTIFIER_WEBHOOK_URL"] = "http://x/hook"
|
|
from monlet_server.settings import reset_settings_cache
|
|
|
|
reset_settings_cache()
|
|
try:
|
|
from tests._helpers import make_event, register_agent
|
|
|
|
auth_headers = await register_agent(app_client, ui_auth_headers)
|
|
ev = make_event(status="critical", exit_code=2)
|
|
r = await app_client.post(
|
|
"/api/v1/events",
|
|
json={"agent_id": "agent-1", "events": [ev]},
|
|
headers=auth_headers,
|
|
)
|
|
assert r.status_code == 202
|
|
rows = (await session.execute(select(NotificationOutbox))).scalars().all()
|
|
names = sorted(r.notifier for r in rows)
|
|
assert names == ["debug", "telegram", "webhook"]
|
|
finally:
|
|
for k in (
|
|
"MONLET_NOTIFIER_TELEGRAM_ENABLED",
|
|
"MONLET_NOTIFIER_TELEGRAM_TOKEN",
|
|
"MONLET_NOTIFIER_TELEGRAM_CHAT_ID",
|
|
"MONLET_NOTIFIER_WEBHOOK_ENABLED",
|
|
"MONLET_NOTIFIER_WEBHOOK_URL",
|
|
):
|
|
os.environ.pop(k, None)
|
|
reset_settings_cache()
|