CCMA/tests/test_mail_delivery.py

import pytest

from ccma.services import mail_delivery
from ccma.storage.repository import RepositoryError


class _FakeSmtpClient:
    instances: list["_FakeSmtpClient"] = []

    def __init__(self, host, port, **kwargs):
        self.host = host
        self.port = port
        self.kwargs = kwargs
        self.calls: list[str] = []
        self.login_args: tuple | None = None
        self.sent_messages: list = []
        _FakeSmtpClient.instances.append(self)

    def starttls(self, **kwargs):
        self.calls.append("starttls")

    def login(self, username, password):
        self.login_args = (username, password)
        self.calls.append("login")

    def send_message(self, message):
        self.sent_messages.append(message)
        self.calls.append("send_message")

    def quit(self):
        self.calls.append("quit")


class _FakeImapClient:
    instances: list["_FakeImapClient"] = []
    default_list_result: tuple = ("OK", [])
    error = Exception

    def __init__(self, host, port, **kwargs):
        self.host = host
        self.port = port
        self.kwargs = kwargs
        self.calls: list[str] = []
        self.login_args: tuple | None = None
        self.appended: list[tuple] = []
        self.append_result = ("OK", [b"APPEND completed"])
        self.selected: list[str] = []
        self.select_result = ("OK", [b"1"])
        self.created: list[str] = []
        self.create_result = ("OK", [b"CREATE completed"])
        self.list_result = _FakeImapClient.default_list_result
        _FakeImapClient.instances.append(self)

    def starttls(self, *args, **kwargs):
        self.calls.append("starttls")

    def login(self, username, password):
        self.login_args = (username, password)
        self.calls.append("login")

    def append(self, folder, flags, date_time, message_bytes):
        self.appended.append((folder, flags, date_time, message_bytes))
        self.calls.append("append")
        return self.append_result

    def select(self, folder, readonly=False):
        self.selected.append(folder)
        self.calls.append("select")
        return self.select_result

    def create(self, folder):
        self.created.append(folder)
        self.calls.append("create")
        return self.create_result

    def list(self):
        self.calls.append("list")
        return self.list_result

    def logout(self):
        self.calls.append("logout")


@pytest.fixture(autouse=True)
def _reset_fakes():
    _FakeSmtpClient.instances.clear()
    _FakeImapClient.instances.clear()
    _FakeImapClient.default_list_result = ("OK", [])
    yield


def _patch_smtp(monkeypatch):
    monkeypatch.setattr(mail_delivery.smtplib, "SMTP", _FakeSmtpClient)
    monkeypatch.setattr(mail_delivery.smtplib, "SMTP_SSL", _FakeSmtpClient)


def _patch_imap(monkeypatch):
    monkeypatch.setattr(mail_delivery.imaplib, "IMAP4", _FakeImapClient)
    monkeypatch.setattr(mail_delivery.imaplib, "IMAP4_SSL", _FakeImapClient)


SMTP_SETTINGS = {
    "smtp_host": "mail.example.org",
    "smtp_port": 587,
    "smtp_security": "starttls",
    "smtp_username": "board@example.org",
    "smtp_password": "secret",
}

IMAP_SETTINGS = {
    "imap_host": "mail.example.org",
    "imap_port": 993,
    "imap_security": "ssl",
    "imap_username": "board@example.org",
    "imap_password": "secret",
}

MESSAGE = (
    b"From: Board <board@example.org>\r\n"
    b"To: member@example.org\r\n"
    b"Subject: Test\r\n"
    b"\r\n"
    b"Hello\r\n"
)


def test_send_via_smtp_logs_in_and_sends_reusing_one_connection(monkeypatch) -> None:
    _patch_smtp(monkeypatch)
    with mail_delivery.smtp_session(SMTP_SETTINGS) as client:
        mail_delivery.send_via_smtp(client, MESSAGE)
        mail_delivery.send_via_smtp(client, MESSAGE)

    assert len(_FakeSmtpClient.instances) == 1
    fake = _FakeSmtpClient.instances[0]
    assert fake.host == "mail.example.org"
    assert fake.port == 587
    assert fake.login_args == ("board@example.org", "secret")
    assert fake.calls == ["starttls", "login", "send_message", "send_message", "quit"]
    assert len(fake.sent_messages) == 2


def test_append_message_uses_configured_folder_and_flags(monkeypatch) -> None:
    _patch_imap(monkeypatch)
    with mail_delivery.imap_session(IMAP_SETTINGS) as client:
        mail_delivery.append_message(client, MESSAGE, folder="INBOX.Entwürfe", flags=r"(\Draft)")

    fake = _FakeImapClient.instances[0]
    assert fake.login_args == ("board@example.org", "secret")
    # The folder name travels the wire in modified UTF-7, not raw UTF-8.
    assert fake.appended == [("INBOX.Entw&APw-rfe", r"(\Draft)", None, MESSAGE)]
    assert fake.calls == ["login", "append", "logout"]


def test_append_message_raises_on_rejected_status(monkeypatch) -> None:
    _patch_imap(monkeypatch)
    with mail_delivery.imap_session(IMAP_SETTINGS) as client:
        client.append_result = ("NO", [b"Mailbox does not exist"])
        with pytest.raises(RepositoryError, match="Mailbox does not exist"):
            mail_delivery.append_message(client, MESSAGE, folder="Missing", flags=r"(\Seen)")


def test_ensure_imap_folder_skips_creation_when_folder_already_exists(monkeypatch) -> None:
    _patch_imap(monkeypatch)
    with mail_delivery.imap_session(IMAP_SETTINGS) as client:
        mail_delivery.ensure_imap_folder(client, "INBOX.Sent")

    fake = _FakeImapClient.instances[0]
    assert fake.selected == ["INBOX.Sent"]
    assert fake.created == []


def test_ensure_imap_folder_creates_missing_folder(monkeypatch) -> None:
    _patch_imap(monkeypatch)
    with mail_delivery.imap_session(IMAP_SETTINGS) as client:
        client.select_result = ("NO", [b"Mailbox does not exist"])
        mail_delivery.ensure_imap_folder(client, "Archiv.Neu")

    fake = _FakeImapClient.instances[0]
    assert fake.selected == ["Archiv.Neu"]
    assert fake.created == ["Archiv.Neu"]


def test_ensure_imap_folder_raises_when_creation_fails(monkeypatch) -> None:
    _patch_imap(monkeypatch)
    with mail_delivery.imap_session(IMAP_SETTINGS) as client:
        client.select_result = ("NO", [b"no such mailbox"])
        client.create_result = ("NO", [b"Permission denied"])
        with pytest.raises(RepositoryError, match="Permission denied"):
            mail_delivery.ensure_imap_folder(client, "Verboten")


@pytest.mark.parametrize(
    ("plain", "wire"),
    [
        ("INBOX", "INBOX"),
        ("Entwürfe", "Entw&APw-rfe"),
        ("INBOX.Entwürfe", "INBOX.Entw&APw-rfe"),
        ("Test & Foo", "Test &- Foo"),
    ],
)
def test_imap_utf7_round_trips(plain, wire) -> None:
    assert mail_delivery._encode_imap_utf7(plain) == wire
    assert mail_delivery._decode_imap_utf7(wire) == plain


def test_list_imap_folders_parses_and_decodes_list_response(monkeypatch) -> None:
    _patch_imap(monkeypatch)
    _FakeImapClient.default_list_result = (
        "OK",
        [
            rb'(\HasNoChildren) "." INBOX',
            rb'(\HasNoChildren) "." INBOX.Entw&APw-rfe',
            rb'(\HasNoChildren) "." "INBOX.Gesendete Objekte"',
        ],
    )

    folders = mail_delivery.list_imap_folders(IMAP_SETTINGS)

    assert folders == ["INBOX", "INBOX.Entwürfe", "INBOX.Gesendete Objekte"]
    assert _FakeImapClient.instances[0].calls == ["login", "list", "logout"]


def test_smtp_session_wraps_connection_errors(monkeypatch) -> None:
    def _boom(*args, **kwargs):
        raise OSError("connection refused")

    monkeypatch.setattr(mail_delivery.smtplib, "SMTP", _boom)
    with pytest.raises(RepositoryError, match="SMTP-Verbindung fehlgeschlagen"):
        with mail_delivery.smtp_session({**SMTP_SETTINGS, "smtp_security": "none"}):
            pass


def test_smtp_session_requires_host() -> None:
    with pytest.raises(RepositoryError, match="Kein SMTP-Server konfiguriert"):
        with mail_delivery.smtp_session({}):
            pass


def test_imap_session_requires_host() -> None:
    with pytest.raises(RepositoryError, match="Kein IMAP-Server konfiguriert"):
        with mail_delivery.imap_session({}):
            pass


def test_test_smtp_connection_succeeds_and_closes(monkeypatch) -> None:
    _patch_smtp(monkeypatch)
    mail_delivery.test_smtp_connection(SMTP_SETTINGS)
    assert _FakeSmtpClient.instances[0].calls == ["starttls", "login", "quit"]


def test_test_imap_connection_succeeds_and_closes(monkeypatch) -> None:
    _patch_imap(monkeypatch)
    mail_delivery.test_imap_connection(IMAP_SETTINGS)
    assert _FakeImapClient.instances[0].calls == ["login", "logout"]