PapaTramp
ebb450be92
Singleton notification_policy table, NOTIFY_MIN_SEVERITY/CHANNELS env, central dispatch gate, Settings policy UI, migration 007. Co-authored-by: Cursor <cursoragent@cursor.com>
63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
"""notify_dispatch respects global policy."""
|
|
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import patch
|
|
|
|
from app.models import Event
|
|
from app.services import notify_dispatch
|
|
from app.services.notification_policy import NotificationPolicyConfig
|
|
|
|
|
|
def test_notify_event_skipped_below_policy_severity():
|
|
event = Event(
|
|
event_id="00000000-0000-4000-8000-000000000401",
|
|
host_id=1,
|
|
occurred_at=datetime(2026, 5, 29, 15, 0, tzinfo=timezone.utc),
|
|
category="auth",
|
|
type="rdp.login.success",
|
|
severity="info",
|
|
title="ok",
|
|
summary="skip",
|
|
payload={},
|
|
)
|
|
policy = NotificationPolicyConfig(
|
|
min_severity="warning",
|
|
use_telegram=True,
|
|
use_webhook=False,
|
|
use_email=False,
|
|
source="db",
|
|
)
|
|
with patch.object(notify_dispatch, "get_effective_notification_policy", return_value=policy):
|
|
with patch.object(notify_dispatch, "telegram_notify") as mock_tg:
|
|
notify_dispatch.notify_event(event)
|
|
mock_tg.notify_event.assert_not_called()
|
|
|
|
|
|
def test_notify_event_calls_selected_channels():
|
|
event = Event(
|
|
event_id="00000000-0000-4000-8000-000000000402",
|
|
host_id=1,
|
|
occurred_at=datetime(2026, 5, 29, 15, 0, tzinfo=timezone.utc),
|
|
category="auth",
|
|
type="rdp.login.failed",
|
|
severity="warning",
|
|
title="fail",
|
|
summary="send",
|
|
payload={},
|
|
)
|
|
policy = NotificationPolicyConfig(
|
|
min_severity="warning",
|
|
use_telegram=True,
|
|
use_webhook=True,
|
|
use_email=False,
|
|
source="db",
|
|
)
|
|
with patch.object(notify_dispatch, "get_effective_notification_policy", return_value=policy):
|
|
with patch.object(notify_dispatch, "telegram_notify") as mock_tg:
|
|
with patch.object(notify_dispatch, "webhook_notify") as mock_wh:
|
|
with patch.object(notify_dispatch, "email_notify") as mock_em:
|
|
notify_dispatch.notify_event(event)
|
|
mock_tg.notify_event.assert_called_once()
|
|
mock_wh.notify_event.assert_called_once()
|
|
mock_em.notify_event.assert_not_called()
|