FabledSteward/tests/core/test_mail.py

"""Emails Steward sends read as designed messages (#5335): an HTML part and
a plain-text twin carrying the same facts, names escaped in the HTML, times
in the saved zone beside UTC, and nothing fetched when the mail is opened."""
import re
from datetime import datetime, timezone
from zoneinfo import ZoneInfo

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

ALERT = {"rule_name": 'Disk <b>"root"</b> on web01', "state": "FIRING", "metric": "disk_used_pct",
         "value": 93.5, "operator": ">=", "threshold": 90.0, "resource": "web01",
         "source_module": "host_agent", "note": None, "timestamp": "2026-10-08T17:58:00+00:00"}
CFG = {"host": "mx", "port": 587, "from": "alerts@example.com", "base_url": "https://steward.example/"}
NY = ZoneInfo("America/New_York")
EXTRA_NY = mail.Recipient("ops@example.com", NY, True)     # an Also send to address: UTC beside
PERSON_NY = mail.Recipient("ann@example.com", NY, False, "ann")
REPORT = {"generated_at": datetime(2026, 10, 11, 12, 0, tzinfo=timezone.utc),
          "uptime_summary": [{"name": "web01", "pct_7d": 99.95}, {"name": "nas <x>", "pct_7d": None}],
          "active_alerts": [{"state": "firing", "name": "Disk root", "resource": "web01",
                             "condition": "Disk used >= 90%"}],
          "alert_events": {"fired": 3, "resolved": 1},
          "top_routers": [{"name": "api@docker", "avg_rate": 12.345}], "hosts_down": ["nas"]}

_URL = re.compile(r"""https?://[^\s"'<>)]+""")
_FETCHES = re.compile(r"""\ssrc\s*=|url\(|<link\b|@import""", re.I)


def foreign_or_fetched(html: str, base: str) -> list[str]:
    """What in an email's HTML would reach outside Steward, or load at all."""
    return ([u for u in _URL.findall(html) if not u.startswith(base)]
            + _FETCHES.findall(html))


def test_the_asset_guard_can_fail():
    assert foreign_or_fetched('<img src="https://cdn.example/x.png">', "https://steward.example")
    assert foreign_or_fetched('<td style="background:url(x.png)">', "https://steward.example")
    assert not foreign_or_fetched('<a href="https://steward.example/alerts/">', "https://steward.example")


def test_every_colour_is_a_theme_value():
    colours = {k: v for k, v in mail.palette().items() if k != "font"}
    assert set(colours) >= {"page", "card", "text", "muted", "action", "on_action", "ok", "warn", "crit"}
    assert all(re.fullmatch(r"#[0-9A-Fa-f]{6}", v) for v in colours.values()), colours


async def _sent(monkeypatch, cfg, alert, to=(EXTRA_NY,)):
    out = []
    monkeypatch.setattr(n, "_smtp_send_sync", lambda c, msgs: out.extend(msgs))
    assert (await n._send_email(cfg, alert, list(to)))["sent"]
    return out[0] if len(to) == 1 else out


async def test_an_alert_email_is_html_with_a_plain_text_twin(monkeypatch):
    msg = await _sent(monkeypatch, CFG, ALERT)
    assert msg.get_content_type() == "multipart/alternative"
    text, html = (msg.get_body(("plain",)).get_content(), msg.get_body(("html",)).get_content())
    for part in (text, html):
        for fact in ("Disk used on web01 is 94%.", "Disk used &gt;= 90%" if part is html else "Disk used >= 90%",
                     "Thu 8 Oct 2026 13:58 EDT", "Thu 8 Oct 2026 17:58 UTC", "https://steward.example/alerts/"):
            assert fact in part, (fact, part)
    assert msg["Subject"] == '[Steward] Firing: Disk <b>"root"</b> on web01 — web01 94%'
    assert (msg["From"], msg["To"]) == ("alerts@example.com", "ops@example.com")


async def test_each_recipient_gets_their_own_copy_in_their_own_zone(monkeypatch):
    berlin = mail.Recipient("bo@example.com", ZoneInfo("Europe/Berlin"), False, "bo")
    ann, bo = await _sent(monkeypatch, CFG, ALERT, to=(PERSON_NY, berlin))
    assert (ann["To"], bo["To"]) == ("ann@example.com", "bo@example.com")
    ann_text, bo_text = (m.get_body(("plain",)).get_content() for m in (ann, bo))
    assert "13:58 EDT" in ann_text and "19:58 CEST" in bo_text
    # A person whose zone is known is never shown UTC beside it.
    assert "UTC" not in ann_text + bo_text


async def test_a_name_with_markup_is_escaped_in_the_html(monkeypatch):
    msg = await _sent(monkeypatch, CFG, ALERT)
    html = msg.get_body(("html",)).get_content()
    assert "<b>" not in html and "&lt;b&gt;&#34;root&#34;&lt;/b&gt;" in html
    assert 'Disk <b>"root"</b> on web01' in msg.get_body(("plain",)).get_content()


async def test_opening_an_email_fetches_nothing(monkeypatch):
    msg = await _sent(monkeypatch, CFG, ALERT)
    assert not foreign_or_fetched(msg.get_body(("html",)).get_content(), "https://steward.example")
    _, _, html = mail.report_email(CFG, REPORT, mail.Recipient("x@example.com", timezone.utc, False))
    assert not foreign_or_fetched(html, "https://steward.example")


def test_without_steward_s_address_there_is_no_link_and_utc_is_said_once():
    _, text, html = mail.alert_email({}, ALERT, mail.Recipient("x@example.com", timezone.utc, True))
    assert "View in Steward" not in text + html and "http" not in text + html
    assert text.count("17:58 UTC") == 1


def test_each_state_reads_as_a_sentence():
    head = {s: mail.alert_facts({**ALERT, "state": s, "value": v}, timezone.utc)["headline"]
            for s, v in (("FIRING", 93.5), ("RESOLVED", 71.0), ("NO_DATA", None), ("DATA_RESUMED", 80.0),
                         ("TEST", None))}
    assert head == {"FIRING": "Disk used on web01 is 94%.",
                    "RESOLVED": "Disk used on web01 is back within its limit, at 71%.",
                    "NO_DATA": "Steward has had no disk used reading from web01.",
                    "DATA_RESUMED": "web01 is reporting disk used again.",
                    "TEST": "Disk used on web01 has no reading yet."}
    down = mail.alert_facts({**ALERT, "metric": "is_up", "operator": "<", "threshold": 0.5, "value": 0.0},
                            timezone.utc)
    assert (down["headline"], down["condition"]) == ("web01 is down.", "is down")


def test_the_report_has_its_sections_in_both_parts():
    subject, text, html = mail.report_email(CFG, REPORT, PERSON_NY)
    assert subject == "[Steward] Weekly report, 4 Oct – 11 Oct 2026"
    for part in (text, html):
        for fact in ("1 monitor is down now", "Disk used &gt;= 90%" if part is html else "Disk used >= 90%",
                     "99.95%", "—", "12.35 req/s", "3 alerts fired and 1 resolved",
                     "Sun 11 Oct 2026 08:00 EDT", "https://steward.example/monitors/"):
            assert fact in part, (fact, part)
    assert "nas &lt;x&gt;" in html and "nas <x>" in text
    for stale in ("Generated", "bvandeusen/Steward"):
        assert stale not in text + html


async def test_the_report_is_sent_from_the_configured_address(monkeypatch):
    class _Session:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *exc):
            return False

    class _App:
        config = {"SMTP": {**CFG, "username": "login@example.com"}}
        db_sessionmaker = staticmethod(_Session)

    async def _data(app):
        return REPORT

    async def _recipients(db, kind, smtp_cfg):
        assert kind == "report"
        return [PERSON_NY, EXTRA_NY]

    out = []
    monkeypatch.setattr(reports, "build_report_data", _data)
    monkeypatch.setattr(reports, "recipients", _recipients)
    monkeypatch.setattr(n, "_smtp_send_sync", lambda c, msgs: out.extend(msgs))
    ok, said = await reports.send_report(_App())
    assert ok and said == "Report sent to ann@example.com, ops@example.com."
    assert [(m["From"], m["To"]) for m in out] == [("alerts@example.com", "ann@example.com"),
                                                   ("alerts@example.com", "ops@example.com")]
    assert all(m.get_content_type() == "multipart/alternative" for m in out)


async def test_nobody_to_send_the_report_to_is_said(monkeypatch):
    class _App:
        config = {"SMTP": CFG}

        @staticmethod
        def db_sessionmaker():
            class _S:
                async def __aenter__(self):
                    return self

                async def __aexit__(self, *exc):
                    return False
            return _S()

    async def _nobody(db, kind, smtp_cfg):
        return []

    monkeypatch.setattr(reports, "recipients", _nobody)
    ok, said = await reports.send_report(_App())
    assert not ok and said.startswith("Nobody gets the weekly report")


# ── Who gets an email (milestone 495) ─────────────────────────────────────────

def _user(name, role="viewer", email=None, tz=None, active=True, alerts=None, report=None):
    from steward.models.users import User, UserRole
    return User(username=name, email=email if email is not None else f"{name}@example.com",
                password_hash="x", role=UserRole(role), is_active=active, time_zone=tz,
                notify_alerts=alerts, notify_report=report)


def test_each_account_follows_its_own_choice_or_else_its_role():
    settings = {"notify.alerts.viewer": False, "notify.alerts.admin": True}
    users = [_user("adm", "admin"),                        # role on
             _user("vic"),                                 # role off
             _user("val", alerts=True),                    # role off, chose on
             _user("ada", "admin", alerts=False),          # role on, chose off
             _user("gone", "admin", active=False),         # disabled
             _user("sso", "admin", email="sso@oidc")]      # no address mail can reach
    got = mail.pick_recipients(users, "alerts", settings, [], None)
    assert [r.user for r in got] == ["adm", "val"]
    # A role with no setting yet gets email: the defaults start on.
    assert [r.user for r in mail.pick_recipients([_user("op", "operator")], "report", {}, [], None)] == ["op"]


def test_a_person_reads_their_own_zone_and_extra_addresses_read_the_chosen_one():
    users = [_user("ann", "admin", tz="America/New_York"), _user("utc", "admin")]
    got = mail.pick_recipients(users, "alerts", {}, ["ops@example.com", "ANN@example.com", "nonsense"],
                               "Europe/Berlin")
    assert [(r.address, str(r.tz), r.show_utc) for r in got] == [
        ("ann@example.com", "America/New_York", False),
        ("utc@example.com", "UTC", False),
        ("ops@example.com", "Europe/Berlin", True)]      # ANN@ is ann, once; "nonsense" is no address


def test_the_email_form_changes_only_what_it_says_and_nothing_on_a_bad_value():
    u = _user("ann", alerts=None, report=True)
    changes, error = mail.apply_email_form(u, {"email": "new@example.com", "notify_alerts": "off",
                                               "notify_report": "on"})
    assert error is None and changes == {"email": ["ann@example.com", "new@example.com"],
                                         "notify_alerts": [None, False]}
    assert (u.email, u.notify_alerts, u.notify_report) == ("new@example.com", False, True)
    for bad in ({"email": "ann@oidc", "notify_alerts": "", "notify_report": ""},
                {"notify_alerts": "maybe", "notify_report": ""}):
        changes, error = mail.apply_email_form(u, bad)
        assert changes == {} and error
    assert (u.email, u.notify_alerts) == ("new@example.com", False)
    # Without an address in the form (an admin's view), the address is left alone.
    assert mail.apply_email_form(u, {"notify_alerts": "", "notify_report": ""})[0] == {
        "notify_alerts": [False, None], "notify_report": [True, None]}