webmail-client/backend/tests/mail/test_actions.py

"""Flag / move actions (SPEC §6.3, §9.1): DB-first + enqueue + revert."""

from __future__ import annotations

import uuid
from collections.abc import AsyncIterator
from dataclasses import dataclass, field
from datetime import UTC, datetime
from email.message import EmailMessage

import httpx
import pytest
import pytest_asyncio
from arq import Retry
from httpx import ASGITransport
from sqlalchemy import delete, func, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool

from app import queue as app_queue
from app.config import get_settings
from app.db import session_scope
from app.identity.passwords import hash_password
from app.identity.sessions import create_session
from app.indexer.indexer import persist_message
from app.main import create_app
from app.models import Account, Mailbox, Message, Thread, User
from app.providers.base import ProviderConnectError, RemoteMessage
from app.sync import actions as sync_actions

CSRF = "csrf"


@dataclass
class Ctx:
    client: httpx.AsyncClient
    session: AsyncSession
    user_id: uuid.UUID
    inbox_id: uuid.UUID
    archive_id: uuid.UUID
    message_ids: list[uuid.UUID] = field(default_factory=list)
    thread_id: uuid.UUID | None = None
    enqueued: list[tuple] = field(default_factory=list)


def _hdr() -> dict[str, str]:
    return {"X-CSRF": CSRF}


def _raw(subject: str, mid: str, refs: str | None = None) -> bytes:
    m = EmailMessage()
    m["Subject"] = subject
    m["From"] = "a@x.com"
    m["To"] = "me@example.com"
    m["Message-ID"] = mid
    m["Date"] = "Mon, 09 Mar 2026 08:36:37 +0100"
    if refs:
        m["References"] = refs
    m.set_content("body")
    return m.as_bytes()


@pytest_asyncio.fixture
async def ctx(monkeypatch) -> AsyncIterator[Ctx]:  # type: ignore[no-untyped-def]
    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
    sm = async_sessionmaker(engine, expire_on_commit=False)
    session = sm()
    user = User(handle=f"act{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw"))
    session.add(user)
    await session.flush()
    account = Account(
        user_id=user.id,
        email="me@example.com",
        imap_config={},
        smtp_config={},
        secret_ct=b"",
        discovery_src="manual",
    )
    session.add(account)
    await session.flush()
    inbox = Mailbox(account_id=account.id, user_id=user.id, path="INBOX", special_use="inbox")
    archive = Mailbox(account_id=account.id, user_id=user.id, path="Archive", special_use="archive")
    session.add_all([inbox, archive])
    await session.commit()

    c = Ctx(
        client=None,
        session=session,
        user_id=user.id,
        inbox_id=inbox.id,  # type: ignore[arg-type]
        archive_id=archive.id,
    )
    for i, (subj, mid, refs) in enumerate(
        [
            ("Topic", "<a@x.com>", None),
            ("Re: Topic", "<b@x.com>", "<a@x.com>"),
        ]
    ):
        remote = RemoteMessage(
            uid=i + 1, raw=_raw(subj, mid, refs), flags=frozenset(), internal_date=datetime.now(UTC)
        )
        result = await persist_message(
            session,
            user_id=user.id,
            account_id=account.id,
            mailbox_id=inbox.id,
            mailbox_special_use="inbox",
            remote=remote,
        )
        await session.commit()
        assert result is not None
        c.message_ids.append(result.message.id)
        c.thread_id = result.message.thread_id

    # capture enqueues instead of hitting a real arq pool
    class FakePool:
        async def enqueue_job(self, *args: object, **kwargs: object) -> None:
            c.enqueued.append(args)

    async def fake_pool() -> FakePool:
        return FakePool()

    monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool)

    token = await create_session(user.id, "pytest")
    c.client = httpx.AsyncClient(
        transport=ASGITransport(app=create_app()),
        base_url="http://testserver",
        cookies={"session": token, "csrf": CSRF},
    )
    try:
        yield c
    finally:
        await c.client.aclose()
        await session.execute(delete(User).where(User.id == user.id))
        await session.commit()
        await session.close()
        await engine.dispose()


async def test_mark_message_read_enqueues_push(ctx: Ctx) -> None:
    mid = ctx.message_ids[0]
    resp = await ctx.client.post(
        f"/api/messages/{mid}/flags", json={"add": ["\\Seen"]}, headers=_hdr()
    )
    assert resp.status_code == 200
    assert "\\Seen" in resp.json()["flags"]

    await ctx.session.refresh(await ctx.session.get(Message, mid))
    msg = await ctx.session.get(Message, mid)
    assert msg is not None and "\\Seen" in msg.flags
    assert ctx.enqueued == [("push_flags_job", str(mid), ["\\Seen"], [])]


async def test_noop_flag_change_does_not_enqueue(ctx: Ctx) -> None:
    mid = ctx.message_ids[0]
    await ctx.client.post(f"/api/messages/{mid}/flags", json={"add": ["\\Seen"]}, headers=_hdr())
    ctx.enqueued.clear()
    # already \Seen -> no-op
    resp = await ctx.client.post(
        f"/api/messages/{mid}/flags", json={"add": ["\\Seen"]}, headers=_hdr()
    )
    assert resp.status_code == 200
    assert ctx.enqueued == []


async def test_thread_flags_mark_all_read(ctx: Ctx) -> None:
    resp = await ctx.client.post(
        f"/api/threads/{ctx.thread_id}/flags", json={"add": ["\\Seen"]}, headers=_hdr()
    )
    assert resp.status_code == 200
    thread = await ctx.session.get(Thread, ctx.thread_id)
    await ctx.session.refresh(thread)
    assert thread is not None and thread.unread_count == 0
    assert len(ctx.enqueued) == 2  # both messages pushed


async def test_thread_flag_sets_pinned(ctx: Ctx) -> None:
    resp = await ctx.client.post(
        f"/api/threads/{ctx.thread_id}/flags", json={"add": ["\\Flagged"]}, headers=_hdr()
    )
    assert resp.status_code == 200
    thread = await ctx.session.get(Thread, ctx.thread_id)
    await ctx.session.refresh(thread)
    assert thread is not None and thread.pinned is True

    # Unpinning clears it again.
    await ctx.client.post(
        f"/api/threads/{ctx.thread_id}/flags", json={"remove": ["\\Flagged"]}, headers=_hdr()
    )
    await ctx.session.refresh(thread)
    assert thread.pinned is False


async def test_move_message(ctx: Ctx) -> None:
    mid = ctx.message_ids[0]
    resp = await ctx.client.post(
        f"/api/messages/{mid}/move", json={"mailbox_id": str(ctx.archive_id)}, headers=_hdr()
    )
    assert resp.status_code == 200
    msg = await ctx.session.get(Message, mid)
    await ctx.session.refresh(msg)
    assert msg is not None and msg.mailbox_id == ctx.archive_id
    assert ctx.enqueued[0][0] == "move_job"


async def test_archive_thread_enqueues_move(ctx: Ctx) -> None:
    resp = await ctx.client.post(f"/api/threads/{ctx.thread_id}/archive", headers=_hdr())
    assert resp.status_code == 200
    assert resp.json() == {"status": "queued"}
    assert ctx.enqueued == [
        ("move_thread_to_folder_job", str(ctx.thread_id), "archive"),
    ]


async def test_trash_thread_enqueues_move(ctx: Ctx) -> None:
    resp = await ctx.client.post(f"/api/threads/{ctx.thread_id}/trash", headers=_hdr())
    assert resp.status_code == 200
    assert ctx.enqueued == [
        ("move_thread_to_folder_job", str(ctx.thread_id), "trash"),
    ]


async def test_archive_unknown_thread_404(ctx: Ctx) -> None:
    resp = await ctx.client.post(f"/api/threads/{uuid.uuid4()}/archive", headers=_hdr())
    assert resp.status_code == 404
    assert ctx.enqueued == []


async def test_restore_thread_enqueues_restore_job(ctx: Ctx) -> None:
    resp = await ctx.client.post(f"/api/threads/{ctx.thread_id}/restore", headers=_hdr())
    assert resp.status_code == 200
    assert ctx.enqueued == [
        ("restore_thread_job", str(ctx.thread_id)),
    ]


async def test_restore_unknown_thread_404(ctx: Ctx) -> None:
    resp = await ctx.client.post(f"/api/threads/{uuid.uuid4()}/restore", headers=_hdr())
    assert resp.status_code == 404
    assert ctx.enqueued == []


async def test_delete_thread_enqueues_permanent_delete(ctx: Ctx) -> None:
    resp = await ctx.client.post(f"/api/threads/{ctx.thread_id}/delete", headers=_hdr())
    assert resp.status_code == 200
    assert ctx.enqueued == [("delete_thread_job", str(ctx.thread_id))]


async def test_empty_folder_enqueues_job(ctx: Ctx) -> None:
    resp = await ctx.client.post("/api/folders/trash/empty", headers=_hdr())
    assert resp.status_code == 200
    assert ctx.enqueued == [("empty_folder_job", str(ctx.user_id), "trash")]


async def test_empty_folder_rejects_non_removable_folder(ctx: Ctx) -> None:
    resp = await ctx.client.post("/api/folders/inbox/empty", headers=_hdr())
    assert resp.status_code == 400
    assert ctx.enqueued == []


async def test_ignores_unknown_flags(ctx: Ctx) -> None:
    mid = ctx.message_ids[0]
    resp = await ctx.client.post(
        f"/api/messages/{mid}/flags", json={"add": ["\\Deleted", "bogus"]}, headers=_hdr()
    )
    assert resp.status_code == 200
    msg = await ctx.session.get(Message, mid)
    await ctx.session.refresh(msg)
    assert msg is not None and "\\Deleted" not in msg.flags  # not user-settable here


async def test_delete_thread_job_expunges_and_removes_rows(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no-untyped-def]
    tid = ctx.thread_id
    assert tid is not None
    deleted: list[tuple[str, list[int]]] = []

    class FakeProvider:
        async def delete_messages(self, mailbox: str, uids: list[int]) -> None:
            deleted.append((mailbox, uids))

        async def close(self) -> None:
            return None

    monkeypatch.setattr(sync_actions, "build_provider", lambda _account: FakeProvider())

    await sync_actions.delete_thread_job({}, str(tid))

    # server expunge happened for the thread's INBOX uids
    assert deleted and deleted[0][0] == "INBOX"
    # DB rows for the thread's messages are gone (verify in a fresh session; the
    # job deleted them in its own session, so drop ctx's stale identity map).
    ctx.session.expunge_all()
    async with session_scope() as verify:
        remaining = await verify.scalar(
            select(func.count()).select_from(Message).where(Message.thread_id == tid)
        )
        assert remaining == 0
        assert await verify.scalar(select(Thread).where(Thread.id == tid)) is None


async def test_push_flags_job_reverts_on_failure(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no-untyped-def]
    mid = ctx.message_ids[0]
    # mark read in DB first (as the endpoint would)
    msg = await ctx.session.get(Message, mid)
    assert msg is not None
    msg.flags = ["\\Seen"]
    await ctx.session.commit()

    class FailingProvider:
        async def push_flags(self, *a: object, **k: object) -> None:
            raise RuntimeError("imap down")

        async def close(self) -> None:
            return None

    monkeypatch.setattr(sync_actions, "build_provider", lambda _account: FailingProvider())

    # final try -> should revert (remove the \Seen we added)
    await sync_actions.push_flags_job({"job_try": 5}, str(mid), ["\\Seen"], [])
    reverted = await ctx.session.get(Message, mid)
    await ctx.session.refresh(reverted)
    assert reverted is not None and "\\Seen" not in reverted.flags


# --- the mail server unreachable: real retries (arq Retry), ~24 hours -------------


class _Unreachable:
    """A provider whose server cannot be reached."""

    def __init__(self) -> None:
        self.calls = 0

    async def push_flags(self, *a: object, **k: object) -> None:
        self.calls += 1
        raise ProviderConnectError("cannot connect to imap.example.com")

    async def move(self, *a: object, **k: object) -> int:
        self.calls += 1
        raise ProviderConnectError("cannot connect to imap.example.com")

    async def delete_messages(self, *a: object, **k: object) -> None:
        self.calls += 1
        raise ProviderConnectError("cannot connect to imap.example.com")

    async def close(self) -> None:
        return None


async def _mark_read(ctx: Ctx) -> uuid.UUID:
    mid = ctx.message_ids[0]
    msg = await ctx.session.get(Message, mid)
    assert msg is not None
    msg.flags = ["\\Seen"]
    await ctx.session.commit()
    return mid


async def test_flag_push_retries_while_unreachable(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no-untyped-def]
    mid = await _mark_read(ctx)
    monkeypatch.setattr(sync_actions, "build_provider", lambda _a: _Unreachable())

    with pytest.raises(Retry) as caught:
        await sync_actions.push_flags_job({"job_try": 1}, str(mid), ["\\Seen"], [])
    assert caught.value.defer_score == 30_000  # first gap: 30 s
    msg = await ctx.session.get(Message, mid)
    await ctx.session.refresh(msg)
    assert msg is not None and "\\Seen" in msg.flags  # kept while retrying

    # The last try still cannot reach it: the change is rolled back.
    await sync_actions.push_flags_job(
        {"job_try": sync_actions.ACTION_MAX_TRIES}, str(mid), ["\\Seen"], []
    )
    await ctx.session.refresh(msg)
    assert "\\Seen" not in msg.flags


def test_retry_gaps_cover_about_a_day() -> None:
    total = sum(sync_actions._RETRY_DEFERS_S)
    assert 23 * 3600 < total <= 24 * 3600


async def test_superseded_flag_change_is_not_pushed(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no-untyped-def]
    # "Read" was queued, then the message was marked unread again: the stale
    # "read" must not reach the server after the newer change.
    mid = ctx.message_ids[0]
    provider = _Unreachable()
    monkeypatch.setattr(sync_actions, "build_provider", lambda _a: provider)
    await sync_actions.push_flags_job({"job_try": 3}, str(mid), ["\\Seen"], [])
    assert provider.calls == 0


async def test_archive_retries_while_unreachable(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no-untyped-def]
    provider = _Unreachable()
    monkeypatch.setattr(sync_actions, "build_provider", lambda _a: provider)
    with pytest.raises(Retry):
        await sync_actions.move_thread_to_folder_job({"job_try": 1}, str(ctx.thread_id), "archive")
    assert provider.calls == 1  # stopped at the first failure, not 20 s per message
    ctx.session.expunge_all()
    async with session_scope() as verify:
        inbox = await verify.scalar(
            select(Mailbox).where(Mailbox.user_id == ctx.user_id, Mailbox.special_use == "inbox")
        )
        assert inbox is not None
        moved = await verify.scalar(
            select(func.count())
            .select_from(Message)
            .where(Message.thread_id == ctx.thread_id, Message.mailbox_id != inbox.id)
        )
        assert moved == 0  # nothing recorded as moved


async def test_delete_keeps_rows_while_unreachable(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no-untyped-def]
    monkeypatch.setattr(sync_actions, "build_provider", lambda _a: _Unreachable())
    with pytest.raises(Retry):
        await sync_actions.delete_thread_job({"job_try": 1}, str(ctx.thread_id))
    ctx.session.expunge_all()
    async with session_scope() as verify:
        left = await verify.scalar(
            select(func.count()).select_from(Message).where(Message.thread_id == ctx.thread_id)
        )
        assert left == len(ctx.message_ids)  # kept: the server still has them