v0.18.0 Slice 2: migrate send paths to build_envelope + POST unsubscribe
Five send sites now construct their envelopes through
`email_envelope.build_envelope` instead of building `EmailMessage`
ad hoc:
* `email_otc.py` — OTC mail (no List-Unsubscribe per the
proposal's tradeoff: the recipient explicitly requested the
code, so a list semantic would be wrong).
* `email_invite.py` — admin/per-RFC invite mail (mailto:-only
List-Unsubscribe — the invitee isn't a user yet, so no
per-user opt-out URL exists).
* `email._deliver` — watcher notifications (full one-click
unsubscribe: mailto + signed URL + List-Unsubscribe-Post per
RFC 8058, required by Gmail/Yahoo).
* `email._send_bundle` — the "while you were away" bundle,
one-click to the new `all` synthetic category (which lands
`email_opt_out_all = 1` because the bundle spans multiple
per-category flags).
* `digest.py` — same as the bundle: bulk-adjacent, one-click to
`all`.
`api_notifications.py` gains the POST `/api/email/unsubscribe`
endpoint (the matching receiver for `List-Unsubscribe-Post:
List-Unsubscribe=One-Click`) and accepts the `all` category in
both GET and POST handlers.
`EmailConfig` adds `unsubscribe_mailto` (env: `EMAIL_UNSUBSCRIBE_MAILTO`,
default = `EMAIL_FROM`) so deployments can route unsubscribe
courtesy mail to a humans-monitored mailbox distinct from the
no-reply sender.
The `_SENT` test buffer now also carries `envelope["message"]`
(the `EmailMessage` itself) so new tests can assert on headers
directly. Legacy `to`/`from`/`subject`/`body` keys remain for
backward compatibility.
10 new integration tests across test_otc_vertical /
test_admin_create_user_invite_vertical / test_notifications_vertical
covering: OTC has no List-Unsubscribe; invite has mailto: only;
notification has full one-click; POST one-click flips the
category; `all` category sets global opt-out via both GET and
POST.
Full suite: 277 passed.
This commit is contained in:
@@ -443,6 +443,40 @@ def make_router(config: Config) -> APIRouter:
|
||||
|
||||
# ----- Email: one-click unsubscribe + bounce webhook -----
|
||||
|
||||
# v0.18.0: the category → column map. The `all` synthetic
|
||||
# category lands the bundle's one-click on the global opt-out
|
||||
# flag (per `email._send_bundle` in v0.18.0 Slice 2 — a bundle
|
||||
# spans multiple categories, so a per-category flip wouldn't
|
||||
# honor the user's intent).
|
||||
_CATEGORY_COLUMN: dict[str, str] = {
|
||||
"personal-direct": "email_personal_direct",
|
||||
"structural": "email_watched_structural",
|
||||
"admin-actionable": "email_admin_actionable",
|
||||
"all": "email_opt_out_all",
|
||||
}
|
||||
|
||||
def _apply_unsubscribe(user_id: int, category: str) -> bool:
|
||||
"""Flip the matching column. Returns True on success, False
|
||||
if the category is unknown. Idempotent — running twice on
|
||||
the same (user, category) is harmless (it sets the column
|
||||
to its current value)."""
|
||||
column = _CATEGORY_COLUMN.get(category)
|
||||
if column is None:
|
||||
return False
|
||||
# `all` sets the flag to 1 (opt out); per-category sets to 0
|
||||
# (turn that category off). The column semantic is "1 means
|
||||
# don't send"; the per-category booleans are "1 means do
|
||||
# send". Different polarities, hence the case split.
|
||||
if category == "all":
|
||||
db.conn().execute(
|
||||
f"UPDATE users SET {column} = 1 WHERE id = ?", (user_id,)
|
||||
)
|
||||
else:
|
||||
db.conn().execute(
|
||||
f"UPDATE users SET {column} = 0 WHERE id = ?", (user_id,)
|
||||
)
|
||||
return True
|
||||
|
||||
@router.get("/api/email/unsubscribe")
|
||||
async def email_unsubscribe(t: str = Query(..., description="Signed token from the email footer")) -> HTMLResponse:
|
||||
try:
|
||||
@@ -453,20 +487,51 @@ def make_router(config: Config) -> APIRouter:
|
||||
"<p>Open the app to manage your notification preferences directly.</p>",
|
||||
status_code=400,
|
||||
)
|
||||
column = {
|
||||
"personal-direct": "email_personal_direct",
|
||||
"structural": "email_watched_structural",
|
||||
"admin-actionable": "email_admin_actionable",
|
||||
}.get(category)
|
||||
if column is None:
|
||||
if not _apply_unsubscribe(user_id, category):
|
||||
return HTMLResponse(
|
||||
f"<h1>Unknown category</h1><p>{category}</p>", status_code=400
|
||||
)
|
||||
db.conn().execute(f"UPDATE users SET {column} = 0 WHERE id = ?", (user_id,))
|
||||
return HTMLResponse(
|
||||
f"<h1>Unsubscribed</h1><p>You will no longer receive {category} emails. "
|
||||
f"You can re-enable them in your notification preferences.</p>"
|
||||
)
|
||||
if category == "all":
|
||||
body = (
|
||||
"<h1>Unsubscribed</h1><p>You will no longer receive any email "
|
||||
"from this app. You can re-enable individual categories from "
|
||||
"your notification preferences after signing in.</p>"
|
||||
)
|
||||
else:
|
||||
body = (
|
||||
f"<h1>Unsubscribed</h1><p>You will no longer receive {category} emails. "
|
||||
f"You can re-enable them in your notification preferences.</p>"
|
||||
)
|
||||
return HTMLResponse(body)
|
||||
|
||||
@router.post("/api/email/unsubscribe")
|
||||
async def email_unsubscribe_post(
|
||||
request: Request,
|
||||
t: str = Query(..., description="Signed token from the List-Unsubscribe header"),
|
||||
) -> dict[str, Any]:
|
||||
"""v0.18.0: RFC 8058 one-click endpoint.
|
||||
|
||||
Gmail and Yahoo POST `List-Unsubscribe=One-Click` (as a
|
||||
form-encoded body) to the URL in the `List-Unsubscribe`
|
||||
header when the user clicks their MUA's "Unsubscribe"
|
||||
button. The endpoint MUST accept POST (per the
|
||||
`List-Unsubscribe-Post` header we advertise) and MUST be
|
||||
idempotent.
|
||||
|
||||
The body content is checked loosely — RFC 8058 says it
|
||||
SHOULD be exactly `List-Unsubscribe=One-Click`, but some
|
||||
intermediaries strip / re-encode the body, so the
|
||||
framework accepts any POST to the URL once the token
|
||||
verifies. The bar is that the token signature carries the
|
||||
authority; the body is hint-only.
|
||||
"""
|
||||
try:
|
||||
user_id, category = email_mod.verify_unsubscribe_token(t)
|
||||
except BadSignature:
|
||||
raise HTTPException(400, "Invalid or expired token")
|
||||
if not _apply_unsubscribe(user_id, category):
|
||||
raise HTTPException(400, f"Unknown category: {category}")
|
||||
return {"ok": True, "category": category}
|
||||
|
||||
@router.post("/api/webhooks/email-bounce")
|
||||
async def email_bounce(body: BounceBody, request: Request) -> dict[str, Any]:
|
||||
|
||||
+12
-1
@@ -180,7 +180,18 @@ def assemble_for_user(
|
||||
|
||||
subject = _subject(eligible, cadence)
|
||||
body = _body(eligible, cadence, cfg)
|
||||
sent = email_mod._deliver(cfg, email, subject, body)
|
||||
# v0.18.0: the digest is the bulk-adjacent surface par excellence
|
||||
# (it can carry weeks of accumulated activity), so it gets the
|
||||
# full one-click unsubscribe to the global opt-out. Per-category
|
||||
# opt-outs are managed from the preferences page; this footer is
|
||||
# the "stop sending me anything" escape hatch Gmail and Yahoo
|
||||
# expect for senders at this tier.
|
||||
unsubscribe_url = email_mod.make_unsubscribe_url(user_id, "all")
|
||||
sent = email_mod._deliver(
|
||||
cfg, email, subject, body,
|
||||
unsubscribe_mailto=cfg.unsubscribe_mailto,
|
||||
unsubscribe_url=unsubscribe_url,
|
||||
)
|
||||
if not sent:
|
||||
return False
|
||||
ids = [r["id"] for r, _ in eligible]
|
||||
|
||||
+75
-9
@@ -24,7 +24,6 @@ import os
|
||||
import smtplib
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, time, timezone
|
||||
from email.message import EmailMessage
|
||||
from email.utils import formataddr
|
||||
from itertools import groupby
|
||||
from typing import Any
|
||||
@@ -33,6 +32,7 @@ from urllib.parse import urlencode
|
||||
from itsdangerous import BadSignature, URLSafeSerializer
|
||||
|
||||
from . import db
|
||||
from .email_envelope import build_envelope
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
@@ -69,6 +69,7 @@ class EmailConfig:
|
||||
app_url: str
|
||||
bundle_threshold: int
|
||||
enabled: bool
|
||||
unsubscribe_mailto: str
|
||||
|
||||
@classmethod
|
||||
def from_env(cls) -> "EmailConfig":
|
||||
@@ -84,6 +85,16 @@ class EmailConfig:
|
||||
app_url=os.environ.get("APP_URL", "http://localhost:8000").rstrip("/"),
|
||||
bundle_threshold=int(os.environ.get("EMAIL_BUNDLE_THRESHOLD", "5")),
|
||||
enabled=os.environ.get("EMAIL_ENABLED", "1") not in ("0", "false", "False"),
|
||||
# v0.18.0: the `List-Unsubscribe: <mailto:…>` target on
|
||||
# invite + notification mail. Defaults to the From
|
||||
# address when unset; a deployment can route opt-out
|
||||
# mail to a separate mailbox (e.g., a humans-monitored
|
||||
# account distinct from the no-reply notifications
|
||||
# sender) by setting this explicitly.
|
||||
unsubscribe_mailto=os.environ.get(
|
||||
"EMAIL_UNSUBSCRIBE_MAILTO",
|
||||
os.environ.get("EMAIL_FROM", "notifications@wiggleverse.local"),
|
||||
).strip(),
|
||||
)
|
||||
|
||||
|
||||
@@ -98,6 +109,14 @@ def _signer() -> URLSafeSerializer:
|
||||
|
||||
|
||||
def make_unsubscribe_url(user_id: int, category: str) -> str:
|
||||
"""Build the §15.4 per-category one-click URL.
|
||||
|
||||
`category` is one of `personal-direct`, `structural`,
|
||||
`admin-actionable` (the three per-category flags) or `all`
|
||||
(v0.18.0: the bundle path, which sets `email_opt_out_all = 1`
|
||||
because a bundle covers multiple categories and a per-category
|
||||
opt-out wouldn't honor the user's intent).
|
||||
"""
|
||||
cfg = EmailConfig.from_env()
|
||||
token = _signer().dumps({"u": user_id, "c": category})
|
||||
qs = urlencode({"t": token})
|
||||
@@ -250,7 +269,19 @@ def _send_one(user: Any, notif_id: int, payload: dict, category: str) -> None:
|
||||
return
|
||||
subject = _subject(payload)
|
||||
body = _body(payload, user["id"], category, cfg)
|
||||
sent = _deliver(cfg, user["email"], subject, body)
|
||||
# v0.18.0: notification mail is bulk-adjacent (a watcher can
|
||||
# accumulate dozens of structural events on a busy RFC), so it
|
||||
# carries the full one-click unsubscribe — Gmail and Yahoo
|
||||
# require this for senders at OHM's volume tier per RFC 8058.
|
||||
unsubscribe_url = make_unsubscribe_url(user["id"], category)
|
||||
sent = _deliver(
|
||||
cfg,
|
||||
user["email"],
|
||||
subject,
|
||||
body,
|
||||
unsubscribe_mailto=cfg.unsubscribe_mailto,
|
||||
unsubscribe_url=unsubscribe_url,
|
||||
)
|
||||
if not sent:
|
||||
return
|
||||
db.conn().execute(
|
||||
@@ -305,23 +336,45 @@ def _deep_link(payload: dict, cfg: EmailConfig) -> str:
|
||||
return cfg.app_url
|
||||
|
||||
|
||||
def _deliver(cfg: EmailConfig, to_address: str, subject: str, body: str) -> bool:
|
||||
def _deliver(
|
||||
cfg: EmailConfig,
|
||||
to_address: str,
|
||||
subject: str,
|
||||
body: str,
|
||||
*,
|
||||
unsubscribe_mailto: str | None = None,
|
||||
unsubscribe_url: str | None = None,
|
||||
) -> bool:
|
||||
"""Build the envelope via the shared `build_envelope` helper and
|
||||
hand it to SMTP.
|
||||
|
||||
The `_SENT` buffer carries the helper's `EmailMessage` under
|
||||
`message` plus the legacy `to`/`from`/`subject`/`body` keys for
|
||||
backward-compatibility with tests that read those directly.
|
||||
Newer tests can assert on the header surface by inspecting
|
||||
`envelope["message"]`.
|
||||
"""
|
||||
msg = build_envelope(
|
||||
to_address=to_address,
|
||||
from_address=cfg.from_address,
|
||||
from_name=cfg.from_name,
|
||||
subject=subject,
|
||||
body_plain=body,
|
||||
unsubscribe_mailto=unsubscribe_mailto,
|
||||
unsubscribe_url=unsubscribe_url,
|
||||
)
|
||||
envelope = {
|
||||
"to": to_address,
|
||||
"from": formataddr((cfg.from_name, cfg.from_address)),
|
||||
"subject": subject,
|
||||
"body": body,
|
||||
"message": msg,
|
||||
}
|
||||
_SENT.append(envelope)
|
||||
if not cfg.smtp_host:
|
||||
log.info("email (stdout fallback): to=%s subject=%s", to_address, subject)
|
||||
return True
|
||||
try:
|
||||
msg = EmailMessage()
|
||||
msg["From"] = envelope["from"]
|
||||
msg["To"] = to_address
|
||||
msg["Subject"] = subject
|
||||
msg.set_content(body)
|
||||
smtp = smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=30)
|
||||
try:
|
||||
if cfg.smtp_starttls:
|
||||
@@ -440,13 +493,26 @@ def _send_bundle(cfg: EmailConfig, user: Any, emailable: list) -> int:
|
||||
for r, _cat, extras in group_rows:
|
||||
summary = _summary_for(r["event_kind"], r["actor_display"], r["rfc_title"], extras)
|
||||
sections.append(f" · {summary}")
|
||||
# v0.18.0: the bundle covers multiple categories, so a
|
||||
# per-category opt-out can't honor the user's intent. The
|
||||
# `all` category lands at the §15.4 endpoint and sets
|
||||
# `email_opt_out_all = 1`.
|
||||
unsubscribe_url = make_unsubscribe_url(user["id"], "all")
|
||||
body = (
|
||||
"Activity on RFCs you watch, accumulated during your quiet hours:\n"
|
||||
+ "\n".join(sections)
|
||||
+ f"\n\nOpen your inbox: {cfg.app_url}/inbox\n"
|
||||
+ f"Manage all preferences: {cfg.app_url}/settings/notifications\n"
|
||||
+ f"Unsubscribe from all email: {unsubscribe_url}\n"
|
||||
)
|
||||
sent = _deliver(
|
||||
cfg,
|
||||
user["email"],
|
||||
subject,
|
||||
body,
|
||||
unsubscribe_mailto=cfg.unsubscribe_mailto,
|
||||
unsubscribe_url=unsubscribe_url,
|
||||
)
|
||||
sent = _deliver(cfg, user["email"], subject, body)
|
||||
if not sent:
|
||||
return 0
|
||||
ids = [r["id"] for r, _, _ in emailable]
|
||||
|
||||
@@ -31,10 +31,10 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import smtplib
|
||||
from email.message import EmailMessage
|
||||
from email.utils import formataddr
|
||||
|
||||
from .email import EmailConfig, _SENT
|
||||
from .email_envelope import build_envelope
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
@@ -59,12 +59,29 @@ def send_invite_email(
|
||||
cfg = EmailConfig.from_env()
|
||||
subject = _subject(inviter_display, cfg)
|
||||
body = _body(claim_url, inviter_display, inviter_email, custom_message, cfg)
|
||||
# v0.18.0: invite mail carries a `List-Unsubscribe: <mailto:…>`
|
||||
# only (no signed URL) — the invitee isn't a user yet, so there
|
||||
# is no per-user opt-out row to flip. The operator handles
|
||||
# ad-hoc opt-outs from the mailto: target. Per the proposal's
|
||||
# "Tradeoff discussion": the invite was unsolicited from the
|
||||
# recipient's perspective, so the courtesy header is right;
|
||||
# but it can't be a one-click URL because the row doesn't
|
||||
# exist yet.
|
||||
msg = build_envelope(
|
||||
to_address=to_address,
|
||||
from_address=cfg.from_address,
|
||||
from_name=cfg.from_name,
|
||||
subject=subject,
|
||||
body_plain=body,
|
||||
unsubscribe_mailto=cfg.unsubscribe_mailto,
|
||||
)
|
||||
envelope = {
|
||||
"to": to_address,
|
||||
"from": formataddr((cfg.from_name, cfg.from_address)),
|
||||
"subject": subject,
|
||||
"body": body,
|
||||
"kind": "invite",
|
||||
"message": msg,
|
||||
}
|
||||
_SENT.append(envelope)
|
||||
|
||||
@@ -79,11 +96,6 @@ def send_invite_email(
|
||||
return True
|
||||
|
||||
try:
|
||||
msg = EmailMessage()
|
||||
msg["From"] = envelope["from"]
|
||||
msg["To"] = to_address
|
||||
msg["Subject"] = subject
|
||||
msg.set_content(body)
|
||||
smtp = smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=30)
|
||||
try:
|
||||
if cfg.smtp_starttls:
|
||||
|
||||
@@ -23,10 +23,10 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import smtplib
|
||||
from email.message import EmailMessage
|
||||
from email.utils import formataddr
|
||||
|
||||
from .email import EmailConfig, _SENT
|
||||
from .email_envelope import build_envelope
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
@@ -44,12 +44,25 @@ def send_otc_email(to_address: str, code: str) -> bool:
|
||||
cfg = EmailConfig.from_env()
|
||||
subject = f"Your sign-in code for {cfg.from_name}"
|
||||
body = _body(code, cfg)
|
||||
# v0.18.0: OTC mail carries NO List-Unsubscribe — the recipient
|
||||
# explicitly requested the code; advertising an unsubscribe
|
||||
# header would imply OHM has them on a list, which it doesn't.
|
||||
# See the proposal's "Tradeoff discussion" for the binding
|
||||
# rationale.
|
||||
msg = build_envelope(
|
||||
to_address=to_address,
|
||||
from_address=cfg.from_address,
|
||||
from_name=cfg.from_name,
|
||||
subject=subject,
|
||||
body_plain=body,
|
||||
)
|
||||
envelope = {
|
||||
"to": to_address,
|
||||
"from": formataddr((cfg.from_name, cfg.from_address)),
|
||||
"subject": subject,
|
||||
"body": body,
|
||||
"kind": "otc",
|
||||
"message": msg,
|
||||
}
|
||||
_SENT.append(envelope)
|
||||
|
||||
@@ -64,11 +77,6 @@ def send_otc_email(to_address: str, code: str) -> bool:
|
||||
return True
|
||||
|
||||
try:
|
||||
msg = EmailMessage()
|
||||
msg["From"] = envelope["from"]
|
||||
msg["To"] = to_address
|
||||
msg["Subject"] = subject
|
||||
msg.set_content(body)
|
||||
smtp = smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=30)
|
||||
try:
|
||||
if cfg.smtp_starttls:
|
||||
|
||||
@@ -647,3 +647,82 @@ def test_pending_invites_listing_admin_only(app_with_fake_gitea):
|
||||
)
|
||||
r = client.get("/api/admin/users/invites")
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# v0.18.0: invite-envelope header shape — Slice 2
|
||||
#
|
||||
# Invite mail goes through `build_envelope` and MUST land Date,
|
||||
# Message-ID, Auto-Submitted, AND a `List-Unsubscribe: <mailto:…>`
|
||||
# (no URL — the invitee isn't a user yet, so no per-user opt-out
|
||||
# row exists). The mailto: target is the operator's `EMAIL_FROM`
|
||||
# by default; the operator can override via `EMAIL_UNSUBSCRIBE_MAILTO`.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _provision_admin_and_send_invite(client, app_with_fake_gitea_fixture, *, to: str = "headers@ex.co"):
|
||||
provision_user_row(user_id=400, login="adminH", role="admin")
|
||||
sign_in_as(
|
||||
client, user_id=400, gitea_login="adminH",
|
||||
display_name="Admin H", role="admin",
|
||||
email="adminh@test",
|
||||
)
|
||||
_reset_outbound()
|
||||
r = client.post(
|
||||
"/api/admin/users",
|
||||
json={
|
||||
"email": to,
|
||||
"first_name": "Header",
|
||||
"last_name": "Test",
|
||||
"role": "contributor",
|
||||
"custom_message": "",
|
||||
},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
return _outbound_invite_envelopes(to)[-1]
|
||||
|
||||
|
||||
def test_invite_envelope_sets_date_messageid_autosubmitted(app_with_fake_gitea):
|
||||
from fastapi.testclient import TestClient
|
||||
from email.utils import parsedate_to_datetime
|
||||
|
||||
app, _fake = app_with_fake_gitea
|
||||
with TestClient(app) as client:
|
||||
env = _provision_admin_and_send_invite(client, (app, _fake))
|
||||
msg = env["message"]
|
||||
assert parsedate_to_datetime(msg["Date"]) is not None
|
||||
assert msg["Message-ID"].startswith("<") and msg["Message-ID"].endswith(">")
|
||||
assert msg["Auto-Submitted"] == "auto-generated"
|
||||
|
||||
|
||||
def test_invite_envelope_has_mailto_list_unsubscribe_only(app_with_fake_gitea):
|
||||
"""The invitee isn't a user yet — no per-user opt-out URL is
|
||||
available. The `List-Unsubscribe` MUST be a mailto: form, and
|
||||
the `List-Unsubscribe-Post` header MUST be absent (the
|
||||
one-click semantic requires a URL the MUA can POST to)."""
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
app, _fake = app_with_fake_gitea
|
||||
with TestClient(app) as client:
|
||||
env = _provision_admin_and_send_invite(client, (app, _fake))
|
||||
msg = env["message"]
|
||||
lu = msg["List-Unsubscribe"]
|
||||
assert lu is not None and lu.startswith("<mailto:")
|
||||
# No URL part — invite is mailto-only.
|
||||
assert "https://" not in lu and "http://" not in lu
|
||||
assert msg["List-Unsubscribe-Post"] is None
|
||||
|
||||
|
||||
def test_invite_envelope_respects_email_unsubscribe_mailto_override(app_with_fake_gitea, monkeypatch):
|
||||
"""When `EMAIL_UNSUBSCRIBE_MAILTO` is set, the mailto: target on
|
||||
`List-Unsubscribe` honors it (lets a deployment route opt-outs
|
||||
to a humans-monitored mailbox distinct from the no-reply
|
||||
sender)."""
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
monkeypatch.setenv("EMAIL_UNSUBSCRIBE_MAILTO", "ohm@wiggleverse.org?subject=remove")
|
||||
app, _fake = app_with_fake_gitea
|
||||
with TestClient(app) as client:
|
||||
env = _provision_admin_and_send_invite(client, (app, _fake))
|
||||
msg = env["message"]
|
||||
assert "ohm@wiggleverse.org?subject=remove" in msg["List-Unsubscribe"]
|
||||
|
||||
@@ -612,3 +612,127 @@ def test_explicit_watch_set_overrides_auto(app_with_fake_gitea):
|
||||
# the user put them.
|
||||
assert row["set_by"] == "explicit"
|
||||
assert row["state"] == "following"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# v0.18.0 — envelope headers + RFC 8058 one-click POST endpoint
|
||||
#
|
||||
# Watcher notifications are bulk-adjacent (a busy RFC can produce
|
||||
# dozens of structural events); per the proposal, they MUST carry
|
||||
# `Date`, `Message-ID`, `Auto-Submitted`, full `List-Unsubscribe`
|
||||
# (mailto + signed URL), AND `List-Unsubscribe-Post:
|
||||
# List-Unsubscribe=One-Click` per RFC 8058. Gmail and Yahoo
|
||||
# enforce this for senders at OHM's volume tier.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_notification_envelope_carries_full_one_click_headers(app_with_fake_gitea):
|
||||
"""A `proposal_merged` event lands a watcher notification email
|
||||
with the full one-click unsubscribe shape."""
|
||||
from fastapi.testclient import TestClient
|
||||
from email.utils import parsedate_to_datetime
|
||||
from app import db, email as email_mod
|
||||
|
||||
app, fake = app_with_fake_gitea
|
||||
with TestClient(app) as client:
|
||||
provision_user_row(user_id=2, login="alice", role="contributor")
|
||||
provision_user_row(user_id=1, login="ben", role="owner")
|
||||
|
||||
sign_in_as(client, user_id=2, gitea_login="alice", display_name="Alice", role="contributor", email="alice@test")
|
||||
r = client.post("/api/rfcs/propose", json={"title": "OHM", "slug": "ohm", "pitch": PITCH, "tags": []})
|
||||
assert r.status_code == 200
|
||||
email_mod.reset_sent_envelopes()
|
||||
sign_in_as(client, user_id=1, gitea_login="ben", display_name="Ben", role="owner", email="ben@test")
|
||||
merge_r = client.post(f"/api/proposals/{r.json()['pr_number']}/merge")
|
||||
assert merge_r.status_code == 200, merge_r.text
|
||||
|
||||
envelopes = [e for e in email_mod.sent_envelopes() if e["to"] == "alice@test"]
|
||||
assert envelopes, "watcher notification did not fire"
|
||||
msg = envelopes[-1]["message"]
|
||||
# Always-present headers from the helper.
|
||||
assert parsedate_to_datetime(msg["Date"]) is not None
|
||||
assert msg["Message-ID"].startswith("<") and msg["Message-ID"].endswith(">")
|
||||
assert msg["Auto-Submitted"] == "auto-generated"
|
||||
# Full one-click unsubscribe.
|
||||
lu = msg["List-Unsubscribe"]
|
||||
assert lu is not None
|
||||
assert "<mailto:" in lu
|
||||
# URL part carries the signed token per make_unsubscribe_url.
|
||||
assert "/api/email/unsubscribe?t=" in lu
|
||||
assert msg["List-Unsubscribe-Post"] == "List-Unsubscribe=One-Click"
|
||||
|
||||
|
||||
def test_email_unsubscribe_post_one_click_flips_category_off(app_with_fake_gitea):
|
||||
"""RFC 8058: Gmail/Yahoo POST `List-Unsubscribe=One-Click` to the
|
||||
URL in the List-Unsubscribe header. The endpoint MUST accept POST
|
||||
+ the same token shape as the GET handler + return 200 + flip the
|
||||
flag."""
|
||||
from fastapi.testclient import TestClient
|
||||
from app import db, email as email_mod
|
||||
|
||||
app, _fake = app_with_fake_gitea
|
||||
with TestClient(app) as client:
|
||||
provision_user_row(user_id=2, login="alice", role="contributor")
|
||||
token = email_mod.make_unsubscribe_url(2, "personal-direct").split("t=", 1)[1]
|
||||
|
||||
r = client.post(
|
||||
f"/api/email/unsubscribe?t={token}",
|
||||
data={"List-Unsubscribe": "One-Click"},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
body = r.json()
|
||||
assert body["ok"] is True
|
||||
assert body["category"] == "personal-direct"
|
||||
|
||||
row = db.conn().execute(
|
||||
"SELECT email_personal_direct FROM users WHERE id = 2"
|
||||
).fetchone()
|
||||
assert row["email_personal_direct"] == 0
|
||||
|
||||
|
||||
def test_email_unsubscribe_post_all_sets_global_opt_out(app_with_fake_gitea):
|
||||
"""The v0.18.0 `all` synthetic category (used by the bundle +
|
||||
digest paths) MUST set `email_opt_out_all = 1`."""
|
||||
from fastapi.testclient import TestClient
|
||||
from app import db, email as email_mod
|
||||
|
||||
app, _fake = app_with_fake_gitea
|
||||
with TestClient(app) as client:
|
||||
provision_user_row(user_id=2, login="alice", role="contributor")
|
||||
token = email_mod.make_unsubscribe_url(2, "all").split("t=", 1)[1]
|
||||
r = client.post(f"/api/email/unsubscribe?t={token}")
|
||||
assert r.status_code == 200
|
||||
assert r.json() == {"ok": True, "category": "all"}
|
||||
row = db.conn().execute(
|
||||
"SELECT email_opt_out_all FROM users WHERE id = 2"
|
||||
).fetchone()
|
||||
assert row["email_opt_out_all"] == 1
|
||||
|
||||
|
||||
def test_email_unsubscribe_get_all_sets_global_opt_out(app_with_fake_gitea):
|
||||
"""GET handler also accepts the `all` category and lands the
|
||||
global opt-out (so an MUA that doesn't honor RFC 8058 POST and
|
||||
just opens the URL in a browser still works)."""
|
||||
from fastapi.testclient import TestClient
|
||||
from app import db, email as email_mod
|
||||
|
||||
app, _fake = app_with_fake_gitea
|
||||
with TestClient(app) as client:
|
||||
provision_user_row(user_id=2, login="alice", role="contributor")
|
||||
token = email_mod.make_unsubscribe_url(2, "all").split("t=", 1)[1]
|
||||
r = client.get(f"/api/email/unsubscribe?t={token}")
|
||||
assert r.status_code == 200
|
||||
assert "Unsubscribed" in r.text
|
||||
row = db.conn().execute(
|
||||
"SELECT email_opt_out_all FROM users WHERE id = 2"
|
||||
).fetchone()
|
||||
assert row["email_opt_out_all"] == 1
|
||||
|
||||
|
||||
def test_email_unsubscribe_post_refuses_invalid_token(app_with_fake_gitea):
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
app, _fake = app_with_fake_gitea
|
||||
with TestClient(app) as client:
|
||||
r = client.post("/api/email/unsubscribe?t=not-a-valid-token")
|
||||
assert r.status_code == 400
|
||||
|
||||
@@ -347,3 +347,53 @@ def test_otc_re_request_invalidates_prior_unused_code(app_with_fake_gitea, monke
|
||||
# The new code still works.
|
||||
r = client.post("/auth/otc/verify", json={"email": "alice@example.com", "code": second})
|
||||
assert r.status_code == 200
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# v0.18.0: envelope headers — Slice 2
|
||||
#
|
||||
# OTC mail goes through `build_envelope` and MUST land Date,
|
||||
# Message-ID, and Auto-Submitted but MUST NOT carry a
|
||||
# List-Unsubscribe header (the recipient explicitly requested the
|
||||
# code; advertising a list semantic would be wrong).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _last_otc_envelope():
|
||||
from app import email as email_mod
|
||||
otc = [e for e in email_mod.sent_envelopes() if e.get("kind") == "otc"]
|
||||
assert otc, "no OTC envelope in the buffer"
|
||||
return otc[-1]
|
||||
|
||||
|
||||
def test_otc_envelope_sets_date_messageid_autosubmitted(app_with_fake_gitea):
|
||||
from fastapi.testclient import TestClient
|
||||
from email.utils import parsedate_to_datetime
|
||||
|
||||
app, _fake = app_with_fake_gitea
|
||||
with TestClient(app) as client:
|
||||
_reset_outbound()
|
||||
client.post("/auth/otc/request", json={"email": "headers@example.com"})
|
||||
msg = _last_otc_envelope()["message"]
|
||||
# Date is RFC 5322 parseable.
|
||||
assert parsedate_to_datetime(msg["Date"]) is not None
|
||||
# Message-ID is bracketed and carries the From-address @-domain.
|
||||
mid = msg["Message-ID"]
|
||||
assert mid.startswith("<") and mid.endswith(">")
|
||||
# Auto-Submitted prevents auto-responder loops.
|
||||
assert msg["Auto-Submitted"] == "auto-generated"
|
||||
|
||||
|
||||
def test_otc_envelope_has_no_list_unsubscribe(app_with_fake_gitea):
|
||||
"""The recipient explicitly typed their email and asked for a
|
||||
code; the framework MUST NOT advertise a list semantic on this
|
||||
mail. Per the v0.18.0 proposal's tradeoff discussion."""
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
app, _fake = app_with_fake_gitea
|
||||
with TestClient(app) as client:
|
||||
_reset_outbound()
|
||||
client.post("/auth/otc/request", json={"email": "headers@example.com"})
|
||||
msg = _last_otc_envelope()["message"]
|
||||
assert msg["List-Unsubscribe"] is None
|
||||
assert msg["List-Unsubscribe-Post"] is None
|
||||
|
||||
Reference in New Issue
Block a user