FabledSteward/tests/integration/test_notification_test_send.py

"""Integration: the test-send buttons on Settings → Notifications (#5064).

Each one sends a sample alert with the form's values, unsaved, and answers
with what happened and the body that went out. The routes run for real
against Postgres; only the delivery itself is replaced.
"""
from __future__ import annotations

import asyncio
import json
import os
import time
import uuid

import pytest

pytestmark = pytest.mark.integration


@pytest.fixture
def app():
    if not os.environ.get("STEWARD_DATABASE_URL"):
        pytest.skip("needs Postgres")
    from steward.app import create_app
    return create_app(testing=False)


async def _admin_client(app):
    from steward.models.users import User, UserRole
    uid = str(uuid.uuid4())
    async with app.db_sessionmaker() as s:
        async with s.begin():
            s.add(User(id=uid, username=f"notify-{uid[:8]}", email=f"{uid}@test",
                       password_hash="x", role=UserRole.admin, is_active=True))
    client = app.test_client()
    async with client.session_transaction() as sess:
        sess["user_id"] = uid
        sess["user_role"] = "admin"
        sess["username"] = f"notify-{uid[:8]}"
        sess["_csrf"] = "test-token"
        sess["login_at"] = time.time()
    return client, uid


async def _audits(app, uid):
    from sqlalchemy import select
    from steward.models.audit import AuditEvent
    async with app.db_sessionmaker() as s:
        rows = (await s.execute(select(AuditEvent).where(AuditEvent.user_id == uid)
                                .order_by(AuditEvent.timestamp))).scalars().all()
    return [(r.action, json.loads(r.detail_json)) for r in rows]


def test_test_webhook_sends_the_form_template_and_shows_the_body(app, monkeypatch):
    from steward.core import notifications
    sent = []

    async def fake_send(cfg, alert):
        sent.append((cfg, alert))
        return {"sent": False, "error": "Client error '404 Not Found' for url '<webhook URL>'"}

    monkeypatch.setattr(notifications, "_send_webhook", fake_send)

    async def _go():
        client, uid = await _admin_client(app)
        csrf = {"X-CSRF-Token": "test-token"}
        url = "https://hook.example/api/webhooks/1/s3cr3t"
        resp = await client.post("/settings/notifications/test-webhook", form={
            "webhook.url": url, "webhook.template": '{"text": "{{ alert.rule_name }}"}'}, headers=csrf)
        assert resp.status_code == 200
        page = await resp.get_data(as_text=True)
        assert "alert-error" in page and "404 Not Found" in page
        # The body shown is the one that would go out, and the quoted name kept it JSON.
        assert "Test alert \\&#34;from\\&#34; Steward" in page
        ((cfg, alert),) = sent
        assert cfg["url"] == url and alert["rule_name"] == 'Test alert "from" Steward'

        # A template that cannot make JSON is refused before anything is sent.
        resp = await client.post("/settings/notifications/test-webhook", form={
            "webhook.url": url, "webhook.template": '{"text": {{ raw.rule_name }}}'}, headers=csrf)
        page = await resp.get_data(as_text=True)
        assert "not valid JSON" in page and len(sent) == 1

        audits = await _audits(app, uid)
        assert [a for a, _ in audits] == ["settings.notification.tested"] * 2
        assert all(d == {"channel": "webhook", "sent": False} for _, d in audits)
        assert "s3cr3t" not in json.dumps([d for _, d in audits])

    asyncio.run(_go())


def test_test_email_uses_the_form_settings_and_names_the_tls_mode(app, monkeypatch):
    from steward.core import notifications
    sent = []

    async def fake_send(cfg, alert, to):
        sent.append((cfg, [r.address for r in to]))
        return {"sent": True, "error": None}

    monkeypatch.setattr(notifications, "_send_email", fake_send)

    async def _go():
        client, uid = await _admin_client(app)
        csrf = {"X-CSRF-Token": "test-token"}
        resp = await client.post("/settings/notifications/test-email", form={
            "smtp.host": "mx.example", "smtp.port": "465", "smtp.username": "u",
            "smtp.password": "pw", "smtp.from": "alerts@example.com",
            "smtp.recipients": "a@example.com, b@example.com"}, headers=csrf)
        page = await resp.get_data(as_text=True)
        assert "alert-success" in page
        assert "Sent to a@example.com, b@example.com via mx.example:465 (implicit TLS)." in page
        assert "Rule: Test alert" in page
        # The admin's own address (uuid@test) isn't one mail can reach, so only the typed ones.
        assert sent == [({"host": "mx.example", "port": 465, "tls": False, "username": "u",
                          "password": "pw", "from": "alerts@example.com", "base_url": ""},
                         ["a@example.com", "b@example.com"])]

        resp = await client.post("/settings/notifications/test-email", form={
            "smtp.host": "", "smtp.port": "587", "smtp.recipients": ""}, headers=csrf)
        assert "Enter an SMTP host." in await resp.get_data(as_text=True)
        assert len(sent) == 1

        assert [d for _, d in await _audits(app, uid)] == [
            {"channel": "email", "sent": True}, {"channel": "email", "sent": False}]

    asyncio.run(_go())