import base64 from datetime import datetime, timezone from bson import ObjectId from app.routers.emails import ( decode_cursor, decode_outbound_attachments, encode_cursor, sanitize_attachment_filename, summarize, ) from app.schemas import EmailAttachmentCreate def test_cursor_round_trip_preserves_sort_key(): entry = { "_id": ObjectId(), "created_at": datetime(2026, 8, 9, 12, 30, tzinfo=timezone.utc), } created_at, entry_id = decode_cursor(encode_cursor(entry)) assert created_at == entry["created_at"] assert entry_id == entry["_id"] def test_naive_cursor_timestamp_is_treated_as_utc(): naive = datetime(2026, 8, 9, 12, 30) created_at, _entry_id = decode_cursor( encode_cursor({"_id": ObjectId(), "created_at": naive}) ) assert created_at.tzinfo is not None assert created_at == datetime(2026, 8, 9, 12, 30, tzinfo=timezone.utc) def test_summary_uses_account_mailbox_read_state(): now = datetime(2026, 8, 9, 12, 30, tzinfo=timezone.utc) message_id = ObjectId() message = { "_id": message_id, "direction": "inbound", "sender": {"name": "Sender", "address": "sender@example.com"}, "to": [{"name": "user", "address": "user@gentic.email"}], "subject": "Hello", "snippet": "A short preview", "status": "received", "created_at": now, "received_at": now, "attachments": [], } entry = { "_id": ObjectId(), "message_id": message_id, "mailbox": "inbox", "is_read": False, "trashed_at": None, "created_at": now, } result = summarize(message, entry) assert result.id == str(message_id) assert result.is_read is False assert result.mailbox == "inbox" assert result.snippet == "A short preview" def test_sanitize_attachment_filename_strips_paths(): assert sanitize_attachment_filename("../../secret.pdf") == "secret.pdf" def test_decode_outbound_attachments_splits_inline_and_files(): pdf = EmailAttachmentCreate( filename="invoice.pdf", content_type="application/pdf", content_base64=base64.b64encode(b"%PDF-1.4").decode(), disposition="attachment", ) image = EmailAttachmentCreate( filename="logo.png", content_type="image/png", content_base64=base64.b64encode(b"PNG").decode(), disposition="inline", content_id="logo@agmail", ) inline_parts, files, metadata = decode_outbound_attachments([pdf, image]) assert [part.filename for part in files] == ["invoice.pdf"] assert [part.filename for part in inline_parts] == ["logo.png"] assert metadata[0]["size"] == 8 assert metadata[1]["disposition"] == "inline"