79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from uuid import uuid4
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from monlet_server.services.notifiers.alertmanager import AlertmanagerNotifier
|
|
from monlet_server.services.notifiers.telegram import TelegramNotifier
|
|
from monlet_server.services.notifiers.webhook import WebhookNotifier
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
def _live_enabled() -> bool:
|
|
return os.environ.get("MONLET_LIVE_NOTIFIER_TESTS") == "1"
|
|
|
|
|
|
def _payload() -> dict:
|
|
return {
|
|
"incident_id": str(uuid4()),
|
|
"agent_id": "release-smoke",
|
|
"check_id": "notifier",
|
|
"status": "critical",
|
|
"severity": "critical",
|
|
"incident_key": f"release-smoke:notifier:{uuid4()}",
|
|
"observed_at": "2026-06-23T00:00:00+00:00",
|
|
"opened_at": "2026-06-23T00:00:00+00:00",
|
|
"summary": "Monlet release notifier smoke",
|
|
"output": "release notifier smoke",
|
|
}
|
|
|
|
|
|
def _require_live_env(*names: str) -> None:
|
|
missing = [name for name in names if not os.environ.get(name)]
|
|
if not _live_enabled() or missing:
|
|
pytest.skip(
|
|
"set MONLET_LIVE_NOTIFIER_TESTS=1 and required notifier env to run live smoke"
|
|
)
|
|
|
|
|
|
async def test_live_webhook_notifier_smoke():
|
|
_require_live_env("MONLET_LIVE_WEBHOOK_URL")
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
notifier = WebhookNotifier(
|
|
client,
|
|
os.environ["MONLET_LIVE_WEBHOOK_URL"],
|
|
os.environ.get("MONLET_LIVE_WEBHOOK_TOKEN", ""),
|
|
"UTC",
|
|
)
|
|
result = await notifier.deliver("firing", _payload())
|
|
assert result.ok, result.error
|
|
|
|
|
|
async def test_live_telegram_notifier_smoke():
|
|
_require_live_env("MONLET_LIVE_TELEGRAM_TOKEN", "MONLET_LIVE_TELEGRAM_CHAT_ID")
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
notifier = TelegramNotifier(
|
|
client,
|
|
os.environ["MONLET_LIVE_TELEGRAM_TOKEN"],
|
|
os.environ["MONLET_LIVE_TELEGRAM_CHAT_ID"],
|
|
"UTC",
|
|
)
|
|
result = await notifier.deliver("firing", _payload())
|
|
assert result.ok, result.error
|
|
|
|
|
|
async def test_live_alertmanager_notifier_smoke():
|
|
_require_live_env("MONLET_LIVE_ALERTMANAGER_URL")
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
notifier = AlertmanagerNotifier(
|
|
client,
|
|
os.environ["MONLET_LIVE_ALERTMANAGER_URL"],
|
|
"UTC",
|
|
)
|
|
result = await notifier.deliver("firing", _payload())
|
|
assert result.ok, result.error
|