agentic.email/backend/tests/test_email_helpers.py
2026-10-06 18:27:06 +05:30

58 lines
1.7 KiB
Python

from datetime import datetime, timezone
from bson import ObjectId
from app.routers.emails import decode_cursor, encode_cursor, summarize
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"