webmail-client/backend/loadtest/mock_provider.py

"""In-process mock ``MailProvider`` for the sync soak (acceptance §10.7).

No sockets, no network: ``idle`` sleeps and returns "no change", ``poll`` returns
an empty delta. Two process-global counters make the soak's assertions meaningful
without real file descriptors:

* ``persistent_open`` — providers currently open that have issued an IDLE, i.e. the
  hot-tier persistent connections. Warm/cold runners never call ``idle`` so they
  must not contribute here.
* ``open`` — every currently-open provider (persistent + transient poll passes).

``last_sync[account_id]`` records the wall-clock of the account's most recent poll,
so the harness can compute the oldest sync lag across all accounts.
"""

from __future__ import annotations

import asyncio
import random
import time
import uuid
from collections.abc import AsyncIterator
from datetime import UTC, datetime
from email.message import EmailMessage
from email.utils import format_datetime

from app.providers.base import MailboxDelta, RemoteMailbox, RemoteMessage


class MockStats:
    open = 0
    persistent_open = 0
    last_sync: dict[uuid.UUID, float] = {}

    # --- optional mail injection (exercises parse/blob/DB/index under load) ----
    mail_enabled = False
    mail_per_poll = 0.0  # expected new INBOX messages per poll (fractional => prob)
    msg_bytes = 10_000
    big_pct = 0  # percent of messages that exceed STREAM_PARSE_MB (streaming path)
    big_bytes = 10_000_000
    delivered = 0  # total synthetic messages handed to the sync engine
    _uid: dict[uuid.UUID, int] = {}

    @classmethod
    def reset(cls) -> None:
        cls.open = 0
        cls.persistent_open = 0
        cls.last_sync = {}
        cls.delivered = 0
        cls._uid = {}


def _make_raw(size_bytes: int) -> bytes:
    """A unique, well-formed RFC822 message padded to roughly ``size_bytes``."""
    msg = EmailMessage()
    msg["Subject"] = f"soak {uuid.uuid4().hex[:12]}"
    msg["From"] = "sender@load.local"
    msg["To"] = "rcpt@load.local"
    msg["Message-ID"] = f"<{uuid.uuid4().hex}@load.local>"
    msg["Date"] = format_datetime(datetime.now(UTC))
    msg.set_content("x" * max(0, size_bytes - 256))
    return msg.as_bytes()


def _draw_count() -> int:
    n = int(MockStats.mail_per_poll)
    frac = MockStats.mail_per_poll - n
    return n + (1 if random.random() < frac else 0)


class MockProvider:
    """A MailProvider that simulates a quiet mailbox with no real I/O."""

    def __init__(self, account_id: uuid.UUID, idle_s: float) -> None:
        self.account_id = account_id
        self.idle_s = idle_s
        self._open = False
        self._persistent = False

    def _open_socket(self) -> None:
        if not self._open:
            self._open = True
            MockStats.open += 1

    async def idle(self, mailbox: str) -> None:
        self._open_socket()
        if not self._persistent:  # first IDLE marks this a persistent (hot) connection
            self._persistent = True
            MockStats.persistent_open += 1
        await asyncio.sleep(self.idle_s)  # simulate the IDLE wait / 25-min re-issue

    async def poll(
        self, mailbox: str, *, last_seen_uid: int, highestmodseq: int | None
    ) -> MailboxDelta:
        self._open_socket()
        MockStats.last_sync[self.account_id] = time.time()
        new: list[RemoteMessage] = []
        if MockStats.mail_enabled and mailbox == "INBOX":
            for _ in range(_draw_count()):
                uid = MockStats._uid.get(self.account_id, 0) + 1
                MockStats._uid[self.account_id] = uid
                big = random.random() < MockStats.big_pct / 100.0
                size = MockStats.big_bytes if big else MockStats.msg_bytes
                new.append(
                    RemoteMessage(
                        uid=uid,
                        raw=_make_raw(size),
                        flags=frozenset(),
                        internal_date=datetime.now(UTC),
                    )
                )
            MockStats.delivered += len(new)
        return MailboxDelta(
            new=new,
            flag_changes=[],
            expunged_uids=[],
            new_uidvalidity=None,
            new_highestmodseq=highestmodseq,
        )

    async def close(self) -> None:
        if self._open:
            self._open = False
            MockStats.open -= 1
        if self._persistent:
            self._persistent = False
            MockStats.persistent_open -= 1

    # --- unused in the steady-state soak, defined for interface completeness ----

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

    async def list_mailboxes(self) -> list[RemoteMailbox]:
        return []

    async def backfill(
        self, mailbox: str, *, newest_first: bool = True, batch_size: int = 200
    ) -> AsyncIterator[list[RemoteMessage]]:
        return
        yield  # pragma: no cover - makes this an async generator

    async def push_flags(
        self, mailbox: str, uid: int, add: frozenset[str], remove: frozenset[str]
    ) -> None:
        return None

    async def move(self, uid: int, src: str, dst: str) -> int:
        return uid

    async def delete_messages(self, mailbox: str, uids: list[int]) -> None:
        return None

    async def append(self, mailbox: str, raw: bytes, flags: frozenset[str]) -> int:
        return 0

    async def send(self, raw: bytes, envelope_from: str, envelope_to: list[str]) -> None:
        return None