Mailversand ueber Ausgangstabelle (ADR-0041)

mail.py reiht Mails in mail_ausgang ein (Migration 0009), eine
Versandschleife im Dienst liefert sie per SMTP mit Anmeldung ein und
wiederholt bei Fehlern. Voreinstellung ist Trockenlauf; das Passwort
kommt als systemd-Credential. Dazu scripts/testmail.py und Tests.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

README.md

@@ -34,6 +34,44 @@ Env-Variablen:
 
 Start: `.venv/bin/uvicorn app:app --port 8081`
 
+## Mailversand
+
+Siehe [ADR-0041](https://git.tuxflotte.de/admin/platform-docs/raw/branch/main/adr/0041-personalisierung-lokales-konto-und-geraetesperre.md).
+Aufrufer tragen Mails mit `mail.einreihen()` in die Tabelle `mail_ausgang`
+ein (Migration 0009). Eine Versandschleife im Dienst liefert sie per SMTP
+mit Anmeldung beim Mailanbieter der Domain ein und wiederholt bei Fehlern
+(nach 1, 5 und 15 Minuten, dann „fehlgeschlagen“). Der Mailtext wird nach
+dem Endzustand geleert, Einträge werden nach 90 Tagen gelöscht.
+
+Env-Variablen:
+
+- `KUNDENPLATTFORM_MAIL_MODUS` — `trockenlauf` (Default, schreibt die Mail
+  ins Journal statt sie zu verschicken) oder `smtp`
+- `KUNDENPLATTFORM_MAIL_HOST`, `KUNDENPLATTFORM_MAIL_PORT` (Default `587`,
+  STARTTLS)
+- `KUNDENPLATTFORM_MAIL_BENUTZER` — Anmeldename des Postfachs
+- `KUNDENPLATTFORM_MAIL_ABSENDER` — z.B. `Tuxflotte <konto@tuxflotte.de>`
+- `KUNDENPLATTFORM_MAIL_ANTWORT_AN` — optional
+
+Das Passwort steht nicht in der Umgebung, sondern kommt als
+systemd-Credential `mail_passwort` (siehe `kundenplattform.service`); für
+Aufrufe außerhalb des Dienstes `KUNDENPLATTFORM_MAIL_PASSWORT_DATEI`.
+
+Testmail (prüft Konfiguration und Zugangsdaten):
+
+```
+KUNDENPLATTFORM_DATABASE_URL=... KUNDENPLATTFORM_MAIL_MODUS=smtp ... \
+    .venv/bin/python scripts/testmail.py --an jemand@example.org
+```
+
+Tests (brauchen eine Wegwerfdatenbank mit „test“ im Namen und eingespielter
+Migration 0009; SMTP ist gemockt):
+
+```
+KUNDENPLATTFORM_DATABASE_URL=postgresql://.../kundenplattform_test \
+    .venv/bin/python -m unittest discover -s tests -v
+```
+
 ## Konto anlegen
 
 Kein Self-Service in v1 (siehe ADR-0011) — Konten werden manuell angelegt:

app.py

@@ -1,7 +1,11 @@
+import asyncio
+from contextlib import asynccontextmanager
+
 from fastapi import FastAPI, Request
 from fastapi.staticfiles import StaticFiles
 
 import auth
+import mail
 from routers import (
     admin_accounts,
     admin_dashboard,
@@ -26,7 +30,17 @@ from routers import (
 # erreichbar und legten die komplette interne Routenstruktur (inkl. aller
 # /admin/*-Pfade) offen - fuer eine Plattform in aktiver Entwicklung
 # unnoetige Aufklaerungshilfe fuer Angreifer.
-app = FastAPI(title="Tuxflotte Flotten-Management", docs_url=None, redoc_url=None, openapi_url=None)
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Versandschleife für die Ausgangstabelle (siehe mail.py, ADR-0041).
+    versand = asyncio.create_task(mail.versand_schleife())
+    yield
+    versand.cancel()
+
+
+app = FastAPI(
+    title="Tuxflotte Flotten-Management", docs_url=None, redoc_url=None, openapi_url=None, lifespan=lifespan
+)
 app.mount("/static", StaticFiles(directory="static"), name="static")
 
 

kundenplattform.service

@@ -8,6 +8,11 @@ User=kundenplattform
 Group=kundenplattform
 WorkingDirectory=/opt/kundenplattform
 EnvironmentFile=/etc/tuxflotte/kundenplattform.env
+# Mailversand (mail.py, ADR-0041): Das Postfach-Passwort kommt als
+# systemd-Credential, nicht über die Umgebung. Erst einkommentieren, wenn
+# die Datei existiert (root:root, 0600) - ob der Dienst ohne die Quelldatei
+# startet, ist nicht geprüft. Bis dahin läuft der Versand als Trockenlauf.
+#LoadCredential=mail_passwort:/etc/tuxflotte/kundenplattform-mail-passwort
 ExecStart=/opt/kundenplattform/.venv/bin/uvicorn app:app --host 0.0.0.0 --port 8081
 Restart=always
 RestartSec=3

mail.py

@@ -0,0 +1,308 @@
+import asyncio
+import logging
+import os
+import smtplib
+import ssl
+import sys
+import time
+import uuid
+from dataclasses import dataclass
+from email.message import EmailMessage
+from email.utils import formatdate, make_msgid, parseaddr
+from pathlib import Path
+
+from email_validator import EmailNotValidError, validate_email
+
+from db import get_database_connection
+
+
+# Mailversand der Plattform (ADR-0041, 12.10.2026). Aufrufer tragen eine
+# Mail mit einreihen() in die Tabelle mail_ausgang ein, die Versandschleife
+# (in app.py beim Start angestoßen) liefert sie per SMTP mit Anmeldung beim
+# Mailanbieter der Domain ein. anode verschickt nie direkt an Empfänger -
+# Mails von einer Heimanschluss-Adresse landen im Spam.
+#
+# Bewusst nur Standardbibliothek (smtplib), keine neue Abhängigkeit. Der
+# Dienst läuft als einzelner uvicorn-Prozess (siehe auth.py), es gibt also
+# genau eine Versandschleife; FOR UPDATE SKIP LOCKED schützt trotzdem
+# gegen einen parallel laufenden scripts/testmail.py.
+
+log = logging.getLogger("kundenplattform.mail")
+if not log.handlers:
+    # Die Anwendung konfiguriert sonst kein Logging; ohne eigenen Handler
+    # gingen INFO-Meldungen (Trockenlauf!) verloren. stderr landet im Journal.
+    log.addHandler(logging.StreamHandler(sys.stderr))
+    log.setLevel(logging.INFO)
+
+# Wartezeiten vor dem 2., 3. und 4. Versuch. Danach gilt die Mail als
+# fehlgeschlagen. Zusammen gut 20 Minuten - passt in die Stundenfrist der
+# Personalisierungs-Bestätigung.
+WIEDERHOLUNG_NACH_SEKUNDEN = (60, 300, 900)
+
+SCHLEIFEN_INTERVALL_SEKUNDEN = 10
+FEHLER_PAUSE_SEKUNDEN = 300
+SMTP_TIMEOUT_SEKUNDEN = 30
+
+# Die Tabelle enthält Empfängeradressen (personenbezogen) - alte Einträge
+# werden gelöscht.
+AUFBEWAHRUNG_TAGE = 90
+AUFRAEUMEN_INTERVALL_SEKUNDEN = 60 * 60
+
+
+@dataclass
+class Konfiguration:
+    modus: str
+    host: str
+    port: int
+    benutzer: str
+    absender: str
+    antwort_an: str
+    passwort_datei: str
+
+
+def _konfiguration() -> Konfiguration:
+    """
+    Bei jedem Aufruf frisch aus der Umgebung gelesen (nicht beim Import wie
+    in db.py/encryption.py), damit Tests und scripts/testmail.py den Modus
+    umschalten können.
+
+    KUNDENPLATTFORM_MAIL_MODUS: "trockenlauf" (Default - schreibt die Mail
+    ins Protokoll statt sie zu verschicken) oder "smtp".
+    Das Passwort steht nie in der Umgebung, sondern in einer Datei:
+    systemd-Credential "mail_passwort" (siehe kundenplattform.service) oder
+    ausdrücklich KUNDENPLATTFORM_MAIL_PASSWORT_DATEI.
+    """
+
+    passwort_datei = os.environ.get("KUNDENPLATTFORM_MAIL_PASSWORT_DATEI", "")
+    if not passwort_datei and os.environ.get("CREDENTIALS_DIRECTORY"):
+        passwort_datei = os.path.join(os.environ["CREDENTIALS_DIRECTORY"], "mail_passwort")
+
+    return Konfiguration(
+        modus=os.environ.get("KUNDENPLATTFORM_MAIL_MODUS", "trockenlauf"),
+        host=os.environ.get("KUNDENPLATTFORM_MAIL_HOST", ""),
+        port=int(os.environ.get("KUNDENPLATTFORM_MAIL_PORT", "587")),
+        benutzer=os.environ.get("KUNDENPLATTFORM_MAIL_BENUTZER", ""),
+        absender=os.environ.get("KUNDENPLATTFORM_MAIL_ABSENDER", ""),
+        antwort_an=os.environ.get("KUNDENPLATTFORM_MAIL_ANTWORT_AN", ""),
+        passwort_datei=passwort_datei,
+    )
+
+
+def _als_text(wert):
+    # psycopg liefert TEXT-Spalten hier teils als bytes (siehe auth.py).
+    return wert.decode("utf-8") if isinstance(wert, bytes) else wert
+
+
+def einreihen(empfaenger: str, betreff: str, text: str, zweck: str, bezug: str | None = None, conn=None) -> str:
+    """
+    Trägt eine Mail zum Versand ein und gibt ihre ID zurück (für
+    mail_status()). Mit conn läuft der Eintrag in der Transaktion des
+    Aufrufers - die Mail geht dann nur raus, wenn auch dessen übrige
+    Änderungen festgeschrieben werden.
+
+    Wirft ValueError bei ungültiger Adresse oder Zeilenumbruch im Betreff.
+    """
+
+    try:
+        adresse = validate_email(empfaenger.strip(), check_deliverability=False).normalized
+    except EmailNotValidError as fehler:
+        raise ValueError(f"Ungültige E-Mail-Adresse: {fehler}") from fehler
+
+    if "\n" in betreff or "\r" in betreff:
+        raise ValueError("Der Betreff darf keinen Zeilenumbruch enthalten.")
+
+    mail_id = uuid.uuid4()
+    sql = "INSERT INTO mail_ausgang (id, empfaenger, betreff, text, zweck, bezug) VALUES (%s, %s, %s, %s, %s, %s)"
+    werte = (mail_id, adresse, betreff, text, zweck, bezug)
+
+    if conn is not None:
+        with conn.cursor() as cur:
+            cur.execute(sql, werte)
+    else:
+        with get_database_connection() as eigene_verbindung:
+            with eigene_verbindung.cursor() as cur:
+                cur.execute(sql, werte)
+
+    return str(mail_id)
+
+
+def mail_status(mail_id: str) -> dict | None:
+    with get_database_connection() as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                SELECT status, versuche, letzter_fehler, created_at, gesendet_at
+                FROM mail_ausgang WHERE id = %s
+                """,
+                (mail_id,),
+            )
+            row = cur.fetchone()
+
+    if row is None:
+        return None
+
+    status, versuche, letzter_fehler, created_at, gesendet_at = row
+
+    return {
+        "status": _als_text(status),
+        "versuche": versuche,
+        "letzter_fehler": _als_text(letzter_fehler),
+        "created_at": created_at,
+        "gesendet_at": gesendet_at,
+    }
+
+
+def _nachricht(konfiguration: Konfiguration, empfaenger: str, betreff: str, text: str) -> EmailMessage:
+    nachricht = EmailMessage()
+    nachricht["From"] = konfiguration.absender
+    nachricht["To"] = empfaenger
+    nachricht["Subject"] = betreff
+    nachricht["Date"] = formatdate(localtime=True)
+    nachricht["Message-ID"] = make_msgid(domain=parseaddr(konfiguration.absender)[1].rpartition("@")[2] or None)
+    # Kennzeichnet die Mail als automatisch erzeugt, damit
+    # Abwesenheitsnotizen nicht darauf antworten.
+    nachricht["Auto-Submitted"] = "auto-generated"
+    if konfiguration.antwort_an:
+        nachricht["Reply-To"] = konfiguration.antwort_an
+    nachricht.set_content(text)
+
+    return nachricht
+
+
+def _per_smtp_senden(konfiguration: Konfiguration, nachricht: EmailMessage) -> None:
+    fehlend = [
+        name for name, wert in (
+            ("KUNDENPLATTFORM_MAIL_HOST", konfiguration.host),
+            ("KUNDENPLATTFORM_MAIL_BENUTZER", konfiguration.benutzer),
+            ("KUNDENPLATTFORM_MAIL_ABSENDER", konfiguration.absender),
+            ("Passwortdatei (Credential mail_passwort)", konfiguration.passwort_datei),
+        ) if not wert
+    ]
+    if fehlend:
+        raise RuntimeError("Mailversand nicht konfiguriert, es fehlt: " + ", ".join(fehlend))
+
+    passwort = Path(konfiguration.passwort_datei).read_text(encoding="utf-8").strip()
+
+    with smtplib.SMTP(konfiguration.host, konfiguration.port, timeout=SMTP_TIMEOUT_SEKUNDEN) as smtp:
+        smtp.starttls(context=ssl.create_default_context())
+        smtp.login(konfiguration.benutzer, passwort)
+        smtp.send_message(nachricht)
+
+
+def ausgang_abarbeiten(limit: int = 20) -> int:
+    """
+    Verschickt die fälligen Mails (höchstens limit) und gibt zurück, wie
+    viele bearbeitet wurden. Jede Mail läuft in einer eigenen Transaktion,
+    die Zeile bleibt während des Versands gesperrt.
+    """
+
+    konfiguration = _konfiguration()
+    bearbeitet = 0
+
+    with get_database_connection() as conn:
+        while bearbeitet < limit:
+            with conn.cursor() as cur:
+                cur.execute(
+                    """
+                    SELECT id, empfaenger, betreff, text, zweck, versuche
+                    FROM mail_ausgang
+                    WHERE status = 'wartet' AND naechster_versuch_at <= CURRENT_TIMESTAMP
+                    ORDER BY created_at
+                    LIMIT 1
+                    FOR UPDATE SKIP LOCKED
+                    """
+                )
+                row = cur.fetchone()
+                if row is None:
+                    break
+
+                mail_id, empfaenger, betreff, text, zweck, versuche = row
+                empfaenger, betreff, text, zweck = (_als_text(w) for w in (empfaenger, betreff, text, zweck))
+                versuche += 1
+
+                status, fehler, wartezeit = _zustellen(konfiguration, empfaenger, betreff, text, zweck, versuche)
+
+                if status == "wartet":
+                    cur.execute(
+                        """
+                        UPDATE mail_ausgang
+                        SET versuche = %s, letzter_fehler = %s,
+                            naechster_versuch_at = CURRENT_TIMESTAMP + make_interval(secs => %s)
+                        WHERE id = %s
+                        """,
+                        (versuche, fehler, wartezeit, mail_id),
+                    )
+                else:
+                    cur.execute(
+                        """
+                        UPDATE mail_ausgang
+                        SET status = %s, versuche = %s, letzter_fehler = %s, text = NULL,
+                            gesendet_at = CASE WHEN %s = 'gesendet' THEN CURRENT_TIMESTAMP END
+                        WHERE id = %s
+                        """,
+                        (status, versuche, fehler, status, mail_id),
+                    )
+
+            conn.commit()
+            bearbeitet += 1
+
+    return bearbeitet
+
+
+def _zustellen(konfiguration, empfaenger, betreff, text, zweck, versuche) -> tuple[str, str | None, int]:
+    """Ein Zustellversuch. Gibt (neuer Status, Fehlertext, Wartezeit bis zum nächsten Versuch) zurück."""
+
+    if konfiguration.modus != "smtp":
+        log.info("Mail (Trockenlauf) an %s\nBetreff: %s\n%s\n---", empfaenger, betreff, text)
+        return "trockenlauf", None, 0
+
+    try:
+        _per_smtp_senden(konfiguration, _nachricht(konfiguration, empfaenger, betreff, text))
+    except smtplib.SMTPRecipientsRefused as fehler:
+        # Der Mailserver lehnt die Adresse ab - ein weiterer Versuch ändert
+        # daran nichts.
+        log.warning("Mail (%s) an %s abgelehnt: %s", zweck, empfaenger, fehler)
+        return "fehlgeschlagen", f"Empfänger abgelehnt: {fehler}", 0
+    except Exception as fehler:  # noqa: BLE001 - jeder Fehler führt zur Wiederholung
+        if versuche > len(WIEDERHOLUNG_NACH_SEKUNDEN):
+            log.error("Mail (%s) an %s endgültig fehlgeschlagen nach %s Versuchen: %s", zweck, empfaenger, versuche, fehler)
+            return "fehlgeschlagen", str(fehler), 0
+
+        wartezeit = WIEDERHOLUNG_NACH_SEKUNDEN[versuche - 1]
+        log.warning("Mail (%s) an %s: Versuch %s fehlgeschlagen (%s), nächster in %s s", zweck, empfaenger, versuche, fehler, wartezeit)
+        return "wartet", str(fehler), wartezeit
+
+    log.info("Mail (%s) an %s verschickt.", zweck, empfaenger)
+    return "gesendet", None, 0
+
+
+def aufraeumen() -> int:
+    with get_database_connection() as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                "DELETE FROM mail_ausgang WHERE created_at < CURRENT_TIMESTAMP - make_interval(days => %s)",
+                (AUFBEWAHRUNG_TAGE,),
+            )
+            return cur.rowcount
+
+
+async def versand_schleife() -> None:
+    """Läuft für die Lebensdauer des Dienstes (siehe lifespan in app.py)."""
+
+    zuletzt_aufgeraeumt = 0.0
+
+    while True:
+        try:
+            await asyncio.to_thread(ausgang_abarbeiten)
+
+            if time.monotonic() - zuletzt_aufgeraeumt > AUFRAEUMEN_INTERVALL_SEKUNDEN:
+                await asyncio.to_thread(aufraeumen)
+                zuletzt_aufgeraeumt = time.monotonic()
+        except Exception:  # noqa: BLE001 - die Schleife darf nie enden
+            # Typisch: Datenbank nicht erreichbar oder Migration 0009 fehlt.
+            # Lange Pause, damit das Journal nicht vollläuft.
+            log.exception("Versandschleife: Durchlauf fehlgeschlagen, nächster in %s s.", FEHLER_PAUSE_SEKUNDEN)
+            await asyncio.sleep(FEHLER_PAUSE_SEKUNDEN)
+            continue
+
+        await asyncio.sleep(SCHLEIFEN_INTERVALL_SEKUNDEN)

migrations/0009_mail_ausgang.sql

@@ -0,0 +1,30 @@
+BEGIN;
+
+-- Mailversand der Plattform (ADR-0041, 12.10.2026): Ausgangstabelle für
+-- Bestätigungs- und Reset-Mails. Jede Mail wird erst eingetragen und dann
+-- von der Versandschleife (mail.py) verschickt, mit Wiederholung bei
+-- Fehlern - der Aufrufer kann am Status ablesen, ob die Mail raus ist.
+-- status: wartet | gesendet | fehlgeschlagen | trockenlauf.
+-- text enthält Links mit Einmal-Token und wird deshalb geleert, sobald die
+-- Mail ihren Endzustand erreicht hat; übrig bleiben nur die Metadaten.
+-- bezug ist ein freier Verweis des Aufrufers (z. B. Geräte-ID), ohne
+-- Foreign Key, weil Geräte in der provisioning-server-Datenbank liegen.
+CREATE TABLE mail_ausgang (
+    id UUID PRIMARY KEY,
+    empfaenger TEXT NOT NULL,
+    betreff TEXT NOT NULL,
+    text TEXT,
+    zweck TEXT NOT NULL,
+    bezug TEXT,
+    status TEXT NOT NULL DEFAULT 'wartet'
+        CHECK (status IN ('wartet', 'gesendet', 'fehlgeschlagen', 'trockenlauf')),
+    versuche INTEGER NOT NULL DEFAULT 0,
+    letzter_fehler TEXT,
+    naechster_versuch_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    gesendet_at TIMESTAMPTZ,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX mail_ausgang_wartet ON mail_ausgang (naechster_versuch_at) WHERE status = 'wartet';
+
+COMMIT;

scripts/testmail.py

@@ -0,0 +1,55 @@
+"""
+Testmail über die Ausgangstabelle verschicken - prüft Konfiguration,
+Zugangsdaten und Zustellbarkeit (SPF/DKIM/DMARC in den Kopfzeilen der
+empfangenen Mail ansehen). Beispiel:
+
+    KUNDENPLATTFORM_DATABASE_URL=... KUNDENPLATTFORM_MAIL_MODUS=smtp ... \\
+        .venv/bin/python scripts/testmail.py --an jemand@example.org
+
+Ohne KUNDENPLATTFORM_MAIL_MODUS=smtp läuft das als Trockenlauf.
+"""
+
+import argparse
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import mail  # noqa: E402
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--an", required=True, help="Empfängeradresse")
+    args = parser.parse_args()
+
+    if not os.environ.get("KUNDENPLATTFORM_DATABASE_URL"):
+        sys.exit("KUNDENPLATTFORM_DATABASE_URL ist nicht gesetzt.")
+
+    try:
+        mail_id = mail.einreihen(
+            args.an,
+            "[Tuxflotte] Testmail",
+            "Dies ist eine Testmail der Tuxflotte-Kundenplattform.\n"
+            "Wenn sie ankommt, funktioniert der Mailversand.\n",
+            zweck="testmail",
+        )
+    except ValueError as fehler:
+        sys.exit(str(fehler))
+
+    # Läuft der Dienst, kann seine Versandschleife die Mail zuerst greifen -
+    # der Status unten stimmt in beiden Fällen.
+    mail.ausgang_abarbeiten()
+    status = mail.mail_status(mail_id)
+
+    print(f"Status: {status['status']} (Versuche: {status['versuche']})")
+    if status["letzter_fehler"]:
+        print(f"Fehler: {status['letzter_fehler']}")
+    if status["status"] == "wartet":
+        print("Die Mail bleibt in der Ausgangstabelle und wird von der Versandschleife erneut versucht.")
+
+    sys.exit(0 if status["status"] in ("gesendet", "trockenlauf") else 1)
+
+
+if __name__ == "__main__":
+    main()

tests/test_mail.py

@@ -0,0 +1,199 @@
+"""
+Tests für mail.py gegen eine echte PostgreSQL-Wegwerfdatenbank (Migration
+0009 muss eingespielt sein); SMTP wird durch einen Mock ersetzt.
+
+    KUNDENPLATTFORM_DATABASE_URL=postgresql://.../kundenplattform_test \\
+        .venv/bin/python -m unittest discover -s tests -v
+
+Die Tests leeren mail_ausgang - sie laufen deshalb nur, wenn der
+Datenbankname "test" enthält.
+"""
+
+import os
+import smtplib
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from psycopg.conninfo import conninfo_to_dict  # noqa: E402
+
+DATABASE_URL = os.environ.get("KUNDENPLATTFORM_DATABASE_URL", "")
+TESTDATENBANK = "test" in (conninfo_to_dict(DATABASE_URL).get("dbname") or "") if DATABASE_URL else False
+
+if TESTDATENBANK:
+    import mail
+    from db import get_database_connection
+
+
+def _zeile(mail_id):
+    with get_database_connection() as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                "SELECT status, versuche, text, letzter_fehler, gesendet_at, "
+                "naechster_versuch_at > CURRENT_TIMESTAMP FROM mail_ausgang WHERE id = %s",
+                (mail_id,),
+            )
+            return cur.fetchone()
+
+
+def _sofort_faellig():
+    with get_database_connection() as conn:
+        with conn.cursor() as cur:
+            cur.execute("UPDATE mail_ausgang SET naechster_versuch_at = CURRENT_TIMESTAMP WHERE status = 'wartet'")
+
+
+@unittest.skipUnless(TESTDATENBANK, "KUNDENPLATTFORM_DATABASE_URL zeigt nicht auf eine Testdatenbank.")
+class MailTest(unittest.TestCase):
+    def setUp(self):
+        with get_database_connection() as conn:
+            with conn.cursor() as cur:
+                cur.execute("TRUNCATE mail_ausgang")
+
+        self.passwort_datei = tempfile.NamedTemporaryFile("w", suffix=".passwort")
+        self.passwort_datei.write("geheim-123\n")
+        self.passwort_datei.flush()
+        self.addCleanup(self.passwort_datei.close)
+
+        self.smtp_umgebung = {
+            "KUNDENPLATTFORM_MAIL_MODUS": "smtp",
+            "KUNDENPLATTFORM_MAIL_HOST": "smtp.example.org",
+            "KUNDENPLATTFORM_MAIL_BENUTZER": "konto@example.org",
+            "KUNDENPLATTFORM_MAIL_ABSENDER": "Tuxflotte <konto@example.org>",
+            "KUNDENPLATTFORM_MAIL_ANTWORT_AN": "hilfe@example.org",
+            "KUNDENPLATTFORM_MAIL_PASSWORT_DATEI": self.passwort_datei.name,
+        }
+
+    def _einreihen(self, **kwargs):
+        return mail.einreihen("Jemand@Example.ORG", "Betreff ä", "Link: https://example.org/t/abc\n", zweck="test", **kwargs)
+
+    def test_ungueltige_eingaben_werden_abgelehnt(self):
+        with self.assertRaises(ValueError):
+            mail.einreihen("keine-adresse", "Betreff", "Text", zweck="test")
+        with self.assertRaises(ValueError):
+            mail.einreihen("a@example.org", "Betreff\nBcc: b@example.org", "Text", zweck="test")
+
+    def test_trockenlauf_verschickt_nichts(self):
+        mail_id = self._einreihen()
+
+        with mock.patch.dict(os.environ, {"KUNDENPLATTFORM_MAIL_MODUS": "trockenlauf"}), \
+                mock.patch("smtplib.SMTP") as smtp, self.assertLogs("kundenplattform.mail", "INFO") as protokoll:
+            self.assertEqual(mail.ausgang_abarbeiten(), 1)
+
+        smtp.assert_not_called()
+        self.assertIn("https://example.org/t/abc", protokoll.output[0])
+        status, versuche, text, fehler, gesendet_at, _ = _zeile(mail_id)
+        self.assertEqual((status, versuche, text, fehler, gesendet_at), ("trockenlauf", 1, None, None, None))
+
+    def test_smtp_versand(self):
+        mail_id = self._einreihen(bezug="geraet-1")
+
+        with mock.patch.dict(os.environ, self.smtp_umgebung), mock.patch("smtplib.SMTP") as smtp:
+            self.assertEqual(mail.ausgang_abarbeiten(), 1)
+            # Nichts mehr fällig.
+            self.assertEqual(mail.ausgang_abarbeiten(), 0)
+
+        smtp.assert_called_once_with("smtp.example.org", 587, timeout=mail.SMTP_TIMEOUT_SEKUNDEN)
+        verbindung = smtp.return_value.__enter__.return_value
+        verbindung.starttls.assert_called_once()
+        verbindung.login.assert_called_once_with("konto@example.org", "geheim-123")
+        nachricht = verbindung.send_message.call_args.args[0]
+        self.assertEqual(nachricht["To"], "Jemand@example.org")
+        self.assertEqual(nachricht["From"], "Tuxflotte <konto@example.org>")
+        self.assertEqual(nachricht["Subject"], "Betreff ä")
+        self.assertEqual(nachricht["Reply-To"], "hilfe@example.org")
+        self.assertTrue(nachricht["Message-ID"].endswith("@example.org>"))
+        self.assertIn("https://example.org/t/abc", nachricht.get_content())
+
+        status, versuche, text, fehler, gesendet_at, _ = _zeile(mail_id)
+        self.assertEqual((status, versuche, text, fehler), ("gesendet", 1, None, None))
+        self.assertIsNotNone(gesendet_at)
+        self.assertEqual(mail.mail_status(mail_id)["status"], "gesendet")
+
+    def test_fehler_wird_wiederholt_und_gibt_dann_auf(self):
+        mail_id = self._einreihen()
+
+        with mock.patch.dict(os.environ, self.smtp_umgebung), \
+                mock.patch("smtplib.SMTP", side_effect=OSError("Netz weg")), self.assertLogs("kundenplattform.mail"):
+            self.assertEqual(mail.ausgang_abarbeiten(), 1)
+            status, versuche, text, fehler, _, in_zukunft = _zeile(mail_id)
+            self.assertEqual((status, versuche, fehler), ("wartet", 1, "Netz weg"))
+            self.assertIsNotNone(text)
+            self.assertTrue(in_zukunft)
+            # Noch nicht wieder fällig.
+            self.assertEqual(mail.ausgang_abarbeiten(), 0)
+
+            for _ in mail.WIEDERHOLUNG_NACH_SEKUNDEN:
+                _sofort_faellig()
+                self.assertEqual(mail.ausgang_abarbeiten(), 1)
+
+        status, versuche, text, fehler, gesendet_at, _ = _zeile(mail_id)
+        self.assertEqual((status, versuche, text, gesendet_at), ("fehlgeschlagen", 4, None, None))
+        self.assertEqual(fehler, "Netz weg")
+
+    def test_wiederholung_gelingt(self):
+        mail_id = self._einreihen()
+
+        with mock.patch.dict(os.environ, self.smtp_umgebung):
+            with mock.patch("smtplib.SMTP", side_effect=OSError("Netz weg")), self.assertLogs("kundenplattform.mail"):
+                mail.ausgang_abarbeiten()
+            _sofort_faellig()
+            with mock.patch("smtplib.SMTP"):
+                mail.ausgang_abarbeiten()
+
+        self.assertEqual(_zeile(mail_id)[:2], ("gesendet", 2))
+
+    def test_abgelehnter_empfaenger_wird_nicht_wiederholt(self):
+        mail_id = self._einreihen()
+
+        with mock.patch.dict(os.environ, self.smtp_umgebung), mock.patch("smtplib.SMTP") as smtp, \
+                self.assertLogs("kundenplattform.mail"):
+            smtp.return_value.__enter__.return_value.send_message.side_effect = smtplib.SMTPRecipientsRefused(
+                {"jemand@example.org": (550, b"unbekannt")}
+            )
+            mail.ausgang_abarbeiten()
+
+        status, versuche, text, fehler, _, _ = _zeile(mail_id)
+        self.assertEqual((status, versuche, text), ("fehlgeschlagen", 1, None))
+        self.assertIn("Empfänger abgelehnt", fehler)
+
+    def test_fehlende_konfiguration_ist_ein_klarer_fehler(self):
+        mail_id = self._einreihen()
+        umgebung = dict(self.smtp_umgebung, KUNDENPLATTFORM_MAIL_HOST="")
+
+        with mock.patch.dict(os.environ, umgebung), mock.patch("smtplib.SMTP") as smtp, \
+                self.assertLogs("kundenplattform.mail"):
+            mail.ausgang_abarbeiten()
+
+        smtp.assert_not_called()
+        status, _, _, fehler, _, _ = _zeile(mail_id)
+        self.assertEqual(status, "wartet")
+        self.assertIn("KUNDENPLATTFORM_MAIL_HOST", fehler)
+
+    def test_einreihen_in_fremder_transaktion(self):
+        with get_database_connection() as conn:
+            mail_id = self._einreihen(conn=conn)
+            conn.rollback()
+
+        self.assertIsNone(mail.mail_status(mail_id))
+
+    def test_aufraeumen_loescht_nur_alte_eintraege(self):
+        alt = self._einreihen()
+        neu = self._einreihen()
+
+        with get_database_connection() as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    "UPDATE mail_ausgang SET created_at = CURRENT_TIMESTAMP - make_interval(days => %s) WHERE id = %s",
+                    (mail.AUFBEWAHRUNG_TAGE + 1, alt),
+                )
+
+        self.assertEqual(mail.aufraeumen(), 1)
+        self.assertIsNone(mail.mail_status(alt))
+        self.assertIsNotNone(mail.mail_status(neu))
+
+
+if __name__ == "__main__":
+    unittest.main()