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 \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"]