# tests/test_notify_channels.py — External notifications (#168) """Tests for backend.notify (Discord, Telegram, SMTP, webhook), its REST routes and the ``notify_external`` AI tool. Network calls are mocked: no test ever hits the real Internet (hermetic CI). """ import pytest from backend import notify as _notify from backend.tools.api import ToolContext, call_tool @pytest.fixture def isolated_store(tmp_path, monkeypatch): """Redirect the channel + secret stores to a tmp dir.""" channels = tmp_path / "notify_channels.json" secrets = tmp_path / "notify_secrets.json" monkeypatch.setattr(_notify, "CHANNELS_FILE", channels) monkeypatch.setattr(_notify, "SECRETS_FILE", secrets) monkeypatch.setenv("OBSIGATE_WEBHOOK_ALLOW_PRIVATE", "true") return tmp_path def _ctx(**kwargs) -> ToolContext: user = {"username": "tester", "role": "admin", "vaults": ["*"]} kwargs.setdefault("audit_enabled", False) return ToolContext(user=user, **kwargs) class TestValidation: def test_unknown_type_rejected(self, isolated_store): with pytest.raises(ValueError): _notify.create_channel("x", "slack", {}) def test_discord_requires_url(self, isolated_store): with pytest.raises(ValueError): _notify.create_channel("d", "discord", {"webhook_url": "not-a-url"}) def test_telegram_requires_chat(self, isolated_store): with pytest.raises(ValueError): _notify.create_channel("t", "telegram", {}) def test_smtp_requires_fields(self, isolated_store): with pytest.raises(ValueError): _notify.create_channel("m", "smtp", {"host": "smtp.example.com"}) def test_bad_trigger_falls_back_to_manual(self, isolated_store): channel = _notify.create_channel( "w", "webhook", {"url": "https://example.com/hook", "triggers": ["nope"]}, ) assert channel["config"]["triggers"] == ["manual"] class TestCrud: def test_create_masks_secret(self, isolated_store): channel = _notify.create_channel( "disc", "discord", { "webhook_url": "https://discord.com/api/webhooks/123/abcdef-secret-token", "triggers": ["manual", "schedule_failure"], }, ) assert channel["has_secret"] is True assert channel["config"]["webhook_url"] == "***" # Le secret vit dans le store dédié, pas dans le fichier public. raw = isolated_store.joinpath("notify_channels.json").read_text(encoding="utf-8") assert "abcdef-secret-token" not in raw def test_update_and_delete(self, isolated_store): channel = _notify.create_channel("w", "webhook", {"url": "https://example.com/a"}) updated = _notify.update_channel(channel["id"], {"enabled": False}) assert updated is not None and updated["enabled"] is False assert _notify.delete_channel(channel["id"]) is True assert _notify.delete_channel(channel["id"]) is False def test_update_unknown_returns_none(self, isolated_store): assert _notify.update_channel("nope", {"enabled": True}) is None class TestDispatch: def test_broadcast_skips_unsubscribed_trigger(self, isolated_store, monkeypatch): _notify.create_channel("w", "webhook", {"url": "https://example.com/a", "triggers": ["manual"]}) calls: list = [] monkeypatch.setattr(_notify, "send_via_channel", lambda ch, t, m, trig="manual": calls.append(ch["id"])) results = _notify.broadcast("schedule_failure", "T", "M") assert results == [] assert calls == [] def test_broadcast_delivers_and_records_failure(self, isolated_store, monkeypatch): channel = _notify.create_channel( "w", "webhook", {"url": "https://example.com/a", "triggers": ["manual"]} ) monkeypatch.setattr(_notify, "send_via_channel", lambda ch, t, m, trig="manual": None) results = _notify.broadcast("manual", "T", "M") assert results == [{"channel_id": channel["id"], "ok": True}] def _boom(ch, t, m, trig="manual"): raise RuntimeError("down") monkeypatch.setattr(_notify, "send_via_channel", _boom) results = _notify.broadcast("manual", "T", "M") assert results[0]["ok"] is False assert "down" in results[0]["error"] def test_smtp_send_uses_smtplib(self, isolated_store, monkeypatch): _notify.create_channel( "mail", "smtp", { "host": "smtp.example.com", "port": 587, "from_addr": "a@example.com", "to_addr": "b@example.com", "username": "a", "password": "s3cret-pwd", "triggers": ["manual"], }, ) sent: dict = {} class _FakeSMTP: def __init__(self, *a, **k): pass def __enter__(self): return self def __exit__(self, *a): return False def starttls(self): sent["tls"] = True def login(self, user, pwd): sent["login"] = (user, pwd) def send_message(self, msg): sent["subject"] = msg["Subject"] monkeypatch.setattr(_notify.smtplib, "SMTP", _FakeSMTP) results = _notify.broadcast("manual", "Sujet", "Corps") assert results[0]["ok"] is True assert sent["tls"] is True assert sent["login"] == ("a", "s3cret-pwd") assert "Sujet" in sent["subject"] class TestNotifyTool: def test_tool_broadcasts(self, isolated_store, monkeypatch): monkeypatch.setattr(_notify, "broadcast", lambda trig, t, m: [{"channel_id": "c", "ok": True}]) result = call_tool( "notify_external", _ctx(confirmed=True), {"title": "Hello", "message": "World", "trigger": "manual"}, ) assert result.ok assert result.data["deliveries"] == [{"channel_id": "c", "ok": True}] def test_tool_unknown_channel(self, isolated_store): from backend.tools.context import ToolError with pytest.raises(ToolError): call_tool( "notify_external", _ctx(confirmed=True), {"title": "H", "message": "M", "channel_id": "unknown"}, ) class TestNotifyApi: def test_channels_crud(self, client, isolated_store): created = client.post( "/api/notify/channels", json={"name": "w", "type": "webhook", "config": {"url": "https://example.com/a"}}, ) assert created.status_code == 200, created.text channel_id = created.json()["id"] assert created.json()["has_secret"] is False listed = client.get("/api/notify/channels") assert listed.status_code == 200 assert any(c["id"] == channel_id for c in listed.json()) patched = client.patch(f"/api/notify/channels/{channel_id}", json={"enabled": False}) assert patched.status_code == 200 assert patched.json()["enabled"] is False deleted = client.delete(f"/api/notify/channels/{channel_id}") assert deleted.status_code == 200 def test_create_rejects_bad_type(self, client, isolated_store): resp = client.post("/api/notify/channels", json={"name": "x", "type": "slack", "config": {}}) assert resp.status_code == 400 def test_test_endpoint_no_channel(self, client, isolated_store): resp = client.post("/api/notify/test", json={"title": "T", "message": "M"}) assert resp.status_code == 200 assert resp.json()["deliveries"] == []