1110 lines
38 KiB
Python
1110 lines
38 KiB
Python
|
"""Tests for the Mail plugin — covers all 40 acceptance criteria."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import os
|
||
|
|
import uuid
|
||
|
|
from unittest.mock import AsyncMock, patch
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
import pytest_asyncio
|
||
|
|
from httpx import ASGITransport, AsyncClient
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
|
||
|
|
|
||
|
|
from app.core.db import close_engine, reset_engine_for_testing
|
||
|
|
from app.core.service_container import get_container
|
||
|
|
from app.main import create_app
|
||
|
|
from app.plugins.builtins.mail import MailPlugin
|
||
|
|
from app.plugins.builtins.mail.models import (
|
||
|
|
Mail,
|
||
|
|
MailAccount,
|
||
|
|
MailAttachment,
|
||
|
|
)
|
||
|
|
from app.plugins.builtins.mail.services import (
|
||
|
|
_compute_thread_id,
|
||
|
|
decrypt_password,
|
||
|
|
encrypt_password,
|
||
|
|
matches_condition,
|
||
|
|
sanitize_html,
|
||
|
|
substitute_template_vars,
|
||
|
|
)
|
||
|
|
from app.plugins.registry import reset_registry_for_testing
|
||
|
|
from app.services.plugin_service import reset_plugin_service_for_testing
|
||
|
|
from tests.conftest import ORIGIN_HEADER, login_client, seed_tenant_and_users
|
||
|
|
|
||
|
|
# ─── Mail Fixtures ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest_asyncio.fixture
|
||
|
|
async def mail_app(engine: AsyncEngine, redis_client):
|
||
|
|
"""FastAPI app with Mail plugin registered."""
|
||
|
|
reset_engine_for_testing(engine)
|
||
|
|
app = create_app()
|
||
|
|
registry = reset_registry_for_testing()
|
||
|
|
registry.initialize(engine, app)
|
||
|
|
container = get_container()
|
||
|
|
await container.initialize()
|
||
|
|
registry.register_plugin(MailPlugin())
|
||
|
|
reset_plugin_service_for_testing(registry)
|
||
|
|
# Pre-install and activate
|
||
|
|
sf = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)
|
||
|
|
async with sf() as session:
|
||
|
|
await registry.install(session, "mail")
|
||
|
|
await registry.activate(session, "mail")
|
||
|
|
await session.commit()
|
||
|
|
yield app
|
||
|
|
await close_engine()
|
||
|
|
|
||
|
|
|
||
|
|
@pytest_asyncio.fixture
|
||
|
|
async def mail_client(mail_app) -> AsyncClient:
|
||
|
|
transport = ASGITransport(app=mail_app)
|
||
|
|
async with AsyncClient(transport=transport, base_url="http://test") as c:
|
||
|
|
yield c
|
||
|
|
|
||
|
|
|
||
|
|
@pytest_asyncio.fixture
|
||
|
|
async def mail_authed_client(
|
||
|
|
mail_client: AsyncClient, db_session: AsyncSession
|
||
|
|
) -> tuple[AsyncClient, dict]:
|
||
|
|
"""Authenticated admin client with seeded data and mail plugin activated."""
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
await login_client(mail_client, "admin@tenanta.com")
|
||
|
|
return mail_client, seed
|
||
|
|
|
||
|
|
|
||
|
|
async def _create_account(client: AsyncClient, is_shared: bool = False) -> dict:
|
||
|
|
"""Helper: create a mail account via API."""
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/mail/accounts",
|
||
|
|
json={
|
||
|
|
"email_address": "test@example.com",
|
||
|
|
"display_name": "Test Account",
|
||
|
|
"imap_host": "imap.example.com",
|
||
|
|
"imap_port": 993,
|
||
|
|
"imap_ssl": True,
|
||
|
|
"smtp_host": "smtp.example.com",
|
||
|
|
"smtp_port": 587,
|
||
|
|
"smtp_tls": True,
|
||
|
|
"username": "test@example.com",
|
||
|
|
"password": "secret123",
|
||
|
|
"is_shared": is_shared,
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 201, f"Create account failed: {resp.status_code} {resp.text}"
|
||
|
|
return resp.json()
|
||
|
|
|
||
|
|
|
||
|
|
async def _get_inbox_folder(client: AsyncClient, account_id: str) -> dict:
|
||
|
|
"""Helper: get the INBOX folder for an account."""
|
||
|
|
resp = await client.get(
|
||
|
|
f"/api/v1/mail/folders?account_id={account_id}",
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
folders = resp.json()
|
||
|
|
for f in folders:
|
||
|
|
if f["imap_name"] == "INBOX":
|
||
|
|
return f
|
||
|
|
raise AssertionError("INBOX folder not found")
|
||
|
|
|
||
|
|
|
||
|
|
async def _create_mail_direct(db: AsyncSession, tenant_id, account_id, folder_id, **kwargs) -> Mail:
|
||
|
|
"""Helper: create a Mail object directly in DB."""
|
||
|
|
mail = Mail(
|
||
|
|
tenant_id=tenant_id,
|
||
|
|
account_id=account_id,
|
||
|
|
folder_id=folder_id,
|
||
|
|
message_id=kwargs.get("message_id", f"<msg-{uuid.uuid4()}@test>"),
|
||
|
|
thread_id=kwargs.get("thread_id", ""),
|
||
|
|
in_reply_to=kwargs.get("in_reply_to"),
|
||
|
|
references_header=kwargs.get("references_header"),
|
||
|
|
subject=kwargs.get("subject", "Test Subject"),
|
||
|
|
from_address=kwargs.get("from_address", "sender@example.com"),
|
||
|
|
to_addresses=kwargs.get("to_addresses", "recipient@example.com"),
|
||
|
|
cc_addresses=kwargs.get("cc_addresses", ""),
|
||
|
|
bcc_addresses=kwargs.get("bcc_addresses", ""),
|
||
|
|
body_text=kwargs.get("body_text", "Hello world"),
|
||
|
|
body_html=kwargs.get("body_html", "<p>Hello world</p>"),
|
||
|
|
body_html_sanitized=kwargs.get("body_html_sanitized", sanitize_html("<p>Hello world</p>")),
|
||
|
|
is_seen=kwargs.get("is_seen", False),
|
||
|
|
is_flagged=kwargs.get("is_flagged", False),
|
||
|
|
has_attachments=kwargs.get("has_attachments", False),
|
||
|
|
size_bytes=kwargs.get("size_bytes", 100),
|
||
|
|
)
|
||
|
|
db.add(mail)
|
||
|
|
await db.flush()
|
||
|
|
return mail
|
||
|
|
|
||
|
|
|
||
|
|
# ═══════════════════════════════════════════════════════════════
|
||
|
|
# AC Tests — grouped by feature area
|
||
|
|
# ═══════════════════════════════════════════════════════════════
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-14: Mail Account CRUD ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_list_accounts(mail_authed_client):
|
||
|
|
"""AC: GET /api/v1/mail/accounts -> 200 + account list (password not in response)."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
await _create_account(client)
|
||
|
|
resp = await client.get("/api/v1/mail/accounts", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
accounts = resp.json()
|
||
|
|
assert len(accounts) >= 1
|
||
|
|
# Password must NOT be in response
|
||
|
|
assert "password" not in accounts[0]
|
||
|
|
assert "encrypted_password" not in accounts[0]
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_create_account_password_encrypted(mail_authed_client, db_session):
|
||
|
|
"""AC: POST /api/v1/mail/accounts -> 201, password AES-256 encrypted in DB."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
# Verify in DB that password is encrypted
|
||
|
|
from sqlalchemy import select
|
||
|
|
|
||
|
|
result = await db_session.execute(
|
||
|
|
select(MailAccount).where(MailAccount.id == uuid.UUID(account["id"]))
|
||
|
|
)
|
||
|
|
db_account = result.scalar_one()
|
||
|
|
assert db_account.encrypted_password != "secret123"
|
||
|
|
assert db_account.encrypted_password != account.get("password", "")
|
||
|
|
# Verify decryption works
|
||
|
|
decrypted = decrypt_password(db_account.encrypted_password)
|
||
|
|
assert decrypted == "secret123"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_update_account(mail_authed_client):
|
||
|
|
"""AC: PATCH /api/v1/mail/accounts/{id} -> 200."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
resp = await client.patch(
|
||
|
|
f"/api/v1/mail/accounts/{account['id']}",
|
||
|
|
json={"display_name": "Updated Name", "password": "newpass456"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
assert data["display_name"] == "Updated Name"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_list_shared_accounts(mail_authed_client):
|
||
|
|
"""AC: GET /api/v1/mail/accounts/shared -> 200 + shared mailboxes."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
await _create_account(client, is_shared=True)
|
||
|
|
resp = await client.get("/api/v1/mail/accounts/shared", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
accounts = resp.json()
|
||
|
|
assert len(accounts) >= 1
|
||
|
|
assert accounts[0]["is_shared"] is True
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_assign_shared_users(mail_authed_client):
|
||
|
|
"""AC: POST /api/v1/mail/accounts/{id}/users -> 200, shared mailbox users assigned."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client, is_shared=True)
|
||
|
|
viewer_id = str(seed["viewer_a"].id)
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/mail/accounts/{account['id']}/users",
|
||
|
|
json={"user_ids": [viewer_id]},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert viewer_id in resp.json()["assigned"]
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_create_delegate(mail_authed_client):
|
||
|
|
"""AC: POST /api/v1/mail/accounts/{id}/delegates -> 200, delegate access granted."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
editor_id = str(seed["editor_a"].id)
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/mail/accounts/{account['id']}/delegates",
|
||
|
|
json={"delegate_user_id": editor_id, "access_level": "read"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["access_level"] == "read"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_create_send_permission(mail_authed_client):
|
||
|
|
"""AC: POST /api/v1/mail/accounts/{id}/send-permissions -> 200, send permission granted."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
editor_id = str(seed["editor_a"].id)
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/mail/accounts/{account['id']}/send-permissions",
|
||
|
|
json={"user_id": editor_id},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["user_id"] == editor_id
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-01, F-MAIL-19: Mail Folders ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_list_folders(mail_authed_client):
|
||
|
|
"""AC: GET /api/v1/mail/folders?account_id=X -> 200 + folder list with counts."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
resp = await client.get(
|
||
|
|
f"/api/v1/mail/folders?account_id={account['id']}",
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
folders = resp.json()
|
||
|
|
assert len(folders) >= 4 # INBOX, Sent, Drafts, Spam
|
||
|
|
assert any(f["imap_name"] == "INBOX" for f in folders)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_create_folder(mail_authed_client):
|
||
|
|
"""AC: POST /api/v1/mail/folders -> 201, folder created."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/mail/folders",
|
||
|
|
json={
|
||
|
|
"account_id": account["id"],
|
||
|
|
"name": "Custom Folder",
|
||
|
|
"imap_name": "Custom",
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 201
|
||
|
|
assert resp.json()["name"] == "Custom Folder"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_update_folder(mail_authed_client):
|
||
|
|
"""AC: PATCH /api/v1/mail/folders/{id} -> 200, renamed."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
folders = (
|
||
|
|
await client.get(f"/api/v1/mail/folders?account_id={account['id']}", headers=ORIGIN_HEADER)
|
||
|
|
).json()
|
||
|
|
folder_id = folders[0]["id"]
|
||
|
|
resp = await client.patch(
|
||
|
|
f"/api/v1/mail/folders/{folder_id}",
|
||
|
|
json={"name": "Renamed Folder"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["name"] == "Renamed Folder"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_delete_folder(mail_authed_client):
|
||
|
|
"""AC: DELETE /api/v1/mail/folders/{id} -> 204, deleted."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/mail/folders",
|
||
|
|
json={"account_id": account["id"], "name": "To Delete", "imap_name": "ToDelete"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
folder_id = resp.json()["id"]
|
||
|
|
resp = await client.delete(f"/api/v1/mail/folders/{folder_id}", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 204
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-01: Mail List & Detail ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_list_mails(mail_authed_client, db_session):
|
||
|
|
"""AC: GET /api/v1/mail?folder_id=X&page=1 -> 200 + paginated mails."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
folder = await _get_inbox_folder(client, account["id"])
|
||
|
|
# Create mail directly in DB
|
||
|
|
await _create_mail_direct(
|
||
|
|
db_session,
|
||
|
|
seed["tenant_a"].id,
|
||
|
|
uuid.UUID(account["id"]),
|
||
|
|
uuid.UUID(folder["id"]),
|
||
|
|
subject="Test Mail 1",
|
||
|
|
)
|
||
|
|
await db_session.commit()
|
||
|
|
resp = await client.get(
|
||
|
|
f"/api/v1/mail?folder_id={folder['id']}&page=1",
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
assert "mails" in data
|
||
|
|
assert "total" in data
|
||
|
|
assert data["page"] == 1
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_get_mail_detail(mail_authed_client, db_session):
|
||
|
|
"""AC: GET /api/v1/mail/{id} -> 200 + mail detail (body_html sanitized, attachments listed)."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
folder = await _get_inbox_folder(client, account["id"])
|
||
|
|
mail = await _create_mail_direct(
|
||
|
|
db_session,
|
||
|
|
seed["tenant_a"].id,
|
||
|
|
uuid.UUID(account["id"]),
|
||
|
|
uuid.UUID(folder["id"]),
|
||
|
|
subject="Detail Test",
|
||
|
|
body_html="<p>Hello</p><script>alert('xss')</script>",
|
||
|
|
body_html_sanitized=sanitize_html("<p>Hello</p><script>alert('xss')</script>"),
|
||
|
|
)
|
||
|
|
await db_session.commit()
|
||
|
|
resp = await client.get(f"/api/v1/mail/{mail.id}", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
assert data["subject"] == "Detail Test"
|
||
|
|
assert "attachments" in data
|
||
|
|
assert "labels" in data
|
||
|
|
# body_html_sanitized should NOT contain script tags
|
||
|
|
assert "<script" not in data["body_html_sanitized"]
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-02: Send, Reply, Forward ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_send_mail(mail_authed_client):
|
||
|
|
"""AC: POST /api/v1/mail/send -> 200, mail sent via SMTP."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
with patch("app.plugins.builtins.mail.services.aiosmtplib.SMTP") as mock_smtp_class:
|
||
|
|
mock_smtp_inst = AsyncMock()
|
||
|
|
mock_smtp_class.return_value = mock_smtp_inst
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/mail/send",
|
||
|
|
json={
|
||
|
|
"account_id": account["id"],
|
||
|
|
"to": ["recipient@test.com"],
|
||
|
|
"subject": "Test Send",
|
||
|
|
"body_html": "<p>Hello</p>",
|
||
|
|
"body_text": "Hello",
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["status"] == "sent"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_reply_mail(mail_authed_client, db_session):
|
||
|
|
"""AC: POST /api/v1/mail/{id}/reply -> 200, reply sent with In-Reply-To header."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
folder = await _get_inbox_folder(client, account["id"])
|
||
|
|
mail = await _create_mail_direct(
|
||
|
|
db_session,
|
||
|
|
seed["tenant_a"].id,
|
||
|
|
uuid.UUID(account["id"]),
|
||
|
|
uuid.UUID(folder["id"]),
|
||
|
|
subject="Original",
|
||
|
|
)
|
||
|
|
await db_session.commit()
|
||
|
|
with patch("app.plugins.builtins.mail.services.aiosmtplib.SMTP") as mock_smtp_class:
|
||
|
|
mock_smtp_class.return_value = AsyncMock()
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/mail/{mail.id}/reply",
|
||
|
|
json={"body_html": "<p>Reply body</p>", "body_text": "Reply body"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["status"] == "sent"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_forward_mail(mail_authed_client, db_session):
|
||
|
|
"""AC: POST /api/v1/mail/{id}/forward -> 200, forwarded with original as attachment."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
folder = await _get_inbox_folder(client, account["id"])
|
||
|
|
mail = await _create_mail_direct(
|
||
|
|
db_session,
|
||
|
|
seed["tenant_a"].id,
|
||
|
|
uuid.UUID(account["id"]),
|
||
|
|
uuid.UUID(folder["id"]),
|
||
|
|
subject="Forward Me",
|
||
|
|
)
|
||
|
|
await db_session.commit()
|
||
|
|
with patch("app.plugins.builtins.mail.services.aiosmtplib.SMTP") as mock_smtp_class:
|
||
|
|
mock_smtp_class.return_value = AsyncMock()
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/mail/{mail.id}/forward",
|
||
|
|
json={"to": ["forward@test.com"], "body_html": "<p>Fwd</p>"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["status"] == "sent"
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-09: Flags ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_update_flags(mail_authed_client, db_session):
|
||
|
|
"""AC: PATCH /api/v1/mail/{id}/flags -> 200, seen/flagged toggled."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
folder = await _get_inbox_folder(client, account["id"])
|
||
|
|
mail = await _create_mail_direct(
|
||
|
|
db_session, seed["tenant_a"].id, uuid.UUID(account["id"]), uuid.UUID(folder["id"])
|
||
|
|
)
|
||
|
|
await db_session.commit()
|
||
|
|
resp = await client.patch(
|
||
|
|
f"/api/v1/mail/{mail.id}/flags",
|
||
|
|
json={"is_seen": True, "is_flagged": True},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
assert data["is_seen"] is True
|
||
|
|
assert data["is_flagged"] is True
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-04: Attachments ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_download_attachment(mail_authed_client, db_session):
|
||
|
|
"""AC: GET /api/v1/mail/{id}/attachments/{att_id} -> 200 + file stream."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
folder = await _get_inbox_folder(client, account["id"])
|
||
|
|
mail = await _create_mail_direct(
|
||
|
|
db_session, seed["tenant_a"].id, uuid.UUID(account["id"]), uuid.UUID(folder["id"])
|
||
|
|
)
|
||
|
|
# Create attachment
|
||
|
|
test_file_path = f"/tmp/mail_test_{uuid.uuid4()}.txt"
|
||
|
|
with open(test_file_path, "w") as f:
|
||
|
|
f.write("attachment content")
|
||
|
|
att = MailAttachment(
|
||
|
|
tenant_id=seed["tenant_a"].id,
|
||
|
|
mail_id=mail.id,
|
||
|
|
filename="test.txt",
|
||
|
|
mime_type="text/plain",
|
||
|
|
size_bytes=17,
|
||
|
|
storage_path=test_file_path,
|
||
|
|
)
|
||
|
|
db_session.add(att)
|
||
|
|
await db_session.commit()
|
||
|
|
resp = await client.get(
|
||
|
|
f"/api/v1/mail/{mail.id}/attachments/{att.id}",
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
# Clean up
|
||
|
|
os.remove(test_file_path)
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-10: Contact/Company Linking ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_link_mail(mail_authed_client, db_session):
|
||
|
|
"""AC: POST /api/mail/{id}/link -> 200, manual contact/company link created."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
folder = await _get_inbox_folder(client, account["id"])
|
||
|
|
mail = await _create_mail_direct(
|
||
|
|
db_session, seed["tenant_a"].id, uuid.UUID(account["id"]), uuid.UUID(folder["id"])
|
||
|
|
)
|
||
|
|
await db_session.commit()
|
||
|
|
contact_id = str(seed["company_a"].id) # Use company as stand-in
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/mail/{mail.id}/link",
|
||
|
|
json={"company_id": contact_id},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["linked"] is True
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-11: Create Event from Mail ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_create_event_from_mail(mail_authed_client, db_session):
|
||
|
|
"""AC: POST /api/v1/mail/{id}/create-event -> 200, calendar event created from mail."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
folder = await _get_inbox_folder(client, account["id"])
|
||
|
|
mail = await _create_mail_direct(
|
||
|
|
db_session,
|
||
|
|
seed["tenant_a"].id,
|
||
|
|
uuid.UUID(account["id"]),
|
||
|
|
uuid.UUID(folder["id"]),
|
||
|
|
subject="Meeting Request",
|
||
|
|
)
|
||
|
|
await db_session.commit()
|
||
|
|
# Try without calendar plugin - should return gracefully
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/mail/{mail.id}/create-event",
|
||
|
|
json={
|
||
|
|
"calendar_id": str(uuid.uuid4()),
|
||
|
|
"title": "Meeting",
|
||
|
|
"start": "2026-07-01T10:00:00",
|
||
|
|
"end": "2026-07-01T11:00:00",
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
# Calendar plugin may not be loaded, so we accept 200 with created=False or 404
|
||
|
|
assert resp.status_code in (200, 404)
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-03: Search ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_search_mails(mail_authed_client, db_session):
|
||
|
|
"""AC: GET /api/v1/mail/search?q=text -> 200 + FTS results (body_tsv)."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
folder = await _get_inbox_folder(client, account["id"])
|
||
|
|
await _create_mail_direct(
|
||
|
|
db_session,
|
||
|
|
seed["tenant_a"].id,
|
||
|
|
uuid.UUID(account["id"]),
|
||
|
|
uuid.UUID(folder["id"]),
|
||
|
|
subject="Important Project Update",
|
||
|
|
body_text="Please review the quarterly report",
|
||
|
|
)
|
||
|
|
await _create_mail_direct(
|
||
|
|
db_session,
|
||
|
|
seed["tenant_a"].id,
|
||
|
|
uuid.UUID(account["id"]),
|
||
|
|
uuid.UUID(folder["id"]),
|
||
|
|
subject="Lunch tomorrow",
|
||
|
|
body_text="Let us grab lunch",
|
||
|
|
)
|
||
|
|
await db_session.commit()
|
||
|
|
resp = await client.get(
|
||
|
|
"/api/v1/mail/search?q=quarterly",
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
assert "results" in data
|
||
|
|
assert data["total"] >= 1
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-05: Threading ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_threads(mail_authed_client, db_session):
|
||
|
|
"""AC: GET /api/v1/mail/threads -> 200 + threaded view grouped by thread_id."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
folder = await _get_inbox_folder(client, account["id"])
|
||
|
|
# Create threaded mails
|
||
|
|
await _create_mail_direct(
|
||
|
|
db_session,
|
||
|
|
seed["tenant_a"].id,
|
||
|
|
uuid.UUID(account["id"]),
|
||
|
|
uuid.UUID(folder["id"]),
|
||
|
|
subject="Thread Start",
|
||
|
|
thread_id="thread-123",
|
||
|
|
message_id="<msg1@test>",
|
||
|
|
)
|
||
|
|
await _create_mail_direct(
|
||
|
|
db_session,
|
||
|
|
seed["tenant_a"].id,
|
||
|
|
uuid.UUID(account["id"]),
|
||
|
|
uuid.UUID(folder["id"]),
|
||
|
|
subject="Re: Thread Start",
|
||
|
|
thread_id="thread-123",
|
||
|
|
message_id="<msg2@test>",
|
||
|
|
in_reply_to="<msg1@test>",
|
||
|
|
)
|
||
|
|
await db_session.commit()
|
||
|
|
resp = await client.get("/api/v1/mail/threads", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
threads = resp.json()
|
||
|
|
assert isinstance(threads, list)
|
||
|
|
# Should have at least one thread
|
||
|
|
assert len(threads) >= 1
|
||
|
|
# Find our thread
|
||
|
|
our_thread = [t for t in threads if t["thread_id"] == "thread-123"]
|
||
|
|
assert len(our_thread) == 1
|
||
|
|
assert our_thread[0]["mail_count"] == 2
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-06: Templates ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_create_template(mail_authed_client):
|
||
|
|
"""AC: POST /api/v1/mail/templates -> 201, template created."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/mail/templates",
|
||
|
|
json={"name": "Welcome", "subject": "Welcome {{name}}", "body_html": "<p>Hi {{name}}</p>"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 201
|
||
|
|
assert resp.json()["name"] == "Welcome"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_list_templates(mail_authed_client):
|
||
|
|
"""AC: GET /api/v1/mail/templates -> 200 + template list."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
await client.post(
|
||
|
|
"/api/v1/mail/templates",
|
||
|
|
json={"name": "T1", "subject": "S1", "body_html": "<p>B1</p>"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
resp = await client.get("/api/v1/mail/templates", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
templates = resp.json()
|
||
|
|
assert len(templates) >= 1
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-13: Signatures ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_create_signature(mail_authed_client):
|
||
|
|
"""AC: POST /api/v1/mail/signatures -> 201, signature created."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/mail/signatures",
|
||
|
|
json={"name": "Default Sig", "body_html": "<p>Best regards</p>", "is_default": True},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 201
|
||
|
|
assert resp.json()["name"] == "Default Sig"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_list_signatures(mail_authed_client):
|
||
|
|
"""AC: GET /api/v1/mail/signatures -> 200 + signature list."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
await client.post(
|
||
|
|
"/api/v1/mail/signatures",
|
||
|
|
json={"name": "S1", "body_html": "<p>Sig</p>"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
resp = await client.get("/api/v1/mail/signatures", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
sigs = resp.json()
|
||
|
|
assert len(sigs) >= 1
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-07: Rules ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_create_rule(mail_authed_client):
|
||
|
|
"""AC: POST /api/v1/mail/rules -> 201, rule created with conditions+actions."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/mail/rules",
|
||
|
|
json={
|
||
|
|
"name": "Move spam",
|
||
|
|
"priority": 1,
|
||
|
|
"conditions": {"from_contains": "spam@example.com"},
|
||
|
|
"actions": {"mark_seen": True},
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 201
|
||
|
|
assert resp.json()["name"] == "Move spam"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_list_rules(mail_authed_client):
|
||
|
|
"""AC: GET /api/v1/mail/rules -> 200 + rule list sorted by priority."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
await client.post(
|
||
|
|
"/api/v1/mail/rules",
|
||
|
|
json={"name": "R2", "priority": 5, "conditions": {}, "actions": {}},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
await client.post(
|
||
|
|
"/api/v1/mail/rules",
|
||
|
|
json={"name": "R1", "priority": 1, "conditions": {}, "actions": {}},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
resp = await client.get("/api/v1/mail/rules", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
rules = resp.json()
|
||
|
|
assert len(rules) >= 2
|
||
|
|
# Should be sorted by priority
|
||
|
|
priorities = [r["priority"] for r in rules]
|
||
|
|
assert priorities == sorted(priorities)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_delete_rule(mail_authed_client):
|
||
|
|
"""AC: DELETE /api/v1/mail/rules/{id} -> 204."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/mail/rules",
|
||
|
|
json={"name": "ToDelete", "priority": 0, "conditions": {}, "actions": {}},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
rule_id = resp.json()["id"]
|
||
|
|
resp = await client.delete(f"/api/v1/mail/rules/{rule_id}", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 204
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-08: Vacation ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_vacation_config(mail_authed_client):
|
||
|
|
"""AC: POST /api/v1/mail/vacation -> 200, vacation auto-reply configured."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/mail/vacation",
|
||
|
|
json={
|
||
|
|
"account_id": account["id"],
|
||
|
|
"is_enabled": True,
|
||
|
|
"subject": "Out of Office",
|
||
|
|
"body_text": "I am away.",
|
||
|
|
"body_html": "<p>I am away.</p>",
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["configured"] is True
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_vacation_dedup(mail_authed_client):
|
||
|
|
"""AC: Vacation dedup: second auto-reply to same sender within 24h -> not sent."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/mail/vacation/test-dedup",
|
||
|
|
params={"account_id": account["id"], "sender": "someone@test.com"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
assert data["first_should_send"] is True
|
||
|
|
assert data["second_should_send"] is False
|
||
|
|
assert data["dedup_works"] is True
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-12: PGP ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_import_pgp_key(mail_authed_client):
|
||
|
|
"""AC: POST /api/v1/mail/pgp/keys -> 201, private key imported (encrypted)."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
# Generate a test PGP key
|
||
|
|
import pgpy
|
||
|
|
|
||
|
|
key = pgpy.PGPKey.new(pgpy.constants.PubKeyAlgorithm.RSAEncryptOrSign, 1024)
|
||
|
|
private_key_armored = str(key)
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/mail/pgp/keys",
|
||
|
|
json={"private_key_armored": private_key_armored, "passphrase": ""},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 201
|
||
|
|
data = resp.json()
|
||
|
|
assert "key_id" in data
|
||
|
|
assert "public_key_armored" in data
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_contact_pgp_key(mail_authed_client):
|
||
|
|
"""AC: POST /api/v1/contacts/{id}/pgp-key -> 201, contact public key stored."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
import pgpy
|
||
|
|
|
||
|
|
key = pgpy.PGPKey.new(pgpy.constants.PubKeyAlgorithm.RSAEncryptOrSign, 1024)
|
||
|
|
pub_key_armored = str(key.pubkey)
|
||
|
|
contact_id = str(uuid.uuid4()) # fake contact
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/mail/contacts/{contact_id}/pgp-key",
|
||
|
|
json={"public_key_armored": pub_key_armored},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 201
|
||
|
|
assert resp.json()["key_id"]
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-09: Labels ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_create_label(mail_authed_client):
|
||
|
|
"""AC: POST /api/v1/mail/labels -> 201, label created."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
resp = await client.post(
|
||
|
|
"/api/v1/mail/labels",
|
||
|
|
json={"name": "Important", "color": "#ff0000"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 201
|
||
|
|
assert resp.json()["name"] == "Important"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_assign_label(mail_authed_client, db_session):
|
||
|
|
"""AC: POST /api/v1/mail/{id}/labels -> 200, label assigned."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
folder = await _get_inbox_folder(client, account["id"])
|
||
|
|
mail = await _create_mail_direct(
|
||
|
|
db_session, seed["tenant_a"].id, uuid.UUID(account["id"]), uuid.UUID(folder["id"])
|
||
|
|
)
|
||
|
|
# Create label
|
||
|
|
label_resp = await client.post(
|
||
|
|
"/api/v1/mail/labels",
|
||
|
|
json={"name": "Work", "color": "#0000ff"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
label_id = label_resp.json()["id"]
|
||
|
|
await db_session.commit()
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/mail/{mail.id}/labels",
|
||
|
|
json={"label_id": label_id},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["assigned"] is True
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-01: IMAP Sync ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_imap_sync(mail_authed_client, db_session):
|
||
|
|
"""AC: IMAP sync (ARQ job): mails fetched and stored with body_tsv."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
# Mock the IMAP sync to simulate fetching mails
|
||
|
|
with patch("app.plugins.builtins.mail.routes.imap_sync_account") as mock_sync:
|
||
|
|
mock_sync.return_value = {"synced": 3}
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/mail/accounts/{account['id']}/sync",
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.json()["synced"] == 3
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-07: Rule Engine ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_rule_engine(mail_authed_client, db_session):
|
||
|
|
"""AC: Mail rule engine: incoming mail matching condition -> action executed."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
folder = await _get_inbox_folder(client, account["id"])
|
||
|
|
mail = await _create_mail_direct(
|
||
|
|
db_session,
|
||
|
|
seed["tenant_a"].id,
|
||
|
|
uuid.UUID(account["id"]),
|
||
|
|
uuid.UUID(folder["id"]),
|
||
|
|
subject="Spam Mail",
|
||
|
|
from_address="spam@example.com",
|
||
|
|
)
|
||
|
|
# Create a rule
|
||
|
|
await client.post(
|
||
|
|
"/api/v1/mail/rules",
|
||
|
|
json={
|
||
|
|
"name": "Mark spam seen",
|
||
|
|
"priority": 1,
|
||
|
|
"conditions": {"from_contains": "spam@example.com"},
|
||
|
|
"actions": {"mark_seen": True},
|
||
|
|
},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
await db_session.commit()
|
||
|
|
resp = await client.post(
|
||
|
|
f"/api/v1/mail/{mail.id}/apply-rules",
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
data = resp.json()
|
||
|
|
assert len(data["applied_rules"]) >= 1
|
||
|
|
assert data["applied_rules"][0]["actions"]["seen"] is True
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL: HTML Sanitization ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_html_sanitization(mail_authed_client, db_session):
|
||
|
|
"""AC: Body HTML sanitized (no script tags) via DOMPurify-equivalent."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
folder = await _get_inbox_folder(client, account["id"])
|
||
|
|
raw_html = '<p>Hello</p><script>alert("xss")</script><img src=x onerror=alert(1)>'
|
||
|
|
mail = await _create_mail_direct(
|
||
|
|
db_session,
|
||
|
|
seed["tenant_a"].id,
|
||
|
|
uuid.UUID(account["id"]),
|
||
|
|
uuid.UUID(folder["id"]),
|
||
|
|
body_html=raw_html,
|
||
|
|
body_html_sanitized=sanitize_html(raw_html),
|
||
|
|
)
|
||
|
|
await db_session.commit()
|
||
|
|
resp = await client.get(f"/api/v1/mail/{mail.id}", headers=ORIGIN_HEADER)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
sanitized = resp.json()["body_html_sanitized"]
|
||
|
|
assert "<script" not in sanitized
|
||
|
|
assert "onerror" not in sanitized
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-14: Password never returned ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_password_never_in_response(mail_authed_client):
|
||
|
|
"""AC: Mail account password never returned in any API response."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client)
|
||
|
|
# GET list
|
||
|
|
resp = await client.get("/api/v1/mail/accounts", headers=ORIGIN_HEADER)
|
||
|
|
for a in resp.json():
|
||
|
|
assert "password" not in a
|
||
|
|
assert "encrypted_password" not in a
|
||
|
|
# GET shared
|
||
|
|
resp = await client.get("/api/v1/mail/accounts/shared", headers=ORIGIN_HEADER)
|
||
|
|
for a in resp.json():
|
||
|
|
assert "password" not in a
|
||
|
|
assert "encrypted_password" not in a
|
||
|
|
# PATCH
|
||
|
|
resp = await client.patch(
|
||
|
|
f"/api/v1/mail/accounts/{account['id']}",
|
||
|
|
json={"display_name": "X"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert "password" not in resp.json()
|
||
|
|
assert "encrypted_password" not in resp.json()
|
||
|
|
|
||
|
|
|
||
|
|
# ─── F-MAIL-15, F-MAIL-16: Shared Mailbox Permissions ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_shared_mailbox_read_only(mail_authed_client, db_session):
|
||
|
|
"""AC: Shared mailbox: delegated user can read but not delete
|
||
|
|
(permission=read -> 403 on DELETE)."""
|
||
|
|
client, seed = mail_authed_client
|
||
|
|
account = await _create_account(client, is_shared=True)
|
||
|
|
# Grant read delegate to editor
|
||
|
|
editor_id = str(seed["editor_a"].id)
|
||
|
|
await client.post(
|
||
|
|
f"/api/v1/mail/accounts/{account['id']}/delegates",
|
||
|
|
json={"delegate_user_id": editor_id, "access_level": "read"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
# Create a folder to try deleting
|
||
|
|
folder_resp = await client.post(
|
||
|
|
"/api/v1/mail/folders",
|
||
|
|
json={"account_id": account["id"], "name": "Test", "imap_name": "Test"},
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
folder_id = folder_resp.json()["id"]
|
||
|
|
# Login as editor and try to delete -> should be 403
|
||
|
|
editor_client = client # same client, re-login
|
||
|
|
await login_client(editor_client, "editor@tenanta.com")
|
||
|
|
resp = await editor_client.delete(
|
||
|
|
f"/api/v1/mail/folders/{folder_id}",
|
||
|
|
headers=ORIGIN_HEADER,
|
||
|
|
)
|
||
|
|
assert resp.status_code == 403
|
||
|
|
|
||
|
|
|
||
|
|
# ─── Unit Tests for Service Functions ───
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_encrypt_decrypt_password():
|
||
|
|
"""Verify AES-256 encrypt/decrypt roundtrip."""
|
||
|
|
plaintext = "my-secret-password-123"
|
||
|
|
encrypted = encrypt_password(plaintext)
|
||
|
|
assert encrypted != plaintext
|
||
|
|
decrypted = decrypt_password(encrypted)
|
||
|
|
assert decrypted == plaintext
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_sanitize_html_removes_script():
|
||
|
|
"""Verify nh3 sanitization removes script tags."""
|
||
|
|
raw = "<p>Hello</p><script>alert(1)</script>"
|
||
|
|
cleaned = sanitize_html(raw)
|
||
|
|
assert "<script" not in cleaned
|
||
|
|
assert "alert" not in cleaned
|
||
|
|
assert "<p>Hello</p>" in cleaned
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_sanitize_html_removes_onerror():
|
||
|
|
"""Verify nh3 removes onerror attributes."""
|
||
|
|
raw = "<img src=x onerror=alert(1)>"
|
||
|
|
cleaned = sanitize_html(raw)
|
||
|
|
assert "onerror" not in cleaned
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_template_substitution():
|
||
|
|
"""Verify template variable substitution."""
|
||
|
|
template = "Hello {{name}}, welcome to {{company}}!"
|
||
|
|
result = substitute_template_vars(template, {"name": "John", "company": "Acme"})
|
||
|
|
assert result == "Hello John, welcome to Acme!"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_thread_id_computation():
|
||
|
|
"""Verify thread ID from References/In-Reply-To."""
|
||
|
|
# No references -> use message_id
|
||
|
|
tid = _compute_thread_id("<msg1@test>", "", None)
|
||
|
|
assert tid == "<msg1@test>"
|
||
|
|
# With references -> use first reference
|
||
|
|
tid = _compute_thread_id("<msg2@test>", "<msg1@test> <msg0@test>", None)
|
||
|
|
assert tid == "<msg1@test>"
|
||
|
|
# With in_reply_to -> use it
|
||
|
|
tid = _compute_thread_id("<msg2@test>", "", "<msg1@test>")
|
||
|
|
assert tid == "<msg1@test>"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_rule_condition_matching():
|
||
|
|
"""Verify rule condition matching logic."""
|
||
|
|
|
||
|
|
# Create a mock-like mail object
|
||
|
|
class FakeMail:
|
||
|
|
from_address = "spam@example.com"
|
||
|
|
subject = "Buy now!"
|
||
|
|
to_addresses = "recipient@test.com"
|
||
|
|
body_text = "Buy our product"
|
||
|
|
body_html = "<p>Buy our product</p>"
|
||
|
|
has_attachments = False
|
||
|
|
is_flagged = False
|
||
|
|
|
||
|
|
mail = FakeMail()
|
||
|
|
assert matches_condition(mail, {"from_contains": "spam@example.com"})
|
||
|
|
assert not matches_condition(mail, {"from_contains": "legit@example.com"})
|
||
|
|
assert matches_condition(mail, {"subject_contains": "Buy"})
|
||
|
|
assert not matches_condition(mail, {"subject_contains": "Newsletter"})
|
||
|
|
assert matches_condition(mail, {"from_contains": "spam@example.com", "subject_contains": "Buy"})
|
||
|
|
assert not matches_condition(
|
||
|
|
mail, {"from_contains": "spam@example.com", "subject_contains": "Newsletter"}
|
||
|
|
)
|