"""GET/PUT/POST /api/v1/settings/notifications""" from unittest.mock import patch from app.config import get_settings from app.models.notification_channel import NotificationChannel from app.services.notification_settings import get_effective_telegram_config def test_get_notification_settings_masks_secrets(jwt_headers, client, monkeypatch): monkeypatch.setenv("TELEGRAM_ENABLED", "true") monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "1234567890:ABCDEFghij") monkeypatch.setenv("TELEGRAM_CHAT_ID", "2843230") monkeypatch.setenv("TELEGRAM_MIN_SEVERITY", "warning") get_settings.cache_clear() response = client.get("/api/v1/settings/notifications", headers=jwt_headers) assert response.status_code == 200 body = response.json() assert body["telegram"]["enabled"] is True assert body["telegram"]["configured"] is True 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 assert "email" in body assert body["email"]["enabled"] is False def test_put_email_settings_persists_db(jwt_headers, client, db_session, monkeypatch): monkeypatch.setenv("SMTP_ENABLED", "false") monkeypatch.setenv("SMTP_HOST", "") get_settings.cache_clear() response = client.put( "/api/v1/settings/notifications/email", headers=jwt_headers, json={ "enabled": True, "min_severity": "warning", "smtp_host": "smtp.example.com", "smtp_port": 587, "smtp_user": "monitor", "smtp_password": "secret", "mail_from": "sac@example.com", "mail_to": "admin@example.com,ops@example.com", "smtp_starttls": True, "smtp_ssl": False, }, ) assert response.status_code == 200 body = response.json() assert body["source"] == "db" assert body["configured"] is True assert body["smtp_port"] == 587 row = db_session.get(NotificationChannel, "email") assert row is not None assert row.smtp_host == "smtp.example.com" assert row.mail_from == "sac@example.com" def test_post_email_test_success(jwt_headers, client, db_session, monkeypatch): monkeypatch.setenv("SMTP_ENABLED", "false") get_settings.cache_clear() db_session.add( NotificationChannel( channel="email", enabled=True, min_severity="warning", smtp_host="smtp.example.com", smtp_port=587, mail_from="sac@example.com", mail_to="admin@example.com", ) ) db_session.commit() with patch("app.api.v1.settings.send_email_test_message") as mock_test: response = client.post("/api/v1/settings/notifications/email/test", headers=jwt_headers) assert response.status_code == 200 assert response.json()["ok"] is True mock_test.assert_called_once() 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): monkeypatch.setenv("TELEGRAM_ENABLED", "false") monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "") monkeypatch.setenv("TELEGRAM_CHAT_ID", "") monkeypatch.setenv("TELEGRAM_MIN_SEVERITY", "high") get_settings.cache_clear() response = client.put( "/api/v1/settings/notifications/telegram", headers=jwt_headers, json={ "enabled": True, "min_severity": "warning", "bot_token": "1234567890:TESTTOKEN", "chat_id": "999001", }, ) assert response.status_code == 200 body = response.json() assert body["source"] == "db" assert body["enabled"] is True assert body["min_severity"] == "warning" row = db_session.get(NotificationChannel, "telegram") assert row is not None assert row.bot_token == "1234567890:TESTTOKEN" assert row.chat_id == "999001" cfg = get_effective_telegram_config(db_session) assert cfg.source == "db" assert cfg.min_severity == "warning" def test_post_telegram_test_success(jwt_headers, client, db_session, monkeypatch): monkeypatch.setenv("TELEGRAM_ENABLED", "true") monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "tok") monkeypatch.setenv("TELEGRAM_CHAT_ID", "1") get_settings.cache_clear() db_session.add( NotificationChannel( channel="telegram", enabled=True, min_severity="warning", bot_token="1234567890:ABCDEF", chat_id="2843230", ) ) db_session.commit() with patch("app.api.v1.settings.send_telegram_test_message") as mock_test: response = client.post("/api/v1/settings/notifications/telegram/test", headers=jwt_headers) assert response.status_code == 200 assert response.json()["ok"] is True mock_test.assert_called_once() def test_post_telegram_test_not_configured(jwt_headers, client, monkeypatch): monkeypatch.setenv("TELEGRAM_ENABLED", "false") monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "") monkeypatch.setenv("TELEGRAM_CHAT_ID", "") get_settings.cache_clear() response = client.post("/api/v1/settings/notifications/telegram/test", headers=jwt_headers) assert response.status_code == 400