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

"""A signed-in user with an API client, for endpoint tests in this package."""

from __future__ import annotations

import datetime as dt
import uuid
from collections.abc import AsyncIterator
from dataclasses import dataclass
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, Mailbox, Message, User
from app.providers.base import RemoteMessage

CSRF = "csrf"
HDR = {"X-CSRF": CSRF}


@dataclass
class UserCtx:
    client: httpx.AsyncClient
    user_id: uuid.UUID
    session: AsyncSession  # for arranging rows directly


@pytest_asyncio.fixture
async def user_ctx() -> AsyncIterator[UserCtx]:
    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
    sm = async_sessionmaker(engine, expire_on_commit=False)
    session = sm()
    user = User(handle=f"t{uuid.uuid4().hex[:10]}@test.local", password_hash=hash_password("pw"))
    session.add(user)
    await session.commit()
    user_id = user.id  # kept: a test may expire the session's objects
    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 UserCtx(client=client, user_id=user_id, session=session)
    finally:
        await client.aclose()
        await session.rollback()
        await session.execute(delete(User).where(User.id == user_id))
        await session.commit()
        await session.close()
        await engine.dispose()


@dataclass
class MailCtx(UserCtx):
    account_id: uuid.UUID
    inbox_id: uuid.UUID
    sent_id: uuid.UUID

    async def deliver(
        self,
        raw: bytes,
        uid: int,
        *,
        seen: bool = False,
        at: dt.datetime | None = None,
        folder: str = "inbox",
    ) -> Message:
        """Store a message the way sync does (indexer.persist_message), in the
        Inbox or (``folder="sent"``) the Sent folder."""
        remote = RemoteMessage(
            uid=uid,
            raw=raw,
            flags=frozenset({"\\Seen"}) if seen else frozenset(),
            internal_date=at or dt.datetime.now(dt.UTC),
        )
        result = await persist_message(
            self.session,
            user_id=self.user_id,
            account_id=self.account_id,
            mailbox_id=self.sent_id if folder == "sent" else self.inbox_id,
            mailbox_special_use=folder,
            remote=remote,
        )
        await self.session.commit()
        assert result is not None
        return result.message


def raw_mail(
    subject: str,
    *,
    sender: str = "News <news@shop.example>",
    headers: dict[str, str] | None = None,
    mid: str | None = None,
    html: str | None = None,
) -> bytes:
    m = EmailMessage()
    m["From"] = sender
    m["To"] = "me@example.com"
    m["Subject"] = subject
    m["Message-ID"] = mid or f"<{uuid.uuid4().hex}@shop.example>"
    for name, value in (headers or {}).items():
        m[name] = value
    m.set_content(f"{subject} body")
    if html is not None:
        m.add_alternative(html, subtype="html")
    return m.as_bytes()


@pytest_asyncio.fixture
async def mail_ctx(user_ctx: UserCtx) -> MailCtx:
    s = user_ctx.session
    account = Account(
        user_id=user_ctx.user_id,
        email="me@example.com",
        imap_config={},
        smtp_config={},
        secret_ct=b"",
        discovery_src="manual",
    )
    s.add(account)
    await s.flush()
    inbox = Mailbox(
        account_id=account.id, user_id=user_ctx.user_id, path="INBOX", special_use="inbox"
    )
    trash = Mailbox(
        account_id=account.id, user_id=user_ctx.user_id, path="Trash", special_use="trash"
    )
    sent = Mailbox(account_id=account.id, user_id=user_ctx.user_id, path="Sent", special_use="sent")
    s.add_all([inbox, trash, sent])
    await s.commit()
    return MailCtx(
        client=user_ctx.client,
        user_id=user_ctx.user_id,
        session=s,
        account_id=account.id,
        inbox_id=inbox.id,
        sent_id=sent.id,
    )