FabledSteward/tests/core/test_notifications.py

"""Notification delivery (#5064): the webhook body is JSON whatever an alert
holds, SMTP picks implicit TLS / STARTTLS / plain from port and box and never
waits without a timeout, and a webhook error never carries the URL."""
import json
from datetime import timezone

import httpx
import pytest

from steward.core import notifications as n
from steward.core.mail import Recipient

ALERT = {"rule_name": 'Disk "root" on\nweb01 \\ backslash', "state": "FIRING", "metric": "disk_used_pct",
         "value": 93.5, "operator": ">=", "threshold": 90.0, "resource": "web01", "source_module": "host_agent",
         "note": None, "timestamp": "2026-10-06T12:00:00+00:00"}


# ── render_webhook ────────────────────────────────────────────────────────────

def test_alert_fields_are_escaped_into_json_strings():
    body = n.render_webhook('{"content": "{{ alert.state }} — {{ alert.rule_name }}"}', ALERT)
    assert json.loads(body) == {"content": 'FIRING — Disk "root" on\nweb01 \\ backslash'}


def test_the_default_template_survives_quotes_in_a_name():
    from steward.core.settings import DEFAULTS
    body = json.loads(n.render_webhook(DEFAULTS["webhook.template"], ALERT))
    assert 'Disk "root" on\nweb01' in body["content"] and "93.5" in body["content"]


def test_raw_fields_give_bare_json_values():
    tpl = '{"value": {{ raw.value | tojson }}, "name": {{ raw.rule_name | tojson }}}'
    assert json.loads(n.render_webhook(tpl, ALERT)) == {"value": 93.5, "name": ALERT["rule_name"]}
    assert json.loads(n.render_webhook(tpl, {**ALERT, "value": None}))["value"] is None


def test_the_names_behind_a_count_read_as_a_list_or_as_an_array():
    alert = {**ALERT, "which": ["a (0 of 1 running)", 'b "q"']}
    body = json.loads(n.render_webhook('{"w": "{{ alert.which }}", "raw": {{ raw.which | tojson }}}', alert))
    assert body == {"w": 'a (0 of 1 running), b "q"', "raw": ["a (0 of 1 running)", 'b "q"']}


def test_a_missing_value_reads_no_data():
    body = n.render_webhook('{"v": "{{ alert.value }}", "note": "{{ alert.note }}"}', {**ALERT, "value": None})
    assert json.loads(body) == {"v": "no data", "note": ""}


@pytest.mark.parametrize("tpl, says", [
    ('{"v": {{ alert.value }', "Template render failed"),       # Jinja syntax error
    ('{"v": {{ raw.rule_name }}}', "not valid JSON"),            # unquoted string
    ("{{ raw.__class__.__mro__ }}", "Template render failed"),   # sandbox refuses
])
def test_a_bad_template_raises_valueerror_saying_why(tpl, says):
    with pytest.raises(ValueError, match=says):
        n.render_webhook(tpl, ALERT)


# ── smtp_mode / smtp_open (via outbox.deliver_sync) ───────────────────────────────────────────────

@pytest.mark.parametrize("port, tls, mode", [
    (465, True, "ssl"), (465, False, "ssl"), ("465", False, "ssl"),
    (587, True, "starttls"), (587, False, "plain"),
    (25, True, "starttls"), (25, False, "plain"),
    (None, True, "starttls"), (None, False, "plain"),
])
def test_smtp_mode(port, tls, mode):
    assert n.smtp_mode(port, tls) == mode


class _FakeSMTP:
    made: list = []

    def __init__(self, host, port, timeout=None, context=None):
        self.calls = [("connect", type(self).__name__, host, port, timeout)]
        _FakeSMTP.made.append(self)

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False

    def starttls(self, context=None):
        self.calls.append(("starttls",))

    def login(self, user, password):
        self.calls.append(("login", user, password))

    def send_message(self, msg):
        self.calls.append(("send", msg["From"]))


class SMTP(_FakeSMTP):
    pass


class SMTP_SSL(_FakeSMTP):
    pass


@pytest.mark.parametrize("port, tls, expected", [
    (465, True, [("connect", "SMTP_SSL", "mx", 465, n.SMTP_TIMEOUT_S)]),
    (587, True, [("connect", "SMTP", "mx", 587, n.SMTP_TIMEOUT_S), ("starttls",)]),
    (25, False, [("connect", "SMTP", "mx", 25, n.SMTP_TIMEOUT_S)]),
])
def test_smtp_send_connects_by_mode_with_a_timeout(monkeypatch, port, tls, expected):
    from email.message import EmailMessage
    from steward.core import outbox
    monkeypatch.setattr(n.smtplib, "SMTP", SMTP)
    monkeypatch.setattr(n.smtplib, "SMTP_SSL", SMTP_SSL)
    _FakeSMTP.made = []
    msgs = [EmailMessage(), EmailMessage()]
    msgs[0]["From"], msgs[1]["From"] = "steward@example.com", "other@example.com"
    cfg = {"host": "mx", "port": port, "tls": tls, "username": "u", "password": "p"}
    assert outbox.deliver_sync(cfg, msgs) == [None, None]
    (smtp,) = _FakeSMTP.made  # every message over one connection
    assert smtp.calls == expected + [("login", "u", "p"), ("send", "steward@example.com"),
                                     ("send", "other@example.com")]


def test_email_from_falls_back_and_a_line_break_in_the_name_still_sends():
    from steward.core.mail import alert_email, message
    sent = []
    to = Recipient("a@example.com", timezone.utc, False)
    base = {"host": "mx", "port": 587}
    for cfg in ({**base, "from": "alerts@example.com", "username": "u@example.com"},
                {**base, "from": "", "username": "u@example.com"},
                {**base, "from": "", "username": ""}):
        m = message(cfg, to.address, *alert_email(cfg, ALERT, to))
        sent.append((m["From"], m["Subject"]))
    assert [f for f, _ in sent] == ["alerts@example.com", "u@example.com", "steward@localhost"]
    # The rule name has a line break; the subject folds it to a space rather than failing.
    assert sent[0][1] == '[Steward] Firing: Disk "root" on web01 \\ backslash — web01 94%'


# ── _send_webhook ─────────────────────────────────────────────────────────────

def _client_with(monkeypatch, handler):
    real = httpx.AsyncClient
    monkeypatch.setattr(n.httpx, "AsyncClient",
                        lambda **kw: real(transport=httpx.MockTransport(handler), **kw))


async def test_a_webhook_error_never_carries_the_url(monkeypatch):
    url = "https://discord.example/api/webhooks/123/s3cr3t-token"
    _client_with(monkeypatch, lambda req: httpx.Response(404))
    result = await n._send_webhook({"url": url, "template": ""}, ALERT)
    assert result["sent"] is False
    assert "s3cr3t-token" not in result["error"] and "<webhook URL>" in result["error"]


async def test_a_webhook_posts_the_rendered_json(monkeypatch):
    seen = []

    def handler(req):
        seen.append(json.loads(req.content))
        return httpx.Response(204)

    _client_with(monkeypatch, handler)
    result = await n._send_webhook({"url": "https://hook.example/x", "template": ""}, ALERT)
    assert result == {"sent": True, "error": None}
    assert seen == [{"content": f'FIRING — {ALERT["rule_name"]}'}]