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>
64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
"""Global notification policy (notif-22)."""
|
|
|
|
from app.config import get_settings
|
|
from app.models.notification_policy import NotificationPolicy
|
|
from app.services.notification_policy import get_effective_notification_policy, upsert_notification_policy
|
|
from app.services.notification_severity import severity_meets_minimum
|
|
|
|
|
|
def test_severity_meets_minimum():
|
|
assert severity_meets_minimum("warning", "warning")
|
|
assert severity_meets_minimum("high", "warning")
|
|
assert not severity_meets_minimum("info", "warning")
|
|
|
|
|
|
def test_put_policy_persists_db(jwt_headers, client, db_session, monkeypatch):
|
|
monkeypatch.setenv("NOTIFY_MIN_SEVERITY", "high")
|
|
monkeypatch.setenv("NOTIFY_CHANNELS", "telegram")
|
|
get_settings.cache_clear()
|
|
|
|
response = client.put(
|
|
"/api/v1/settings/notifications/policy",
|
|
headers=jwt_headers,
|
|
json={
|
|
"min_severity": "warning",
|
|
"use_telegram": True,
|
|
"use_webhook": True,
|
|
"use_email": False,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["min_severity"] == "warning"
|
|
assert body["use_webhook"] is True
|
|
assert body["source"] == "db"
|
|
|
|
row = db_session.get(NotificationPolicy, 1)
|
|
assert row is not None
|
|
assert row.use_webhook is True
|
|
|
|
cfg = get_effective_notification_policy(db_session)
|
|
assert cfg.min_severity == "warning"
|
|
|
|
|
|
def test_upsert_policy_syncs_channel_min_severity(db_session, monkeypatch):
|
|
monkeypatch.setenv("NOTIFY_MIN_SEVERITY", "high")
|
|
get_settings.cache_clear()
|
|
|
|
from app.models.notification_channel import NotificationChannel
|
|
|
|
db_session.add(
|
|
NotificationChannel(channel="telegram", enabled=False, min_severity="high")
|
|
)
|
|
db_session.commit()
|
|
|
|
upsert_notification_policy(
|
|
db_session,
|
|
min_severity="critical",
|
|
use_telegram=True,
|
|
use_webhook=False,
|
|
use_email=False,
|
|
)
|
|
row = db_session.get(NotificationChannel, "telegram")
|
|
assert row.min_severity == "critical"
|