"""Subscriptions: bulk senders worked out from List-* headers, and unsubscribing.""" from __future__ import annotations import datetime as dt from typing import Any import pytest from sqlalchemy import select, update from app.api import subscriptions as api_subs from app.models import Draft, Mailbox, Message from app.subscriptions import backfill_list_headers_job, read_list_headers, unsubscribe_targets from tests.api.conftest import HDR, MailCtx, raw_mail NEWS = { "List-Id": '"Weekly news" ', "List-Unsubscribe": ", ", "List-Unsubscribe-Post": "List-Unsubscribe=One-Click", } def test_read_list_headers() -> None: h = read_list_headers(raw_mail("x", headers=NEWS)) assert h.list_id == "weekly.shop.example" assert h.unsubscribe.startswith("") assert h.one_click is True plain = read_list_headers(raw_mail("hi")) assert (plain.list_id, plain.unsubscribe, plain.one_click) == (None, "", False) def test_unsubscribe_targets() -> None: t = unsubscribe_targets(NEWS["List-Unsubscribe"]) assert t.https == "https://shop.example/u/1" assert t.mailto == "leave@shop.example" and t.mailto_subject == "stop" assert unsubscribe_targets("").https is None assert unsubscribe_targets("").mailto is None async def _list(ctx: MailCtx) -> list[dict[str, Any]]: resp = await ctx.client.get("/api/subscriptions") assert resp.status_code == 200, resp.text return list(resp.json()["subscriptions"]) async def test_groups_by_list_then_sender(mail_ctx: MailCtx) -> None: await mail_ctx.deliver(raw_mail("Issue 1", headers=NEWS), 1, seen=True) await mail_ctx.deliver(raw_mail("Issue 2", headers=NEWS), 2) await mail_ctx.deliver( raw_mail( "Sale", sender="Deals ", headers={"List-Unsubscribe": ""}, ), 3, ) await mail_ctx.deliver(raw_mail("Hi from a friend", sender="ann@friend.example"), 4) subs = {s["key"]: s for s in await _list(mail_ctx)} assert set(subs) == {"list:weekly.shop.example", "from:deals@other.example"} news = subs["list:weekly.shop.example"] assert (news["total"], news["unread"], news["name"]) == (2, 1, "News") assert news["one_click"] and news["has_mailto"] and news["has_web"] assert news["last_subject"] == "Issue 2" threads = await mail_ctx.client.get( "/api/subscriptions/threads", params={"key": "list:weekly.shop.example"} ) assert len(threads.json()) == 2 async def test_one_click_unsubscribe(mail_ctx: MailCtx, monkeypatch: pytest.MonkeyPatch) -> None: await mail_ctx.deliver(raw_mail("Issue", headers=NEWS), 1) posted: list[str] = [] async def fake_one_click(url: str) -> bool: posted.append(url) return True monkeypatch.setattr(api_subs, "_one_click", fake_one_click) resp = await mail_ctx.client.post( "/api/subscriptions/unsubscribe", json={"key": "list:weekly.shop.example"}, headers=HDR ) assert resp.json() == {"status": "done", "method": "one_click", "url": None} assert posted == ["https://shop.example/u/1"] (news,) = await _list(mail_ctx) assert news["unsubscribed_at"] is not None and news["still_sending"] is False async def test_mailto_unsubscribe_and_still_sending( mail_ctx: MailCtx, monkeypatch: pytest.MonkeyPatch ) -> None: enqueued: list[tuple[Any, ...]] = [] async def fake_enqueue(name: str, *args: Any, **kwargs: Any) -> None: enqueued.append((name, *args)) monkeypatch.setattr(api_subs, "enqueue_job", fake_enqueue) headers = {"List-Unsubscribe": ""} await mail_ctx.deliver(raw_mail("Issue", headers=headers), 1) resp = await mail_ctx.client.post( "/api/subscriptions/unsubscribe", json={"key": "from:news@shop.example"}, headers=HDR ) assert resp.json()["method"] == "mailto" assert [e[0] for e in enqueued] == ["send_job"] draft = await mail_ctx.session.scalar(select(Draft).where(Draft.user_id == mail_ctx.user_id)) assert draft is not None assert draft.payload["to"] == ["leave@shop.example"] and draft.payload["subject"] == "stop" # Mail still arriving after the grace period is flagged. later = dt.datetime.now(dt.UTC) + dt.timedelta(days=3) await mail_ctx.deliver(raw_mail("Issue again", headers=headers), 2, at=later) (sub,) = await _list(mail_ctx) assert sub["still_sending"] is True async def test_web_only_unsubscribe_hands_back_the_page(mail_ctx: MailCtx) -> None: headers = {"List-Unsubscribe": ""} await mail_ctx.deliver(raw_mail("Issue", headers=headers), 1) resp = await mail_ctx.client.post( "/api/subscriptions/unsubscribe", json={"key": "from:news@shop.example"}, headers=HDR ) assert resp.json() == {"status": "open", "method": "web", "url": "https://shop.example/prefs"} async def test_backfill_reads_older_mail(mail_ctx: MailCtx) -> None: msg = await mail_ctx.deliver(raw_mail("Old issue", headers=NEWS), 1) await mail_ctx.session.execute( update(Message).where(Message.id == msg.id).values(list_unsubscribe=None, list_id=None) ) await mail_ctx.session.commit() resp = await mail_ctx.client.get("/api/subscriptions") assert resp.json()["pending"] == 1 assert await backfill_list_headers_job({}, str(mail_ctx.user_id)) == 1 resp = await mail_ctx.client.get("/api/subscriptions") assert resp.json()["pending"] == 0 assert [s["key"] for s in resp.json()["subscriptions"]] == ["list:weekly.shop.example"] async def test_trash_all_leaves_mixed_conversations( mail_ctx: MailCtx, monkeypatch: pytest.MonkeyPatch ) -> None: moved: list[str] = [] async def fake_enqueue(name: str, *args: Any, **kwargs: Any) -> None: moved.append(args[0]) monkeypatch.setattr(api_subs, "enqueue_job", fake_enqueue) only = await mail_ctx.deliver(raw_mail("Issue", headers=NEWS, mid=""), 1) await mail_ctx.deliver(raw_mail("Issue 2", headers=NEWS, mid=""), 2) reply = raw_mail( "Re: Issue 2", sender="me@example.com", headers={"In-Reply-To": "", "References": ""}, ) await mail_ctx.deliver(reply, 3) resp = await mail_ctx.client.post( "/api/subscriptions/trash", json={"key": "list:weekly.shop.example"}, headers=HDR ) assert resp.json() == {"moved": 1} assert moved == [str(only.thread_id)] async def test_deleted_mail_is_left_out(mail_ctx: MailCtx) -> None: msg = await mail_ctx.deliver(raw_mail("Issue", headers=NEWS), 1) trash = await mail_ctx.session.scalar( select(Mailbox.id).where( Mailbox.account_id == mail_ctx.account_id, Mailbox.special_use == "trash" ) ) await mail_ctx.session.execute( update(Message).where(Message.id == msg.id).values(mailbox_id=trash) ) await mail_ctx.session.commit() assert await _list(mail_ctx) == [] async def test_reading_pane_offers_unsubscribe( mail_ctx: MailCtx, monkeypatch: pytest.MonkeyPatch ) -> None: async def fake_one_click(url: str) -> bool: return True monkeypatch.setattr(api_subs, "_one_click", fake_one_click) news = await mail_ctx.deliver(raw_mail("Issue", headers=NEWS), 1) plain = await mail_ctx.deliver(raw_mail("Hi", sender="ann@friend.example"), 2) c = mail_ctx.client (env,) = (await c.get(f"/api/threads/{news.thread_id}")).json()["messages"] assert env["subscription_key"] == "list:weekly.shop.example" assert env["unsubscribe_via"] == "one_click" and env["unsubscribed"] is False (other,) = (await c.get(f"/api/threads/{plain.thread_id}")).json()["messages"] assert other["subscription_key"] is None and other["unsubscribe_via"] is None resp = await c.post( "/api/subscriptions/unsubscribe", json={"key": env["subscription_key"], "message_id": env["id"]}, headers=HDR, ) assert resp.json()["status"] == "done" (env,) = (await c.get(f"/api/threads/{news.thread_id}")).json()["messages"] assert env["unsubscribed"] is True # A message without the header can't be used. resp = await c.post( "/api/subscriptions/unsubscribe", json={"key": "from:ann@friend.example", "message_id": str(plain.id)}, headers=HDR, ) assert resp.status_code == 404 async def test_senders_come_in_pages_newest_first(mail_ctx: MailCtx) -> None: now = dt.datetime.now(dt.UTC) for i, name in enumerate(["old", "mid", "new"]): await mail_ctx.deliver( raw_mail( f"From {name}", sender=f"{name}@lists.example", headers={"List-Unsubscribe": f""}, ), i + 1, at=now - dt.timedelta(days=10 - i), ) c = mail_ctx.client first = (await c.get("/api/subscriptions", params={"limit": 2})).json() assert first["total"] == 3 assert [s["email"] for s in first["subscriptions"]] == [ "new@lists.example", "mid@lists.example", ] rest = (await c.get("/api/subscriptions", params={"limit": 2, "offset": 2})).json() assert [s["email"] for s in rest["subscriptions"]] == ["old@lists.example"] assert (await c.get("/api/subscriptions", params={"offset": 9})).json()["subscriptions"] == [] oldest = (await c.get("/api/subscriptions", params={"limit": 2, "order": "asc"})).json() assert [s["email"] for s in oldest["subscriptions"]] == [ "old@lists.example", "mid@lists.example", ]