webmail-client/backend/tests/api/test_subscriptions.py

"""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" <weekly.shop.example>',
    "List-Unsubscribe": "<https://shop.example/u/1>, <mailto:leave@shop.example?subject=stop>",
    "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("<https://shop.example/u/1>")
    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("<http://insecure.example/u>").https is None
    assert unsubscribe_targets("<mailto:a@b.c%0d%0aBcc:x@y.z>").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 <deals@other.example>",
            headers={"List-Unsubscribe": "<mailto:u@other.example>"},
        ),
        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": "<mailto:leave@shop.example?subject=stop>"}
    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": "<https://shop.example/prefs>"}
    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="<i1@shop>"), 1)
    await mail_ctx.deliver(raw_mail("Issue 2", headers=NEWS, mid="<i2@shop>"), 2)
    reply = raw_mail(
        "Re: Issue 2",
        sender="me@example.com",
        headers={"In-Reply-To": "<i2@shop>", "References": "<i2@shop>"},
    )
    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"<mailto:u@{name}.example>"},
            ),
            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",
    ]