bvandeusenandClaude Opus 5.5
5237ffc663
#5507 "Alert emails through the outbox, gathered for 2 minutes" (milestone 501). Replaces the in-memory core/alert_burst.py. - outbox.hold_alert: an alert joins each recipient's held alert email, or opens one due 2 minutes later (System 156 ruling). A Postgres advisory lock keeps two alerts at once from opening two windows for one reader. - email_outbox_events.alert_json (migration 0043) keeps the alert as it fired, so the email written when the window closes says what it said. - write_held writes it then, in the reader's zone: the usual alert email for one alert, the list for several (outbox.alert_words). - after_settle records on each AlertEvent, once every email carrying it has settled: email_sent, or email_error naming each reader it didn't reach and the server's reason. - A held email survives a restart and goes out when its window closes. - A refusal of a message's one recipient no longer repeats the address. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
109 lines
4.4 KiB
Python
109 lines
4.4 KiB
Python
"""Integration: the email outbox's sender against Postgres (#5506).
|
|
|
|
A due row is sent on the sender's first tick, a refusal fails it, a
|
|
temporary failure leaves it waiting for its retry, a held row waits for its
|
|
window, and a claim left by a run that died goes back to the queue. Only the
|
|
SMTP conversation itself 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
|
|
app = create_app(testing=False)
|
|
app.config["SMTP"] = {"host": "mx.test", "port": 587, "from": "steward@test"}
|
|
return app
|
|
|
|
|
|
def test_the_sender_settles_each_row_on_what_the_server_said(app, monkeypatch):
|
|
from sqlalchemy import select
|
|
from steward.core import outbox
|
|
from steward.models.outbox import EmailOutbox
|
|
|
|
tag = uuid.uuid4().hex[:8]
|
|
now = datetime.now(timezone.utc)
|
|
to = {k: f"{k}-{tag}@test" for k in ("ok", "refused", "later", "held", "stale")}
|
|
tried: list[str] = []
|
|
|
|
def fake_deliver(cfg, messages):
|
|
out = []
|
|
for m in messages:
|
|
tried.append(m["To"])
|
|
if m["To"] == to["refused"]:
|
|
out.append(smtplib.SMTPRecipientsRefused({m["To"]: (550, b"no such user")}))
|
|
elif m["To"] == to["later"]:
|
|
out.append(smtplib.SMTPDataError(451, b"greylisted"))
|
|
else:
|
|
out.append(None)
|
|
return out
|
|
|
|
monkeypatch.setattr(outbox, "deliver_sync", fake_deliver)
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
for k in ("ok", "refused", "later"):
|
|
outbox.enqueue(s, kind="report", to=to[k], subject=f"s {k}", text="t", html="<p>t</p>")
|
|
outbox.enqueue(s, kind="report", to=to["held"], subject="s", text="t",
|
|
send_after=now + timedelta(minutes=2))
|
|
# Claimed by a sender that died ten minutes ago.
|
|
dead = outbox.enqueue(s, kind="report", to=to["stale"], subject="s", text="t")
|
|
dead.state, dead.claimed_at = "sending", now - timedelta(minutes=10)
|
|
await outbox.send_due(app)
|
|
async with app.db_sessionmaker() as s:
|
|
rows = (await s.execute(select(EmailOutbox)
|
|
.where(EmailOutbox.recipient.in_(to.values())))).scalars().all()
|
|
return {r.recipient: r for r in rows}
|
|
|
|
rows = asyncio.run(_go())
|
|
by = {k: rows[to[k]] for k in to}
|
|
assert by["ok"].state == "sent" and by["ok"].sent_at is not None and by["ok"].attempts == 1
|
|
assert by["refused"].state == "failed"
|
|
assert by["refused"].last_error == "550 no such user"
|
|
later = by["later"]
|
|
assert later.state == "pending" and later.attempts == 1 and later.last_error == "451 greylisted"
|
|
assert later.next_attempt_at > later.first_attempt_at
|
|
# The held row is not due for two minutes; nothing went to it.
|
|
assert by["held"].state == "held" and by["held"].attempts == 0 and to["held"] not in tried
|
|
# The dead run's claim was released and the row sent in the same tick.
|
|
assert by["stale"].state == "sent" and by["stale"].claimed_at is None
|
|
assert sorted(t for t in tried if t.endswith(f"{tag}@test")) == sorted(
|
|
to[k] for k in ("ok", "refused", "later", "stale"))
|
|
|
|
|
|
def test_with_no_mail_server_a_row_fails_and_says_where_to_set_one(app, monkeypatch):
|
|
from sqlalchemy import select
|
|
from steward.core import outbox
|
|
from steward.models.outbox import EmailOutbox
|
|
|
|
app.config["SMTP"] = {}
|
|
monkeypatch.setattr(outbox, "deliver_sync", lambda cfg, msgs: pytest.fail("nothing to send to"))
|
|
addr = f"nomx-{uuid.uuid4().hex[:8]}@test"
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
row = outbox.enqueue(s, kind="report", to=addr, subject="s", text="t")
|
|
await s.flush()
|
|
rid = row.id
|
|
await outbox.attempt(app, [rid])
|
|
async with app.db_sessionmaker() as s:
|
|
return (await s.execute(select(EmailOutbox).where(EmailOutbox.recipient == addr))).scalar_one()
|
|
|
|
row = asyncio.run(_go())
|
|
assert row.state == "failed"
|
|
assert row.last_error == "No mail server is set (Settings → Notifications)"
|