203 lines
7.6 KiB
Python
203 lines
7.6 KiB
Python
# 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": "[email protected]",
|
|
"to_addr": "[email protected]",
|
|
"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"] == []
|