lukasandClaude Opus 4.8
a5ec0d21cb
Three-pane webmail client. - Backend: FastAPI + async SQLAlchemy + arq worker (Postgres/Redis), IMAP/SMTP sync, threading, search, drafts/outbox with undo, attachments, 2FA. - Frontend: React 18 + Vite + TypeScript + TanStack Query + Tailwind; three-pane UI with rich-text composer, light/dark themes, resizable panels. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
88 lines
2.9 KiB
Python
88 lines
2.9 KiB
Python
"""Test helper: a user with two accounts/mailboxes + a message factory.
|
|
|
|
Kept out of ``conftest.py`` so tests can import :class:`ThreadingEnv` without
|
|
re-importing the conftest module.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models import Account, Mailbox, Message, User
|
|
from app.threading_.threader import thread_message
|
|
|
|
BASE = datetime(2026, 1, 1, 12, 0, 0, tzinfo=UTC)
|
|
|
|
|
|
class ThreadingEnv:
|
|
"""A user with two accounts + one mailbox each, plus a message factory."""
|
|
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
self.user_id: uuid.UUID
|
|
self.account_a: uuid.UUID
|
|
self.account_b: uuid.UUID
|
|
self.mailbox_a: uuid.UUID
|
|
self.mailbox_b: uuid.UUID
|
|
self._uid = 0
|
|
|
|
async def setup(self) -> None:
|
|
user = User(handle=f"t{uuid.uuid4().hex[:8]}@test.local", password_hash="x")
|
|
self.session.add(user)
|
|
await self.session.flush()
|
|
self.user_id = user.id
|
|
for label in ("a", "b"):
|
|
account = Account(
|
|
user_id=self.user_id,
|
|
email=f"{label}@example.com",
|
|
imap_config={},
|
|
smtp_config={},
|
|
secret_ct=b"",
|
|
discovery_src="manual",
|
|
)
|
|
self.session.add(account)
|
|
await self.session.flush()
|
|
mailbox = Mailbox(account_id=account.id, user_id=self.user_id, path="INBOX")
|
|
self.session.add(mailbox)
|
|
await self.session.flush()
|
|
setattr(self, f"account_{label}", account.id)
|
|
setattr(self, f"mailbox_{label}", mailbox.id)
|
|
|
|
async def add(
|
|
self,
|
|
*,
|
|
message_id: str | None,
|
|
subject: str = "Subject",
|
|
references: list[str] | None = None,
|
|
in_reply_to: str | None = None,
|
|
from_email: str = "alice@example.com",
|
|
to_email: str = "bob@example.com",
|
|
minutes: int = 0,
|
|
account: str = "a",
|
|
flags: list[str] | None = None,
|
|
) -> Message:
|
|
self._uid += 1
|
|
msg = Message(
|
|
user_id=self.user_id,
|
|
account_id=getattr(self, f"account_{account}"),
|
|
mailbox_id=getattr(self, f"mailbox_{account}"),
|
|
uid=self._uid,
|
|
message_id_hdr=message_id,
|
|
in_reply_to=in_reply_to,
|
|
references_hdr=references or [],
|
|
subject=subject,
|
|
from_addr=[{"name": "", "email": from_email}],
|
|
to_addr=[{"name": "", "email": to_email}],
|
|
internal_date=BASE + timedelta(minutes=minutes),
|
|
flags=flags or [],
|
|
blob_sha256="0" * 64,
|
|
snippet=subject,
|
|
)
|
|
self.session.add(msg)
|
|
await self.session.flush()
|
|
await thread_message(self.session, msg)
|
|
return msg
|