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

import httpx
import pytest

from steward.core import notifications as n

ALERT = {"rule_name": 'Disk "root" on\nweb01 \\ backslash', "state": "FIRING", "metric": "disk_used_pct",
         "value": 93.5, "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_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_send_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):
    monkeypatch.setattr(n.smtplib, "SMTP", SMTP)
    monkeypatch.setattr(n.smtplib, "SMTP_SSL", SMTP_SSL)
    _FakeSMTP.made = []
    msg = n.EmailMessage()
    msg["From"] = "steward@example.com"
    n._smtp_send_sync({"host": "mx", "port": port, "tls": tls, "username": "u", "password": "p"}, msg)
    (smtp,) = _FakeSMTP.made
    assert smtp.calls == expected + [("login", "u", "p"), ("send", "steward@example.com")]


async def test_email_from_falls_back_and_a_line_break_in_the_name_still_sends(monkeypatch):
    sent = []
    monkeypatch.setattr(n, "_smtp_send_sync", lambda cfg, msg: sent.append((msg["From"], msg["Subject"])))
    base = {"host": "mx", "port": 587, "recipients": ["a@example.com"]}
    for cfg in ({**base, "from": "alerts@example.com", "username": "u@example.com"},
                {**base, "from": "", "username": "u@example.com"},
                {**base, "from": "", "username": ""}):
        assert (await n._send_email(cfg, ALERT))["sent"]
    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'


def test_email_body_names_no_data_and_the_note():
    body = n.email_body({**ALERT, "value": None, "note": "no sample for 18m"})
    assert "Metric: disk_used_pct = no data\n" in body and "Note: no sample for 18m\n" in body
    assert "Note:" not in n.email_body(ALERT)


# ── _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"]}'}]