feat: Telegram settings in DB with UI edit and test send

Store notification_channels (migration 004), effective config DB over env,
PUT/test API endpoints, Settings form, and pass db session into ingest notify.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
PTah
2026-05-29 16:05:32 +10:00
parent e0ba384705
commit 79a78dc7c9
12 changed files with 564 additions and 116 deletions
+30 -27
View File
@@ -3,24 +3,27 @@
from datetime import datetime, timezone
from unittest.mock import patch
import pytest
from app.models import Event, Host, Problem
from app.models import Event, Problem
from app.services import telegram_notify
from app.services.notification_settings import TelegramConfig
def test_notify_event_respects_min_severity(monkeypatch):
def _telegram_cfg(*, min_severity: str) -> TelegramConfig:
return TelegramConfig(
enabled=True,
bot_token="test-token",
chat_id="123",
min_severity=min_severity,
source="env",
)
def test_notify_event_respects_min_severity():
sent: list[str] = []
def fake_send(message: str) -> None:
def fake_send(message: str, **kwargs) -> None:
sent.append(message)
monkeypatch.setenv("TELEGRAM_ENABLED", "true")
monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "test-token")
monkeypatch.setenv("TELEGRAM_CHAT_ID", "123")
monkeypatch.setenv("TELEGRAM_MIN_SEVERITY", "warning")
telegram_notify.get_settings.cache_clear()
event = Event(
event_id="00000000-0000-4000-8000-000000000001",
host_id=1,
@@ -33,28 +36,25 @@ def test_notify_event_respects_min_severity(monkeypatch):
payload={},
)
with patch.object(telegram_notify, "_send_text", side_effect=fake_send):
telegram_notify.notify_event(event)
cfg = _telegram_cfg(min_severity="warning")
with patch.object(telegram_notify, "get_effective_telegram_config", return_value=cfg):
with patch.object(telegram_notify, "send_telegram_text", side_effect=fake_send):
telegram_notify.notify_event(event)
assert sent == []
event.severity = "warning"
with patch.object(telegram_notify, "_send_text", side_effect=fake_send):
telegram_notify.notify_event(event)
with patch.object(telegram_notify, "get_effective_telegram_config", return_value=cfg):
with patch.object(telegram_notify, "send_telegram_text", side_effect=fake_send):
telegram_notify.notify_event(event)
assert len(sent) == 1
def test_notify_problem_respects_min_severity(monkeypatch):
def test_notify_problem_respects_min_severity():
sent: list[str] = []
def fake_send(message: str) -> None:
def fake_send(message: str, **kwargs) -> None:
sent.append(message)
monkeypatch.setenv("TELEGRAM_ENABLED", "true")
monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "test-token")
monkeypatch.setenv("TELEGRAM_CHAT_ID", "123")
monkeypatch.setenv("TELEGRAM_MIN_SEVERITY", "high")
telegram_notify.get_settings.cache_clear()
problem = Problem(
title="brute",
summary="burst",
@@ -63,11 +63,14 @@ def test_notify_problem_respects_min_severity(monkeypatch):
fingerprint="fp1",
)
with patch.object(telegram_notify, "_send_text", side_effect=fake_send):
telegram_notify.notify_problem(problem)
cfg = _telegram_cfg(min_severity="high")
with patch.object(telegram_notify, "get_effective_telegram_config", return_value=cfg):
with patch.object(telegram_notify, "send_telegram_text", side_effect=fake_send):
telegram_notify.notify_problem(problem)
assert sent == []
problem.severity = "high"
with patch.object(telegram_notify, "_send_text", side_effect=fake_send):
telegram_notify.notify_problem(problem)
with patch.object(telegram_notify, "get_effective_telegram_config", return_value=cfg):
with patch.object(telegram_notify, "send_telegram_text", side_effect=fake_send):
telegram_notify.notify_problem(problem)
assert len(sent) == 1