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

"""Recipient suggest, settings, and per-sender remote-image control (SPEC §9.1/§9.4)."""

from __future__ import annotations

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

import httpx
import pytest_asyncio
from httpx import ASGITransport
from sqlalchemy import delete
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool

from app.config import get_settings
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, KnownRecipient, Mailbox, User
from app.providers.base import RemoteMessage
from app.search.pg import PgSearchBackend

CSRF = "csrf"


@dataclass
class Ctx:
    client: httpx.AsyncClient
    session: AsyncSession
    user_id: uuid.UUID
    account_id: uuid.UUID
    mailbox_id: uuid.UUID


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


@pytest_asyncio.fixture
async def ctx() -> AsyncIterator[Ctx]:
    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
    sm = async_sessionmaker(engine, expire_on_commit=False)
    session = sm()
    user = User(handle=f"rs{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()
    mailbox = Mailbox(account_id=account.id, user_id=user.id, path="INBOX", special_use="inbox")
    session.add(mailbox)
    await session.commit()
    token = await create_session(user.id, "pytest")
    client = httpx.AsyncClient(
        transport=ASGITransport(app=create_app()),
        base_url="http://testserver",
        cookies={"session": token, "csrf": CSRF},
    )
    try:
        yield Ctx(
            client=client,
            session=session,
            user_id=user.id,
            account_id=account.id,
            mailbox_id=mailbox.id,
        )
    finally:
        await client.aclose()
        await session.execute(delete(User).where(User.id == user.id))
        await session.commit()
        await session.close()
        await engine.dispose()


async def test_recipient_suggest_prefix_and_ranking(ctx: Ctx) -> None:
    now = datetime.now(UTC)
    ctx.session.add_all(
        [
            KnownRecipient(
                user_id=ctx.user_id,
                email="alice@example.com",
                name="Alice",
                times_used=10,
                last_used_at=now,
            ),
            KnownRecipient(
                user_id=ctx.user_id,
                email="alan@example.com",
                name="Alan",
                times_used=1,
                last_used_at=now - timedelta(days=200),
            ),
            KnownRecipient(
                user_id=ctx.user_id,
                email="bob@example.com",
                name="Bob",
                times_used=99,
                last_used_at=now,
            ),
        ]
    )
    await ctx.session.commit()

    resp = await ctx.client.get("/api/recipients/suggest", params={"q": "al"})
    assert resp.status_code == 200
    emails = [r["email"] for r in resp.json()]
    assert set(emails) == {"alice@example.com", "alan@example.com"}  # prefix match, bob excluded
    assert emails[0] == "alice@example.com"  # higher score (recent + frequent) ranks first


async def test_recipient_suggest_empty_query(ctx: Ctx) -> None:
    resp = await ctx.client.get("/api/recipients/suggest", params={"q": ""})
    assert resp.status_code == 200 and resp.json() == []


async def test_settings_get_and_patch(ctx: Ctx) -> None:
    got = await ctx.client.get("/api/settings")
    assert got.status_code == 200
    assert got.json()["thread_order"] == "asc"

    patched = await ctx.client.patch(
        "/api/settings", json={"thread_order": "desc", "tz": "Europe/Warsaw"}, headers=_hdr()
    )
    assert patched.status_code == 200
    assert patched.json()["thread_order"] == "desc"
    assert patched.json()["tz"] == "Europe/Warsaw"

    # invalid order is ignored
    patched2 = await ctx.client.patch(
        "/api/settings", json={"thread_order": "sideways"}, headers=_hdr()
    )
    assert patched2.json()["thread_order"] == "desc"


async def _ingest_html(ctx: Ctx, *, sender: str, html: str) -> uuid.UUID:
    m = EmailMessage()
    m["Subject"] = "Newsletter"
    m["From"] = sender
    m["To"] = "me@example.com"
    m["Message-ID"] = f"<{uuid.uuid4().hex}@x.com>"
    m["Date"] = "Mon, 09 Mar 2026 08:36:37 +0100"
    m.set_content("text fallback")
    m.add_alternative(html, subtype="html")
    result = await persist_message(
        ctx.session,
        user_id=ctx.user_id,
        account_id=ctx.account_id,
        mailbox_id=ctx.mailbox_id,
        mailbox_special_use="inbox",
        remote=RemoteMessage(
            uid=1, raw=m.as_bytes(), flags=frozenset(), internal_date=datetime.now(UTC)
        ),
    )
    await ctx.session.commit()
    assert result is not None
    await PgSearchBackend().index(result.search_doc)
    return result.message.id


async def test_remote_images_blocked_then_allowed(ctx: Ctx) -> None:
    mid = await _ingest_html(
        ctx,
        sender="News <news@sender.com>",
        html='<p>Hi<img src="http://tracker.example/pixel.gif"></p>',
    )
    # blocked by default
    before = await ctx.client.get(f"/api/messages/{mid}/html")
    assert "tracker.example" not in before.text

    # approve the sender
    allow = await ctx.client.post(f"/api/messages/{mid}/allow-remote-images", headers=_hdr())
    assert allow.status_code == 200
    assert allow.json()["sender"] == "news@sender.com"

    # now the remote image is rendered
    after = await ctx.client.get(f"/api/messages/{mid}/html")
    assert "http://tracker.example/pixel.gif" in after.text