feat: webhook notification channel with UI and ingest dispatch

Add webhook config (DB/env), JSON POST on events/problems, settings API,
Settings UI section, and notify_dispatch for multi-channel ingest.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
PTah
2026-05-29 16:09:40 +10:00
parent f7b5fff04e
commit 20fa7e2c27
13 changed files with 826 additions and 189 deletions
+50
View File
@@ -22,6 +22,56 @@ def test_get_notification_settings_masks_secrets(jwt_headers, client, monkeypatc
assert body["telegram"]["min_severity"] == "warning"
assert body["telegram"]["source"] == "env"
assert body["telegram"]["bot_token_hint"].endswith("ghij")
assert "webhook" in body
assert body["webhook"]["enabled"] is False
def test_put_webhook_settings_persists_db(jwt_headers, client, db_session, monkeypatch):
monkeypatch.setenv("WEBHOOK_ENABLED", "false")
monkeypatch.setenv("WEBHOOK_URL", "")
get_settings.cache_clear()
response = client.put(
"/api/v1/settings/notifications/webhook",
headers=jwt_headers,
json={
"enabled": True,
"min_severity": "warning",
"url": "https://example.com/hooks/sac",
"secret_header": "X-SAC-Token",
"secret": "supersecret",
},
)
assert response.status_code == 200
body = response.json()
assert body["source"] == "db"
assert body["configured"] is True
row = db_session.get(NotificationChannel, "webhook")
assert row is not None
assert row.webhook_url == "https://example.com/hooks/sac"
assert row.webhook_secret_header == "X-SAC-Token"
def test_post_webhook_test_success(jwt_headers, client, db_session, monkeypatch):
monkeypatch.setenv("WEBHOOK_ENABLED", "false")
get_settings.cache_clear()
db_session.add(
NotificationChannel(
channel="webhook",
enabled=True,
min_severity="warning",
webhook_url="https://example.com/hook",
)
)
db_session.commit()
with patch("app.api.v1.settings.send_webhook_test_message") as mock_test:
response = client.post("/api/v1/settings/notifications/webhook/test", headers=jwt_headers)
assert response.status_code == 200
assert response.json()["ok"] is True
mock_test.assert_called_once()
def test_put_telegram_settings_persists_db(jwt_headers, client, db_session, monkeypatch):
+55
View File
@@ -0,0 +1,55 @@
"""Tests for SAC webhook notification helpers."""
from datetime import datetime, timezone
from unittest.mock import patch
from app.models import Event
from app.services import webhook_notify
from app.services.notification_settings import WebhookConfig
def test_notify_event_respects_min_severity():
sent: list[dict] = []
def fake_send(payload: dict, **kwargs) -> None:
sent.append(payload)
event = Event(
event_id="00000000-0000-4000-8000-000000000201",
host_id=1,
occurred_at=datetime(2026, 5, 29, 12, 0, tzinfo=timezone.utc),
category="auth",
type="rdp.login.failed",
severity="warning",
title="failed",
summary="e2e",
payload={},
)
cfg = WebhookConfig(
enabled=True,
url="https://example.com/hook",
secret_header="",
secret="",
min_severity="high",
source="env",
)
with patch.object(webhook_notify, "get_effective_webhook_config", return_value=cfg):
with patch.object(webhook_notify, "send_webhook_payload", side_effect=fake_send):
webhook_notify.notify_event(event)
assert sent == []
cfg_high = WebhookConfig(
enabled=True,
url="https://example.com/hook",
secret_header="",
secret="",
min_severity="warning",
source="env",
)
event.severity = "warning"
with patch.object(webhook_notify, "get_effective_webhook_config", return_value=cfg_high):
with patch.object(webhook_notify, "send_webhook_payload", side_effect=fake_send):
webhook_notify.notify_event(event)
assert len(sent) == 1
assert sent[0]["kind"] == "event"