Copy directly-sent e-mails to an IMAP Sent folder, with a live folder picker

A raw SMTP send has no server-side "Sent" copy on its own (unlike IMAP
drafts, which are inherently server-side) -- add an opt-in checkbox plus a
configurable target folder so directly sent Mahnungen/SEPA-info-mails still
show up in the account's Gesendet/Sent folder like a normal mail client
would leave them. Applies only to "send" delivery; drafts already live on
the server by definition.

Both the Entwürfe- and Gesendet-folder fields are now editable comboboxes:
a new "Ordnerliste laden" button fetches the real folder list from the IMAP
server (needs working credentials first) via LIST, decoding folder names
from modified UTF-7 (RFC 3501) so names like "Entwürfe" render correctly
instead of as "Entw&APw-rfe". Free text still works -- ensure_imap_folder()
creates the folder on first use if it doesn't exist yet, checked once per
batch rather than before every single message.

mail_delivery.append_to_imap_drafts() became the more general
append_message(client, content, folder=, flags=), reused for both the
\Draft and \Seen cases.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Marcel Peterkau
2026-08-15 04:34:34 +02:00
co-authored by Claude Sonnet 5
parent 6b0da82b45
commit cc4aaef895
9 changed files with 505 additions and 28 deletions
+117 -3
View File
@@ -1,6 +1,9 @@
from __future__ import annotations from __future__ import annotations
import base64
import binascii
import imaplib import imaplib
import re
import smtplib import smtplib
import ssl import ssl
from collections.abc import Iterator from collections.abc import Iterator
@@ -90,16 +93,127 @@ def send_via_smtp(client: smtplib.SMTP, message_bytes: bytes) -> None:
raise RepositoryError(f"E-Mail konnte nicht versandt werden: {exc}") from exc raise RepositoryError(f"E-Mail konnte nicht versandt werden: {exc}") from exc
def append_to_imap_drafts(client: imaplib.IMAP4, message_bytes: bytes, *, folder: str) -> None: def append_message(client: imaplib.IMAP4, message_bytes: bytes, *, folder: str, flags: str) -> None:
try: try:
status, response = client.append(folder, r"(\Draft)", None, message_bytes) status, response = client.append(_encode_imap_utf7(folder), flags, None, message_bytes)
except (OSError, imaplib.IMAP4.error) as exc: except (OSError, imaplib.IMAP4.error) as exc:
raise RepositoryError(f"E-Mail konnte nicht als Entwurf abgelegt werden: {exc}") from exc raise RepositoryError(f"E-Mail konnte nicht in „{folder} abgelegt werden: {exc}") from exc
if status != "OK": if status != "OK":
detail = response[0].decode("utf-8", "replace") if response and response[0] else status detail = response[0].decode("utf-8", "replace") if response and response[0] else status
raise RepositoryError(f"IMAP-Server hat die Ablage im Ordner „{folder}“ abgelehnt: {detail}") raise RepositoryError(f"IMAP-Server hat die Ablage im Ordner „{folder}“ abgelehnt: {detail}")
def ensure_imap_folder(client: imaplib.IMAP4, folder: str) -> None:
"""Creates the folder on the server if it doesn't exist yet -- lets someone type
a not-yet-existing Entwürfe/Gesendet folder name and have it just appear, instead
of every append failing until it's created by hand in a separate mail client."""
encoded = _encode_imap_utf7(folder)
try:
status, _response = client.select(encoded, readonly=True)
except (OSError, imaplib.IMAP4.error) as exc:
raise RepositoryError(f"IMAP-Ordner „{folder}“ konnte nicht geprüft werden: {exc}") from exc
if status == "OK":
return
try:
status, response = client.create(encoded)
except (OSError, imaplib.IMAP4.error) as exc:
raise RepositoryError(f"IMAP-Ordner „{folder}“ konnte nicht angelegt werden: {exc}") from exc
if status != "OK":
detail = response[0].decode("utf-8", "replace") if response and response[0] else status
raise RepositoryError(f"IMAP-Ordner „{folder}“ konnte nicht angelegt werden: {detail}")
_LIST_RESPONSE_RE = re.compile(
rb'^\((?P<flags>[^)]*)\)\s+(?:"(?P<qdelim>[^"]*)"|(?P<delim>\S+))\s+(?P<name>.+)$'
)
def list_imap_folders(settings: dict) -> list[str]:
"""Human-readable (decoded) folder names for the Options dialog's folder picker
-- fetched live from the server, so it only works once credentials are entered."""
with imap_session(settings) as client:
try:
status, response = client.list()
except (OSError, imaplib.IMAP4.error) as exc:
raise RepositoryError(f"Ordnerliste konnte nicht abgerufen werden: {exc}") from exc
if status != "OK":
raise RepositoryError("Ordnerliste konnte nicht abgerufen werden.")
folders = []
for raw in response or []:
if not raw:
continue
name = _parse_list_response(raw)
if name:
folders.append(name)
return folders
def _parse_list_response(raw: bytes) -> str | None:
match = _LIST_RESPONSE_RE.match(raw)
if not match:
return None
name = match.group("name").decode("utf-8", "replace").strip()
if name.startswith('"') and name.endswith('"') and len(name) >= 2:
name = name[1:-1]
return _decode_imap_utf7(name)
def _decode_imap_utf7(value: str) -> str:
# Modified UTF-7 (RFC 3501 5.1.3): "&" takes the role of "+", and "," takes the
# role of "/" inside the base64 run; "&-" is a literal ampersand.
if "&" not in value:
return value
result: list[str] = []
index = 0
length = len(value)
while index < length:
char = value[index]
if char != "&":
result.append(char)
index += 1
continue
end = value.find("-", index + 1)
if end == -1:
end = length
chunk = value[index + 1 : end]
if chunk == "":
result.append("&")
else:
base64_chunk = chunk.replace(",", "/")
base64_chunk += "=" * (-len(base64_chunk) % 4)
try:
result.append(base64.b64decode(base64_chunk).decode("utf-16-be"))
except (binascii.Error, UnicodeDecodeError):
result.append("&" + chunk + "-")
index = end + 1
return "".join(result)
def _encode_imap_utf7(value: str) -> str:
if "&" not in value and all(32 <= ord(char) <= 126 for char in value):
return value
result: list[str] = []
index = 0
length = len(value)
while index < length:
char = value[index]
if char == "&":
result.append("&-")
index += 1
continue
if 32 <= ord(char) <= 126:
result.append(char)
index += 1
continue
start = index
while index < length and not (32 <= ord(value[index]) <= 126):
index += 1
chunk = value[start:index]
encoded = base64.b64encode(chunk.encode("utf-16-be")).decode("ascii").rstrip("=")
result.append("&" + encoded.replace("/", ",") + "-")
return "".join(result)
def test_smtp_connection(settings: dict) -> None: def test_smtp_connection(settings: dict) -> None:
with smtp_session(settings): with smtp_session(settings):
pass pass
+19 -4
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import hashlib import hashlib
import os import os
import re import re
from contextlib import nullcontext
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime from datetime import datetime
from email.message import EmailMessage from email.message import EmailMessage
@@ -12,7 +13,13 @@ from pathlib import Path
from ccma.domain.contributions import claim_balance from ccma.domain.contributions import claim_balance
from ccma.domain.dates import format_date_for_display from ccma.domain.dates import format_date_for_display
from ccma.services.mail_delivery import append_to_imap_drafts, imap_session, send_via_smtp, smtp_session from ccma.services.mail_delivery import (
append_message,
ensure_imap_folder,
imap_session,
send_via_smtp,
smtp_session,
)
from ccma.storage.repository import MemberRepository, RepositoryError from ccma.storage.repository import MemberRepository, RepositoryError
@@ -178,12 +185,20 @@ def generate_and_send_reminder_mail(
export_temp.unlink(missing_ok=True) export_temp.unlink(missing_ok=True)
export_path = target export_path = target
elif delivery_mode == "send": elif delivery_mode == "send":
with smtp_session(repository.get_email_settings()) as client: settings = repository.get_email_settings()
send_via_smtp(client, content) sent_copy = imap_session(settings) if settings["imap_sent_enabled"] else nullcontext(None)
with smtp_session(settings) as smtp_client, sent_copy as imap_client:
send_via_smtp(smtp_client, content)
if imap_client is not None:
ensure_imap_folder(imap_client, settings["imap_sent_folder"])
append_message(
imap_client, content, folder=settings["imap_sent_folder"], flags=r"(\Seen)"
)
else: else:
settings = repository.get_email_settings() settings = repository.get_email_settings()
with imap_session(settings) as client: with imap_session(settings) as client:
append_to_imap_drafts(client, content, folder=settings["imap_drafts_folder"]) ensure_imap_folder(client, settings["imap_drafts_folder"])
append_message(client, content, folder=settings["imap_drafts_folder"], flags=r"(\Draft)")
os.replace(archive_temp, archive_path) os.replace(archive_temp, archive_path)
finally: finally:
archive_temp.unlink(missing_ok=True) archive_temp.unlink(missing_ok=True)
+29 -9
View File
@@ -10,7 +10,13 @@ from email.policy import SMTP
from email.utils import format_datetime, make_msgid from email.utils import format_datetime, make_msgid
from pathlib import Path from pathlib import Path
from ccma.services.mail_delivery import append_to_imap_drafts, imap_session, send_via_smtp, smtp_session from ccma.services.mail_delivery import (
append_message,
ensure_imap_folder,
imap_session,
send_via_smtp,
smtp_session,
)
from ccma.services.sepa import DirectDebit from ccma.services.sepa import DirectDebit
from ccma.storage.repository import MemberRepository, RepositoryError from ccma.storage.repository import MemberRepository, RepositoryError
@@ -105,18 +111,21 @@ def generate_debit_mails(
) -> tuple[list[GeneratedDebitMail], list[str]]: ) -> tuple[list[GeneratedDebitMail], list[str]]:
output: Path | None = None output: Path | None = None
email_settings: dict | None = None email_settings: dict | None = None
smtp_ctx = nullcontext(None)
imap_ctx = nullcontext(None)
if delivery_mode == "local": if delivery_mode == "local":
if not output_directory: if not output_directory:
raise RepositoryError("Kein Zielordner für die lokale Ablage angegeben.") raise RepositoryError("Kein Zielordner für die lokale Ablage angegeben.")
output = Path(output_directory) output = Path(output_directory)
output.mkdir(parents=True, exist_ok=True) output.mkdir(parents=True, exist_ok=True)
session = nullcontext(None)
elif delivery_mode == "send": elif delivery_mode == "send":
email_settings = repository.get_email_settings() email_settings = repository.get_email_settings()
session = smtp_session(email_settings) smtp_ctx = smtp_session(email_settings)
if email_settings["imap_sent_enabled"]:
imap_ctx = imap_session(email_settings)
elif delivery_mode == "drafts": elif delivery_mode == "drafts":
email_settings = repository.get_email_settings() email_settings = repository.get_email_settings()
session = imap_session(email_settings) imap_ctx = imap_session(email_settings)
else: else:
raise RepositoryError(f"Unbekannter Versandmodus: {delivery_mode}") raise RepositoryError(f"Unbekannter Versandmodus: {delivery_mode}")
@@ -126,9 +135,14 @@ def generate_debit_mails(
raise RepositoryError("In den Optionen fehlt die Gläubiger-ID.") raise RepositoryError("In den Optionen fehlt die Gläubiger-ID.")
generated: list[GeneratedDebitMail] = [] generated: list[GeneratedDebitMail] = []
warnings: list[str] = [] warnings: list[str] = []
# One connection is opened for the whole batch (a run can cover dozens of # One connection (or two, for "send" with a Sent-folder copy) is opened for the
# members) instead of reconnecting/re-authenticating per recipient. # whole batch -- a run can cover dozens of members -- instead of reconnecting/
with session as client: # re-authenticating per recipient.
with smtp_ctx as smtp_client, imap_ctx as imap_client:
if delivery_mode == "drafts":
ensure_imap_folder(imap_client, email_settings["imap_drafts_folder"])
elif delivery_mode == "send" and imap_client is not None:
ensure_imap_folder(imap_client, email_settings["imap_sent_folder"])
for debit in debits: for debit in debits:
member = repository.get_member(debit.member_id) member = repository.get_member(debit.member_id)
if not member.email.strip(): if not member.email.strip():
@@ -156,9 +170,15 @@ def generate_debit_mails(
export_path = _available_path(output, filename) export_path = _available_path(output, filename)
export_path.write_bytes(content) export_path.write_bytes(content)
elif delivery_mode == "send": elif delivery_mode == "send":
send_via_smtp(client, content) send_via_smtp(smtp_client, content)
if imap_client is not None:
append_message(
imap_client, content, folder=email_settings["imap_sent_folder"], flags=r"(\Seen)"
)
else: else:
append_to_imap_drafts(client, content, folder=email_settings["imap_drafts_folder"]) append_message(
imap_client, content, folder=email_settings["imap_drafts_folder"], flags=r"(\Draft)"
)
try: try:
archive_path.write_bytes(content) archive_path.write_bytes(content)
except OSError: except OSError:
+14
View File
@@ -154,6 +154,10 @@ DEFAULT_CONFIGURATION = {
"imap_username": "", "imap_username": "",
"imap_password": "", "imap_password": "",
"imap_drafts_folder": "INBOX.Entwürfe", "imap_drafts_folder": "INBOX.Entwürfe",
# Sent-copy applies only to "send" (a direct SMTP send has no server-side
# "Sent" copy on its own, unlike drafts which are inherently server-side).
"imap_sent_enabled": False,
"imap_sent_folder": "INBOX.Sent",
}, },
} }
@@ -2660,6 +2664,8 @@ class MemberRepository:
"imap_drafts_folder": str( "imap_drafts_folder": str(
settings.get("imap_drafts_folder", defaults["imap_drafts_folder"]) settings.get("imap_drafts_folder", defaults["imap_drafts_folder"])
), ),
"imap_sent_enabled": bool(settings.get("imap_sent_enabled", defaults["imap_sent_enabled"])),
"imap_sent_folder": str(settings.get("imap_sent_folder", defaults["imap_sent_folder"])),
} }
def save_email_settings( def save_email_settings(
@@ -2677,6 +2683,8 @@ class MemberRepository:
imap_username: str, imap_username: str,
imap_password: str, imap_password: str,
imap_drafts_folder: str, imap_drafts_folder: str,
imap_sent_enabled: bool = False,
imap_sent_folder: str = "",
) -> None: ) -> None:
if delivery_mode not in {"local", "send", "drafts", "ask"}: if delivery_mode not in {"local", "send", "drafts", "ask"}:
raise RepositoryError("Ungültiger Versandmodus.") raise RepositoryError("Ungültiger Versandmodus.")
@@ -2688,6 +2696,10 @@ class MemberRepository:
raise RepositoryError("Für den direkten Versand ist ein SMTP-Server erforderlich.") raise RepositoryError("Für den direkten Versand ist ein SMTP-Server erforderlich.")
if delivery_mode in {"drafts", "ask"} and not imap_host.strip(): if delivery_mode in {"drafts", "ask"} and not imap_host.strip():
raise RepositoryError("Für die Ablage als Entwurf ist ein IMAP-Server erforderlich.") raise RepositoryError("Für die Ablage als Entwurf ist ein IMAP-Server erforderlich.")
if imap_sent_enabled and not imap_host.strip():
raise RepositoryError(
"Für die Ablage im Gesendet-Ordner ist ein IMAP-Server erforderlich."
)
for label, port in (("Der SMTP-Port", smtp_port), ("Der IMAP-Port", imap_port)): for label, port in (("Der SMTP-Port", smtp_port), ("Der IMAP-Port", imap_port)):
if port < 1 or port > 65535: if port < 1 or port > 65535:
raise RepositoryError(f"{label} muss zwischen 1 und 65535 liegen.") raise RepositoryError(f"{label} muss zwischen 1 und 65535 liegen.")
@@ -2705,6 +2717,8 @@ class MemberRepository:
"imap_username": imap_username.strip(), "imap_username": imap_username.strip(),
"imap_password": imap_password, "imap_password": imap_password,
"imap_drafts_folder": imap_drafts_folder.strip() or "INBOX.Entwürfe", "imap_drafts_folder": imap_drafts_folder.strip() or "INBOX.Entwürfe",
"imap_sent_enabled": bool(imap_sent_enabled),
"imap_sent_folder": imap_sent_folder.strip() or "INBOX.Sent",
} }
write_json_atomic(self.root / "repository.json", config) write_json_atomic(self.root / "repository.json", config)
+48 -7
View File
@@ -12,7 +12,7 @@ from ccma.services.intervals import (
IntervalValidationError, IntervalValidationError,
normalize_anniversary_intervals, normalize_anniversary_intervals,
) )
from ccma.services.mail_delivery import test_imap_connection, test_smtp_connection from ccma.services.mail_delivery import list_imap_folders, test_imap_connection, test_smtp_connection
from ccma.storage.repository import MemberRepository, RepositoryError, validate_member_number_pattern from ccma.storage.repository import MemberRepository, RepositoryError, validate_member_number_pattern
from ccma.ui.changelog_view import ChangelogView from ccma.ui.changelog_view import ChangelogView
from ccma.ui.icons import IconStore from ccma.ui.icons import IconStore
@@ -649,7 +649,7 @@ class OptionsDialog(tk.Toplevel):
row=4, column=0, columnspan=4, sticky="w", pady=(8, 0) row=4, column=0, columnspan=4, sticky="w", pady=(8, 0)
) )
imap_frame = ttk.LabelFrame(parent, text="IMAP (für die Ablage als Entwurf)", padding=10) imap_frame = ttk.LabelFrame(parent, text="IMAP (für Entwürfe & Gesendet)", padding=10)
imap_frame.grid(row=3, column=0, columnspan=2, sticky="ew", pady=(16, 0)) imap_frame.grid(row=3, column=0, columnspan=2, sticky="ew", pady=(16, 0))
imap_frame.columnconfigure(1, weight=1) imap_frame.columnconfigure(1, weight=1)
self.imap_host_var = tk.StringVar(value=email_settings["imap_host"]) self.imap_host_var = tk.StringVar(value=email_settings["imap_host"])
@@ -660,6 +660,8 @@ class OptionsDialog(tk.Toplevel):
self.imap_username_var = tk.StringVar(value=email_settings["imap_username"]) self.imap_username_var = tk.StringVar(value=email_settings["imap_username"])
self.imap_password_var = tk.StringVar(value=email_settings["imap_password"]) self.imap_password_var = tk.StringVar(value=email_settings["imap_password"])
self.imap_drafts_folder_var = tk.StringVar(value=email_settings["imap_drafts_folder"]) self.imap_drafts_folder_var = tk.StringVar(value=email_settings["imap_drafts_folder"])
self.imap_sent_enabled_var = tk.BooleanVar(value=email_settings["imap_sent_enabled"])
self.imap_sent_folder_var = tk.StringVar(value=email_settings["imap_sent_folder"])
ttk.Label(imap_frame, text="Server").grid(row=0, column=0, sticky="w", pady=4, padx=(0, 8)) ttk.Label(imap_frame, text="Server").grid(row=0, column=0, sticky="w", pady=4, padx=(0, 8))
ttk.Entry(imap_frame, textvariable=self.imap_host_var).grid( ttk.Entry(imap_frame, textvariable=self.imap_host_var).grid(
row=0, column=1, sticky="ew", pady=4, padx=(0, 8) row=0, column=1, sticky="ew", pady=4, padx=(0, 8)
@@ -685,12 +687,33 @@ class OptionsDialog(tk.Toplevel):
row=3, column=1, columnspan=3, sticky="ew", pady=4 row=3, column=1, columnspan=3, sticky="ew", pady=4
) )
ttk.Label(imap_frame, text="Entwürfe-Ordner").grid(row=4, column=0, sticky="w", pady=4, padx=(0, 8)) ttk.Label(imap_frame, text="Entwürfe-Ordner").grid(row=4, column=0, sticky="w", pady=4, padx=(0, 8))
ttk.Entry(imap_frame, textvariable=self.imap_drafts_folder_var).grid( self.imap_drafts_folder_combo = ttk.Combobox(imap_frame, textvariable=self.imap_drafts_folder_var)
row=4, column=1, columnspan=3, sticky="ew", pady=4 self.imap_drafts_folder_combo.grid(row=4, column=1, columnspan=3, sticky="ew", pady=4)
) ttk.Checkbutton(
ttk.Button(imap_frame, text="Verbindung testen", command=self._test_imap_connection).grid( imap_frame,
row=5, column=0, columnspan=4, sticky="w", pady=(8, 0) text="Direkt versandte Mahnungen/Info-Mails zusätzlich in einen Gesendet-Ordner ablegen",
variable=self.imap_sent_enabled_var,
).grid(row=5, column=0, columnspan=4, sticky="w", pady=(8, 4))
ttk.Label(imap_frame, text="Gesendet-Ordner").grid(row=6, column=0, sticky="w", pady=4, padx=(0, 8))
self.imap_sent_folder_combo = ttk.Combobox(imap_frame, textvariable=self.imap_sent_folder_var)
self.imap_sent_folder_combo.grid(row=6, column=1, columnspan=3, sticky="ew", pady=4)
buttons_row = ttk.Frame(imap_frame)
buttons_row.grid(row=7, column=0, columnspan=4, sticky="w", pady=(8, 0))
ttk.Button(buttons_row, text="Verbindung testen", command=self._test_imap_connection).pack(
side="left", padx=(0, 8)
) )
ttk.Button(buttons_row, text="Ordnerliste laden", command=self._load_imap_folders).pack(side="left")
ttk.Label(
imap_frame,
text=(
"„Ordnerliste laden“ ruft die Ordner live vom Server ab (Zugangsdaten oben müssen "
"dafür bereits ausgefüllt sein) und füllt beide Dropdowns; freie Eingabe bleibt "
"möglich -- ein dabei angegebener, noch nicht vorhandener Ordner wird bei der "
"nächsten Ablage automatisch angelegt."
),
style="Muted.TLabel",
wraplength=660,
).grid(row=8, column=0, columnspan=4, sticky="w", pady=(6, 0))
ttk.Label( ttk.Label(
parent, parent,
@@ -717,6 +740,8 @@ class OptionsDialog(tk.Toplevel):
"imap_username": self.imap_username_var.get(), "imap_username": self.imap_username_var.get(),
"imap_password": self.imap_password_var.get(), "imap_password": self.imap_password_var.get(),
"imap_drafts_folder": self.imap_drafts_folder_var.get(), "imap_drafts_folder": self.imap_drafts_folder_var.get(),
"imap_sent_enabled": self.imap_sent_enabled_var.get(),
"imap_sent_folder": self.imap_sent_folder_var.get(),
} }
def _test_smtp_connection(self) -> None: def _test_smtp_connection(self) -> None:
@@ -747,6 +772,22 @@ class OptionsDialog(tk.Toplevel):
return return
messagebox.showinfo("IMAP-Verbindung", "Verbindung und Anmeldung erfolgreich.", parent=self) messagebox.showinfo("IMAP-Verbindung", "Verbindung und Anmeldung erfolgreich.", parent=self)
def _load_imap_folders(self) -> None:
settings = self._collect_email_settings()
try:
settings["imap_port"] = int(settings["imap_port"])
except ValueError:
messagebox.showerror("Ungültiger Port", "Der IMAP-Port muss eine ganze Zahl sein.", parent=self)
return
try:
folders = list_imap_folders(settings)
except RepositoryError as exc:
messagebox.showerror("Ordnerliste konnte nicht geladen werden", str(exc), parent=self)
return
self.imap_drafts_folder_combo.configure(values=folders)
self.imap_sent_folder_combo.configure(values=folders)
messagebox.showinfo("Ordnerliste geladen", f"{len(folders)} Ordner gefunden.", parent=self)
def _choose_store(self) -> None: def _choose_store(self) -> None:
current = Path(self.store_var.get()).expanduser() if self.store_var.get().strip() else Path.home() current = Path(self.store_var.get()).expanduser() if self.store_var.get().strip() else Path.home()
initial = current if current.is_dir() else current.parent initial = current if current.is_dir() else current.parent
+29
View File
@@ -12,6 +12,8 @@ def test_email_settings_default_to_local_delivery_for_new_repositories(tmp_path)
assert settings["delivery_mode"] == "local" assert settings["delivery_mode"] == "local"
assert settings["smtp_host"] == "" assert settings["smtp_host"] == ""
assert settings["imap_host"] == "" assert settings["imap_host"] == ""
assert settings["imap_sent_enabled"] is False
assert settings["imap_sent_folder"] == "INBOX.Sent"
def test_email_settings_round_trip_through_save_and_get(tmp_path) -> None: def test_email_settings_round_trip_through_save_and_get(tmp_path) -> None:
@@ -31,6 +33,8 @@ def test_email_settings_round_trip_through_save_and_get(tmp_path) -> None:
imap_username="board@example.org", imap_username="board@example.org",
imap_password="secret", imap_password="secret",
imap_drafts_folder="INBOX.Entwürfe", imap_drafts_folder="INBOX.Entwürfe",
imap_sent_enabled=True,
imap_sent_folder="INBOX.Gesendet",
) )
settings = repository.get_email_settings() settings = repository.get_email_settings()
@@ -39,6 +43,31 @@ def test_email_settings_round_trip_through_save_and_get(tmp_path) -> None:
assert settings["smtp_port"] == 587 assert settings["smtp_port"] == 587
assert settings["smtp_password"] == "secret" assert settings["smtp_password"] == "secret"
assert settings["imap_drafts_folder"] == "INBOX.Entwürfe" assert settings["imap_drafts_folder"] == "INBOX.Entwürfe"
assert settings["imap_sent_enabled"] is True
assert settings["imap_sent_folder"] == "INBOX.Gesendet"
def test_email_settings_sent_copy_requires_imap_host_even_in_send_only_mode(tmp_path) -> None:
repository = MemberRepository(tmp_path)
repository.initialize()
with pytest.raises(RepositoryError, match="Gesendet-Ordner ist ein IMAP-Server erforderlich"):
repository.save_email_settings(
delivery_mode="send",
smtp_host="mail.example.org",
smtp_port=587,
smtp_security="starttls",
smtp_username="",
smtp_password="",
imap_host="",
imap_port=993,
imap_security="ssl",
imap_username="",
imap_password="",
imap_drafts_folder="",
imap_sent_enabled=True,
imap_sent_folder="",
)
@pytest.mark.parametrize( @pytest.mark.parametrize(
+88 -5
View File
@@ -33,6 +33,7 @@ class _FakeSmtpClient:
class _FakeImapClient: class _FakeImapClient:
instances: list["_FakeImapClient"] = [] instances: list["_FakeImapClient"] = []
default_list_result: tuple = ("OK", [])
error = Exception error = Exception
def __init__(self, host, port, **kwargs): def __init__(self, host, port, **kwargs):
@@ -43,6 +44,11 @@ class _FakeImapClient:
self.login_args: tuple | None = None self.login_args: tuple | None = None
self.appended: list[tuple] = [] self.appended: list[tuple] = []
self.append_result = ("OK", [b"APPEND completed"]) 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) _FakeImapClient.instances.append(self)
def starttls(self, *args, **kwargs): def starttls(self, *args, **kwargs):
@@ -57,6 +63,20 @@ class _FakeImapClient:
self.calls.append("append") self.calls.append("append")
return self.append_result 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): def logout(self):
self.calls.append("logout") self.calls.append("logout")
@@ -65,6 +85,7 @@ class _FakeImapClient:
def _reset_fakes(): def _reset_fakes():
_FakeSmtpClient.instances.clear() _FakeSmtpClient.instances.clear()
_FakeImapClient.instances.clear() _FakeImapClient.instances.clear()
_FakeImapClient.default_list_result = ("OK", [])
yield yield
@@ -118,23 +139,85 @@ def test_send_via_smtp_logs_in_and_sends_reusing_one_connection(monkeypatch) ->
assert len(fake.sent_messages) == 2 assert len(fake.sent_messages) == 2
def test_append_to_imap_drafts_uses_configured_folder(monkeypatch) -> None: def test_append_message_uses_configured_folder_and_flags(monkeypatch) -> None:
_patch_imap(monkeypatch) _patch_imap(monkeypatch)
with mail_delivery.imap_session(IMAP_SETTINGS) as client: with mail_delivery.imap_session(IMAP_SETTINGS) as client:
mail_delivery.append_to_imap_drafts(client, MESSAGE, folder="INBOX.Entwürfe") mail_delivery.append_message(client, MESSAGE, folder="INBOX.Entwürfe", flags=r"(\Draft)")
fake = _FakeImapClient.instances[0] fake = _FakeImapClient.instances[0]
assert fake.login_args == ("board@example.org", "secret") assert fake.login_args == ("board@example.org", "secret")
assert fake.appended == [("INBOX.Entwürfe", r"(\Draft)", None, MESSAGE)] # 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"] assert fake.calls == ["login", "append", "logout"]
def test_append_to_imap_drafts_raises_on_rejected_status(monkeypatch) -> None: def test_append_message_raises_on_rejected_status(monkeypatch) -> None:
_patch_imap(monkeypatch) _patch_imap(monkeypatch)
with mail_delivery.imap_session(IMAP_SETTINGS) as client: with mail_delivery.imap_session(IMAP_SETTINGS) as client:
client.append_result = ("NO", [b"Mailbox does not exist"]) client.append_result = ("NO", [b"Mailbox does not exist"])
with pytest.raises(RepositoryError, match="Mailbox does not exist"): with pytest.raises(RepositoryError, match="Mailbox does not exist"):
mail_delivery.append_to_imap_drafts(client, MESSAGE, folder="Missing") 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 test_smtp_session_wraps_connection_errors(monkeypatch) -> None:
+63
View File
@@ -137,3 +137,66 @@ def test_reminder_mail_send_mode_delivers_via_smtp_without_local_file(tmp_path,
assert len(sent) == 1 assert len(sent) == 1
data = repository.get_contributions(member.member_id) data = repository.get_contributions(member.member_id)
assert data.reminders[0]["status"] == "sent" assert data.reminders[0]["status"] == "sent"
def test_reminder_mail_send_mode_also_copies_to_sent_folder_when_enabled(tmp_path, monkeypatch):
import ccma.services.reminder_mail as reminder_mail_module
repository, member = _overdue_claim_repository(tmp_path / "store")
member.email = "reminder@example.org"
repository.save_member(member)
repository.save_email_settings(
delivery_mode="send",
smtp_host="mail.example.org",
smtp_port=587,
smtp_security="none",
smtp_username="",
smtp_password="",
imap_host="mail.example.org",
imap_port=993,
imap_security="ssl",
imap_username="",
imap_password="",
imap_drafts_folder="",
imap_sent_enabled=True,
imap_sent_folder="INBOX.Gesendet",
)
reminder = repository.create_reminder_draft(
member.member_id,
"claim-1",
level=1,
name="Zahlungserinnerung",
payment_deadline_days=14,
)
monkeypatch.setattr(
reminder_mail_module, "smtp_session", contextmanager(lambda settings: iter(["smtp-client"]))
)
monkeypatch.setattr(reminder_mail_module, "send_via_smtp", lambda client, content: None)
monkeypatch.setattr(
reminder_mail_module, "imap_session", contextmanager(lambda settings: iter(["imap-client"]))
)
ensured = []
appended = []
monkeypatch.setattr(
reminder_mail_module, "ensure_imap_folder", lambda client, folder: ensured.append((client, folder))
)
monkeypatch.setattr(
reminder_mail_module,
"append_message",
lambda client, content, *, folder, flags: appended.append((client, folder, flags)),
)
generate_and_send_reminder_mail(
repository,
member.member_id,
"claim-1",
reminder["reminder_id"],
delivery_mode="send",
sender_name="Verwaltung C3MA",
sender_email="verwaltung@example.org",
signature="Der Vorstand",
)
assert ensured == [("imap-client", "INBOX.Gesendet")]
assert appended == [("imap-client", "INBOX.Gesendet", r"(\Seen)")]
+98
View File
@@ -290,3 +290,101 @@ def test_generated_mail_send_mode_reuses_one_smtp_connection_for_batch(tmp_path,
assert generated[0].archive_path.is_file() assert generated[0].archive_path.is_file()
assert len(connections_opened) == 1, "one connection should be reused for the whole batch" assert len(connections_opened) == 1, "one connection should be reused for the whole batch"
assert len(sent) == len(debits) assert len(sent) == len(debits)
def test_generated_mail_send_mode_copies_to_sent_folder_reusing_one_imap_connection(
tmp_path, monkeypatch
):
import ccma.services.sepa_mail as sepa_mail_module
repository, ada = _repository(tmp_path)
second = repository.create_member(first_name="Grace", last_name="Hopper", member_number="C3-43")
second.status = "active"
second.email = "grace@example.org"
second.account_holder = "Grace Hopper"
second.iban = "DE89370400440532013000"
second.bic = "COBADEFFXXX"
second.mandate_reference = "MANDAT-43"
second.mandate_signed_at = "2025-01-10"
second.mandate_active = True
repository.save_member(second)
repository.save_contributions(
second.member_id,
ContributionData(
claims=[
{
"claim_id": "due",
"title": "Mitgliedsbeitrag 2026",
"amount": "150.00",
"due_date": "2026-01-31",
"status": "open",
}
]
),
)
organization = repository.get_configuration()["organization"]
organization.update(
{
"name": "Chaos Computer Club Mannheim e.V.",
"email": "verwaltung@example.org",
"iban": "DE89370400440532013000",
"bic": "COBADEFFXXX",
"creditor_id": "DE98ZZZ09999999999",
}
)
repository.save_organization(organization)
repository.save_email_settings(
delivery_mode="send",
smtp_host="mail.example.org",
smtp_port=587,
smtp_security="none",
smtp_username="",
smtp_password="",
imap_host="mail.example.org",
imap_port=993,
imap_security="ssl",
imap_username="",
imap_password="",
imap_drafts_folder="",
imap_sent_enabled=True,
imap_sent_folder="INBOX.Gesendet",
)
debits, _warnings = pending_direct_debits(repository, due_until=date(2026, 12, 31))
assert len(debits) == 2
monkeypatch.setattr(
sepa_mail_module, "smtp_session", contextmanager(lambda settings: iter(["smtp-client"]))
)
monkeypatch.setattr(sepa_mail_module, "send_via_smtp", lambda client, content: None)
imap_connections = []
@contextmanager
def fake_imap_session(settings):
imap_connections.append(settings)
yield "imap-client"
monkeypatch.setattr(sepa_mail_module, "imap_session", fake_imap_session)
ensured = []
appended = []
monkeypatch.setattr(
sepa_mail_module, "ensure_imap_folder", lambda client, folder: ensured.append(folder)
)
monkeypatch.setattr(
sepa_mail_module,
"append_message",
lambda client, content, *, folder, flags: appended.append((client, folder, flags)),
)
generate_debit_mails(
repository,
debits,
collection_date=date(2026, 8, 3),
delivery_mode="send",
sender_name="Verwaltung C3MA",
sender_email="verwaltung@example.org",
signature="Der Vorstand",
)
assert len(imap_connections) == 1, "one IMAP connection should be reused for the whole batch"
assert ensured == ["INBOX.Gesendet"]
assert appended == [("imap-client", "INBOX.Gesendet", r"(\Seen)")] * 2