"""Integration: alert email through the outbox (#5507), against Postgres. Alerts that fire within 2 minutes of the first join one held email per reader; when the window closes it is written as one message listing them, oldest first; each alert's event records the outcome once every email carrying it has settled, naming a reader it didn't reach. Only the SMTP conversation is replaced. """ from __future__ import annotations import asyncio import os import smtplib import uuid from datetime import datetime, timedelta, timezone 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) def _alert(host: str, at: datetime) -> dict: return {"rule_name": f"Agent offline: {host}", "state": "FIRING", "metric": "seconds_since_seen", "value": 400.0, "operator": ">", "threshold": 180.0, "resource": host, "source_module": "host_agent", "note": None, "timestamp": at.isoformat()} async def _events(app, tag: str, times: list[datetime]) -> list[str]: from steward.alerts import service as alert_svc from steward.core.alerts import record_metrics from steward.models.alerts import AlertEvent from steward.models.metrics import PluginMetric from steward.models.users import User, UserRole res = f"box-{tag}" async with app.db_sessionmaker() as s: async with s.begin(): user = User(username=f"ob-{tag}", email=f"ob-{tag}@test", password_hash="x", role=UserRole.admin) s.add(user) await record_metrics(s, [PluginMetric(source_module="host_agent", resource_name=res, metric_name="cpu_pct", value=1.0)]) await s.flush() fields = await alert_svc.rule_fields( s, name=f"hot {tag}", source_module="host_agent", resource_name=res, metric_name="cpu_pct", operator=">", threshold=90) rule = await alert_svc.add_rule(s, fields, created_by=user.id) events = [AlertEvent(rule_id=rule.id, from_state="inactive", to_state="firing", metric_value=95.0, transitioned_at=at) for at in times] s.add_all(events) await s.flush() return [e.id for e in events] async def _emails_for(app, event_ids: list[str]): from sqlalchemy import select from steward.models.outbox import EmailOutbox, email_outbox_events as link async with app.db_sessionmaker() as s: rows = (await s.execute( select(EmailOutbox).where(EmailOutbox.id.in_( select(link.c.outbox_id).where(link.c.event_id.in_(event_ids)))))).scalars().all() links = (await s.execute(select(link.c.outbox_id, link.c.event_id) .where(link.c.event_id.in_(event_ids)))).all() return rows, links async def _event(app, event_id): from steward.models.alerts import AlertEvent async with app.db_sessionmaker() as s: return await s.get(AlertEvent, event_id) def test_alerts_in_one_window_are_one_email_and_each_event_records_it(app, monkeypatch): from steward.core import outbox tag = uuid.uuid4().hex[:8] extra = f"oncall-{tag}@example.com" app.config["SMTP"] = {"host": "mx.test", "port": 587, "from": "steward@test", "recipients": [extra], "tz": "America/New_York"} t0 = datetime(2026, 10, 9, 12, 0, tzinfo=timezone.utc) sent = [] def fake_deliver(cfg, messages): sent.extend(messages) return [smtplib.SMTPRecipientsRefused({m["To"]: (550, b"no such user")}) if m["To"] == extra and "web03" in m["Subject"] else None for m in messages] monkeypatch.setattr(outbox, "deliver_sync", fake_deliver) async def _go(): ev1, ev2, ev3 = await _events(app, tag, [t0, t0 + timedelta(seconds=30), t0 + timedelta(minutes=5)]) assert await outbox.hold_alert(app, _alert("web01", t0), ev1, now=t0) >= 1 await outbox.hold_alert(app, _alert("web02", t0), ev2, now=t0 + timedelta(seconds=30)) rows, links = await _emails_for(app, [ev1, ev2]) mine = [r for r in rows if r.recipient == extra] assert len(mine) == 1, "both alerts joined the one held email" (held,) = mine assert held.state == "held" and held.subject is None assert held.next_attempt_at == t0 + timedelta(minutes=2) assert {e for o, e in links if o == held.id} == {ev1, ev2} # Each reader has exactly one email carrying the pair. assert len({r.recipient for r in rows}) == len(rows) # The window closes: written, sent, and recorded on both events. await outbox.attempt(app, [r.id for r in rows], now=t0 + timedelta(minutes=2)) (msg,) = [m for m in sent if m["To"] == extra] assert msg["Subject"] == "[Steward] 2 alerts: 2 firing" body = msg.get_body(("plain",)).get_content() assert body.index("web01") < body.index("web02") assert "Fri 9 Oct 2026 08:00 EDT" in body # the Also-send-to zone for ev in (ev1, ev2): event = await _event(app, ev) assert (event.email_sent, event.email_error) == (True, None) # A later alert is its own email; refused for one reader, it is # recorded as not sent, naming who and why. await outbox.hold_alert(app, _alert("web03", t0), ev3, now=t0 + timedelta(minutes=5)) rows, _ = await _emails_for(app, [ev3]) assert all(r.state == "held" for r in rows) await outbox.attempt(app, [r.id for r in rows], now=t0 + timedelta(minutes=7)) (refused,) = [m for m in sent if m["To"] == extra and "web03" in m["Subject"]] assert refused["Subject"].startswith("[Steward]") # one alert: its usual email event = await _event(app, ev3) assert event.email_sent is False assert event.email_error == f"{extra}: 550 no such user" asyncio.run(_go()) def test_an_email_whose_alerts_were_deleted_fails_and_says_so(app, monkeypatch): from sqlalchemy import delete from steward.core import outbox from steward.models.alerts import AlertEvent tag = uuid.uuid4().hex[:8] extra = f"gone-{tag}@example.com" app.config["SMTP"] = {"host": "mx.test", "port": 587, "from": "steward@test", "recipients": [extra]} monkeypatch.setattr(outbox, "deliver_sync", lambda cfg, msgs: [None] * len(msgs)) t0 = datetime.now(timezone.utc) async def _go(): (ev,) = await _events(app, tag, [t0]) await outbox.hold_alert(app, _alert("web09", t0), ev, now=t0) rows, _ = await _emails_for(app, [ev]) (row,) = [r for r in rows if r.recipient == extra] async with app.db_sessionmaker() as s: async with s.begin(): await s.execute(delete(AlertEvent).where(AlertEvent.id == ev)) done = await outbox.attempt(app, [row.id]) return done[row.id] # One event loop per test: the app's pool is bound to the loop it ran in. row = asyncio.run(_go()) assert (row.state, row.last_error) == ("failed", "The alerts this email carried were deleted")