"""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 "root" 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, "recipients": ["a@example.com"], "from": "alerts@example.com", "tz": "America/New_York", "base_url": "https://steward.example/"} REPORT = {"generated_at": datetime(2026, 10, 11, 12, 0, tzinfo=timezone.utc), "uptime_summary": [{"name": "web01", "pct_7d": 99.95}, {"name": "nas ", "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\(| 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('', "https://steward.example") assert foreign_or_fetched('', "https://steward.example") assert not foreign_or_fetched('', "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): out = [] monkeypatch.setattr(n, "_smtp_send_sync", lambda c, msg: out.append(msg)) assert (await n._send_email(cfg, alert))["sent"] return out[0] 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 >= 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 "root" on web01 — web01 94%' assert msg["From"] == "alerts@example.com" 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 "" not in html and "<b>"root"</b>" in html assert 'Disk "root" 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, ZoneInfo("UTC")) 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({"recipients": []}, ALERT) 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))} 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."} 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, ZoneInfo("America/New_York")) 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 >= 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 <x>" in html and "nas " 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 _setting(db, key): return "America/New_York" out = [] monkeypatch.setattr(reports, "build_report_data", _data) monkeypatch.setattr(reports, "get_setting", _setting) monkeypatch.setattr(n, "_smtp_send_sync", lambda c, msg: out.append(msg)) ok, _ = await reports.send_report(_App()) assert ok and out[0]["From"] == "alerts@example.com" assert out[0].get_content_type() == "multipart/alternative"