agentic.email/backend/tests/test_inbound_smtp.py

from types import SimpleNamespace

import pytest
from bson import ObjectId

from app.services import inbound_smtp
from app.services.raw_mime_storage import StoredMime


RAW = (
    b"From: Sender <sender@example.com>\r\n"
    b"To: user@gentic.email\r\n"
    b"Subject: Hello\r\n\r\n"
    b"Retry safe body\r\n"
)


class FakeMessages:
    def __init__(self):
        self.id = ObjectId()
        self.updates = 0

    async def update_one(self, *args, **kwargs):
        self.updates += 1
        return SimpleNamespace(upserted_id=ObjectId(), modified_count=0)

    async def find_one(self, *args, **kwargs):
        return {"_id": self.id}


class FakeEntries:
    def __init__(self):
        self.updates = 0

    async def update_one(self, *args, **kwargs):
        self.updates += 1
        return SimpleNamespace(upserted_id=ObjectId(), modified_count=0)


class FakeThreads:
    async def update_one(self, *args, **kwargs):
        return None


class FakeStorage:
    async def put(self, raw, delivery_key):
        return StoredMime("mail", f"raw/{delivery_key}.eml", 7, len(raw), "crc")


class FakeSession:
    def __init__(self, account):
        self.account = account

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        pass

    async def scalar(self, query):
        return self.account


def envelope(raw=RAW):
    return SimpleNamespace(
        original_content=raw,
        rcpt_tos=["user@gentic.email"],
        mail_from="sender@example.com",
        account_ids={"user@gentic.email": "account-1"},
    )


@pytest.mark.asyncio
async def test_rcpt_rejects_unknown_account(monkeypatch):
    monkeypatch.setattr(inbound_smtp, "SessionFactory", lambda: FakeSession(None))
    target = SimpleNamespace(rcpt_tos=[], rcpt_options=[])

    response = await inbound_smtp.InboundHandler().handle_RCPT(
        None,
        None,
        target,
        "missing@gentic.email",
        [],
    )

    assert response.startswith("550 5.1.1")
    assert target.rcpt_tos == []


@pytest.mark.asyncio
async def test_rcpt_accepts_account_and_tracks_owner(monkeypatch):
    account = SimpleNamespace(id="account-1")
    monkeypatch.setattr(inbound_smtp, "SessionFactory", lambda: FakeSession(account))
    target = SimpleNamespace(rcpt_tos=[], rcpt_options=[], mail_from="sender@example.com")

    response = await inbound_smtp.InboundHandler().handle_RCPT(
        None,
        None,
        target,
        "User@GetAll.Help",
        [],
    )

    assert response.startswith("250 2.1.5")
    assert target.rcpt_tos == ["user@getall.help"]
    assert target.account_ids == {"user@getall.help": "account-1"}


@pytest.mark.asyncio
async def test_data_returns_success_only_after_idempotent_records(monkeypatch):
    messages = FakeMessages()
    entries = FakeEntries()
    monkeypatch.setattr(
        inbound_smtp,
        "mongo_db",
        SimpleNamespace(
            messages=messages,
            mailbox_entries=entries,
            email_threads=FakeThreads(),
        ),
    )
    monkeypatch.setattr(inbound_smtp, "get_raw_mime_storage", lambda: FakeStorage())
    async def no_handoff(*args, **kwargs):
        return None

    monkeypatch.setattr(inbound_smtp, "resolve_handoff_from_inbound", no_handoff)
    monkeypatch.setattr(inbound_smtp, "maybe_trigger_handoff", no_handoff)
    async def ensure_thread(*args, **kwargs):
        return ObjectId()

    monkeypatch.setattr(inbound_smtp, "ensure_thread_for_message", ensure_thread)

    response = await inbound_smtp.InboundHandler().handle_DATA(None, None, envelope())

    assert response.startswith("250 2.0.0")
    assert messages.updates >= 1
    assert entries.updates >= 1


@pytest.mark.asyncio
async def test_human_reply_stays_on_the_private_plane(monkeypatch):
    messages = FakeMessages()
    entries = FakeEntries()
    monkeypatch.setattr(
        inbound_smtp,
        "mongo_db",
        SimpleNamespace(
            messages=messages,
            mailbox_entries=entries,
            email_threads=FakeThreads(),
        ),
    )
    monkeypatch.setattr(inbound_smtp, "get_raw_mime_storage", lambda: FakeStorage())

    async def resolve(*args, **kwargs):
        return {"_id": ObjectId()}

    processed = []

    async def process(*args, **kwargs):
        processed.append(kwargs)
        return {"_id": ObjectId()}

    monkeypatch.setattr(inbound_smtp, "resolve_handoff_from_inbound", resolve)
    monkeypatch.setattr(inbound_smtp, "process_human_reply", process)

    response = await inbound_smtp.InboundHandler().handle_DATA(None, None, envelope())

    assert response.startswith("250 2.0.0 Handoff message accepted")
    assert processed
    assert messages.updates == 0
    assert entries.updates == 0


@pytest.mark.asyncio
async def test_data_still_accepted_when_handoff_trigger_fails(monkeypatch):
    messages = FakeMessages()
    entries = FakeEntries()
    monkeypatch.setattr(
        inbound_smtp,
        "mongo_db",
        SimpleNamespace(
            messages=messages,
            mailbox_entries=entries,
            email_threads=FakeThreads(),
        ),
    )
    monkeypatch.setattr(inbound_smtp, "get_raw_mime_storage", lambda: FakeStorage())

    async def no_handoff(*args, **kwargs):
        return None

    async def boom(*args, **kwargs):
        raise RuntimeError("mirror queue unavailable")

    monkeypatch.setattr(inbound_smtp, "resolve_handoff_from_inbound", no_handoff)
    monkeypatch.setattr(inbound_smtp, "maybe_trigger_handoff", boom)

    async def ensure_thread(*args, **kwargs):
        return ObjectId()

    monkeypatch.setattr(inbound_smtp, "ensure_thread_for_message", ensure_thread)

    response = await inbound_smtp.InboundHandler().handle_DATA(None, None, envelope())

    assert response.startswith("250 2.0.0")
    assert messages.updates >= 1


@pytest.mark.asyncio
async def test_data_returns_retryable_status_on_storage_failure(monkeypatch):
    class BrokenStorage:
        async def put(self, raw, delivery_key):
            raise TimeoutError("GCS unavailable")

    monkeypatch.setattr(inbound_smtp, "get_raw_mime_storage", lambda: BrokenStorage())

    response = await inbound_smtp.InboundHandler().handle_DATA(None, None, envelope())

    assert response.startswith("451 4.3.0")


@pytest.mark.asyncio
async def test_data_rejects_oversized_message(monkeypatch):
    monkeypatch.setattr(
        inbound_smtp,
        "get_settings",
        lambda: SimpleNamespace(inbound_max_bytes=10),
    )

    response = await inbound_smtp.InboundHandler().handle_DATA(None, None, envelope())

    assert response.startswith("552 5.3.4")