Files
certain/backend/tests/test_mail_import.py
T
GuidoandClaude Opus 5 c47525ff7e feat: Vorschlaege buchen und Telefon-Dateien nur ablegen
Zwei Luecken geschlossen:

Mail-Vorschlaege standen als "pending" im Eingang, aber es gab keinen Knopf
zum Buchen – der Weg war hinten offen. Jetzt je Eintrag "Buchen" und ein
"Alle buchen". Unklare bleiben liegen: Bei ihnen fehlt etwas oder die
Erkennung war unsicher, die soll man ansehen statt im Stapel durchzuwinken.

Der Auszug vom Telefon bucht nicht mehr sofort, sondern wird nur abgelegt.
Am Telefon prueft man keine 500 Zeilen; die Datei wartet in der
Import-Ansicht und laeuft dort durch die normale Vorschau mit
Saldo-Abgleich, Zeilenfilter und Duplikatpruefung.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-22 21:12:58 +02:00

447 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Postfächer, Verschlüsselung und der Abruf gegen einen IMAP-Doppelgänger.
Ein echter Mailserver wäre im Test weder verfügbar noch wünschenswert. Der
Doppelgänger unten spricht nur die Handvoll IMAP-Befehle, die ``mail_import``
benutzt, und liefert die Muster-Mails aus ``fixtures/mail`` aus.
"""
import pathlib
import pytest
from app.services import geheimnis, mail_import
MUSTER = pathlib.Path(__file__).parent / "fixtures" / "mail"
class FakeImap:
"""Minimaler IMAP-Server: nur was ``mail_import`` tatsächlich aufruft."""
def __init__(self, mails: list[bytes]):
self.mails = mails
self.gesehen: list[bytes] = []
self.kopiert: list[tuple[bytes, str]] = []
self.abgemeldet = False
def login(self, user, passwort):
return "OK", [b""]
def list(self):
return "OK", [b'(\\HasNoChildren) "/" "INBOX"', b'(\\HasNoChildren) "/" "Belege"']
def select(self, ordner):
return ("OK", [b"1"]) if "INBOX" in ordner or "Belege" in ordner else ("NO", [b""])
def search(self, charset, kriterium):
return "OK", [b" ".join(str(i + 1).encode() for i in range(len(self.mails)))]
def fetch(self, uid, teile):
return "OK", [(b"1 (RFC822 {1})", self.mails[int(uid) - 1])]
def store(self, uid, befehl, flags):
if "Seen" in flags:
self.gesehen.append(uid)
return "OK", [b""]
def copy(self, uid, ordner):
self.kopiert.append((uid, ordner))
return "OK", [b""]
def logout(self):
self.abgemeldet = True
return "BYE", [b""]
@pytest.fixture()
def fake_imap(monkeypatch):
"""Ersetzt die Verbindung; gibt eine Funktion zum Bestücken zurück."""
gehalten = {}
def bestuecken(*dateinamen):
mails = [(MUSTER / n).read_bytes() for n in dateinamen]
server = FakeImap(mails)
gehalten["server"] = server
monkeypatch.setattr(mail_import, "_verbinden", lambda quelle: server)
return server
return bestuecken
def _konto(admin, **kw):
body = {"name": "Raiffeisen", "type": "girokonto", "currency": "CHF",
"opening_balance_cents": 0}
body.update(kw)
return admin.post("/api/accounts", json=body).json()
def _postfach(admin, **kw):
body = {"name": "Belege", "host": "imap.example.com", "port": 993,
"username": "kundin@example.com", "password": "geheim123", "folder": "INBOX"}
body.update(kw)
r = admin.post("/api/mail/sources", json=body)
assert r.status_code == 201, r.text
return r.json()
def test_passwort_wird_verschluesselt_abgelegt_und_nie_ausgeliefert(admin):
from app.database import SessionLocal
from app.models import MailSource
quelle = _postfach(admin)
assert "password" not in quelle
assert quelle["password_set"] is True
db = SessionLocal()
try:
zeile = db.get(MailSource, quelle["id"])
assert zeile.password_enc != "geheim123", "Klartext in der Datenbank"
assert geheimnis.entschluesseln(zeile.password_enc) == "geheim123"
finally:
db.close()
def test_unlesbares_passwort_meldet_sich_verstaendlich(admin, monkeypatch):
"""Getauschtes Server-Geheimnis darf nicht in einem Stacktrace enden."""
from app.database import SessionLocal
from app.models import MailSource
quelle = _postfach(admin)
db = SessionLocal()
try:
zeile = db.get(MailSource, quelle["id"])
zeile.password_enc = "kaputt"
db.commit()
with pytest.raises(mail_import.MailFehler) as fehler:
mail_import._verbinden(zeile)
assert "entschlüsseln" in str(fehler.value)
finally:
db.close()
def test_passwort_bleibt_wenn_es_beim_aendern_weggelassen_wird(admin):
from app.database import SessionLocal
from app.models import MailSource
quelle = _postfach(admin)
r = admin.put(f"/api/mail/sources/{quelle['id']}", json={
"host": "imap.example.com", "username": "kundin@example.com", "folder": "Belege"})
assert r.status_code == 200, r.text
assert r.json()["folder"] == "Belege"
db = SessionLocal()
try:
assert geheimnis.entschluesseln(db.get(MailSource, quelle["id"]).password_enc) == "geheim123"
finally:
db.close()
def test_abruf_legt_erkannte_zahlung_in_den_eingang(admin, fake_imap):
konto = _konto(admin)
quelle = _postfach(admin)
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
"name": "Raiffeisen", "from_contains": "raiffeisen.ch", "account_id": konto["id"]})
server = fake_imap("raiffeisen_zahlungsstatus.eml")
r = admin.post(f"/api/mail/sources/{quelle['id']}/fetch")
assert r.status_code == 200, r.text
assert r.json()["neu"] == 1
treffer = [e for e in admin.get("/api/inbox").json()
if e["payee"] == "Sunrise GmbH / Yallo"]
assert len(treffer) == 1
assert treffer[0]["amount_cents"] == -1490
assert treffer[0]["currency"] == "CHF"
assert treffer[0]["date"] == "2026-07-31"
assert treffer[0]["account_id"] == konto["id"]
# Nichts wird stillschweigend gebucht.
assert treffer[0]["status"] == "pending"
assert treffer[0]["transaction_id"] is None
assert server.gesehen, "Mail muss als gelesen markiert werden"
def test_dieselbe_mail_zweimal_ergibt_einen_eintrag(admin, fake_imap):
"""Die Message-ID zählt, nicht die UID – die vergibt der Server neu."""
konto = _konto(admin)
quelle = _postfach(admin)
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
"from_contains": "raiffeisen.ch", "account_id": konto["id"]})
fake_imap("raiffeisen_zahlungsstatus.eml")
admin.post(f"/api/mail/sources/{quelle['id']}/fetch")
vorher = len(admin.get("/api/inbox").json())
zweiter = admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()
assert zweiter["neu"] == 0
assert zweiter["uebersprungen"] == 1
assert len(admin.get("/api/inbox").json()) == vorher
def test_mail_ohne_passende_zuordnung_wird_uebergangen(admin, fake_imap):
quelle = _postfach(admin)
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
"from_contains": "sparkasse.de"})
fake_imap("raiffeisen_zahlungsstatus.eml")
ergebnis = admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()
assert ergebnis["neu"] == 0
assert ergebnis["uebersprungen"] == 1
def test_zuordnung_ohne_konto_landet_als_unklar(admin, fake_imap):
"""Ohne Zielkonto lässt sich nicht buchen – das muss sichtbar sein."""
quelle = _postfach(admin)
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
"from_contains": "revolut.com"})
fake_imap("revolut_ueberweisung.eml")
assert admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()["unklar"] == 1
def test_verbindungstest_liefert_die_ordner(admin, fake_imap):
quelle = _postfach(admin)
fake_imap()
antwort = admin.post(f"/api/mail/sources/{quelle['id']}/test")
assert antwort.status_code == 200, antwort.text
assert "Belege" in antwort.json()["folders"]
def test_verschieben_statt_nur_markieren(admin, fake_imap):
konto = _konto(admin)
quelle = _postfach(admin, done_folder="Belege")
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
"from_contains": "raiffeisen.ch", "account_id": konto["id"]})
server = fake_imap("raiffeisen_zahlungsstatus.eml")
admin.post(f"/api/mail/sources/{quelle['id']}/fetch")
assert server.kopiert and "Belege" in server.kopiert[0][1]
def test_weitergeleitete_mail_wird_am_originalabsender_erkannt(admin, fake_imap, monkeypatch):
"""Leitet man Bank-Mails in ein Sammelpostfach weiter, steht die eigene
Adresse im ``From``. Der Originalabsender bleibt aber in den
Weiterleitungs-Koepfen erhalten – daran muss die Zuordnung greifen."""
konto = _konto(admin)
quelle = _postfach(admin)
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
"from_contains": "raiffeisen.ch", "account_id": konto["id"]})
roh = (MUSTER / "raiffeisen_zahlungsstatus.eml").read_bytes()
weitergeleitet = roh.replace(
b"From: noreply@ebanking.raiffeisen.ch",
b"From: ich@gmail.com\nX-Forwarded-For: noreply@ebanking.raiffeisen.ch ich@gmail.com",
).replace(b"<raiffeisen-muster-1@", b"<weitergeleitet-1@")
server = FakeImap([weitergeleitet])
monkeypatch.setattr(mail_import, "_verbinden", lambda q: server)
assert admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()["neu"] == 1
def test_uebergangene_mails_nennen_absender_und_betreff(admin, monkeypatch):
"""Beim Einrichten muss man sehen, was in den Filter gehoert."""
quelle = _postfach(admin)
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={"from_contains": "passt.nie"})
# Eigene Message-ID: die Testdatenbank lebt ueber den ganzen Lauf, sonst
# gilt die Mail als bereits erfasst und die Zuordnung wird nie geprueft.
roh = (MUSTER / "raiffeisen_zahlungsstatus.eml").read_bytes().replace(
b"<raiffeisen-muster-1@", b"<ohne-zuordnung-1@")
monkeypatch.setattr(mail_import, "_verbinden", lambda q: FakeImap([roh]))
ergebnis = admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()
assert ergebnis["uebersprungen"] == 1
assert "raiffeisen.ch" in ergebnis["ohne_zuordnung"][0]["from"]
assert "Zahlungsstatus" in ergebnis["ohne_zuordnung"][0]["subject"]
def test_bereits_gelesene_mail_wird_trotzdem_geholt(admin, monkeypatch):
"""Wer die Mail vorher im Mailprogramm ansieht, darf sie nicht verlieren.
Frueher wurde auf UNSEEN gefiltert – eine einmal geoeffnete Mail war damit
fuer Certain unsichtbar, und das ohne jeden Hinweis.
"""
konto = _konto(admin)
quelle = _postfach(admin)
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
"from_contains": "raiffeisen.ch", "account_id": konto["id"]})
class NurGelesene(FakeImap):
def search(self, charset, kriterium):
# Ein Server, der auf UNSEEN nichts liefert – wie bei einer Mail,
# die man schon angesehen hat.
assert "UNSEEN" not in kriterium, f"Es darf nicht auf UNSEEN gefiltert werden: {kriterium}"
return "OK", [b"1"]
roh = (MUSTER / "raiffeisen_zahlungsstatus.eml").read_bytes().replace(
b"<raiffeisen-muster-1@", b"<schon-gelesen-1@")
monkeypatch.setattr(mail_import, "_verbinden", lambda q: NurGelesene([roh]))
ergebnis = admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()
assert ergebnis["gefunden"] == 1
assert ergebnis["neu"] == 1
def _weitergeleitet(quelldatei: str, message_id: bytes) -> bytes:
"""Baut aus einer Bank-Mail eine von Hand weitergeleitete nach.
Dabei steht die EIGENE Adresse im From, der Betreff bekommt ein „WG:“ und
der Originalkopf wandert in den Text – genau wie bei Gmail und Proton.
"""
roh = (MUSTER / quelldatei).read_bytes().decode("utf-8")
kopf, koerper = roh.split("\n\n", 1)
zitat = ("---------- Weitergeleitete Nachricht ----------\n"
"Von: noreply@ebanking.raiffeisen.ch\n"
"Betreff: Raiffeisen E-Banking - Änderung Zahlungsstatus\n\n")
neu = (
"From: Alex Muster <alex@example.com>\n"
"To: sammel@example.com\n"
"Date: Fri, 31 Jul 2026 08:00:00 +0200\n"
f"Message-Id: <{message_id.decode()}@example.invalid>\n"
"Subject: =?utf-8?Q?WG=3A_Raiffeisen_E-Banking?=\n"
"Mime-Version: 1.0\n"
"Content-Type: text/plain; charset=utf-8\n"
"\n" + zitat + koerper.replace("=\n", "").replace("=C3=84", "Ä")
.replace("=C3=BC", "ü").replace("=C3=A4", "ä").replace("=C3=BC", "ü")
.replace("=C3=BCber", "über").replace("Beg=C3=BCnstigter", "Begünstigter")
.replace("Ausf=C3=BChrungsdatum", "Ausführungsdatum")
)
return neu.encode("utf-8")
def test_von_hand_weitergeleitete_mail_wird_am_betreff_erkannt(admin, monkeypatch):
"""Der Nutzer trägt „raiffeisen“ ein – das steht bei einer manuellen
Weiterleitung im Betreff, nicht im Absender. Genau dort muss es greifen."""
konto = _konto(admin)
quelle = _postfach(admin)
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
"from_contains": "raiffeisen", "account_id": konto["id"]})
monkeypatch.setattr(mail_import, "_verbinden",
lambda q: FakeImap([_weitergeleitet("raiffeisen_zahlungsstatus.eml", b"wg-1")]))
ergebnis = admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()
assert ergebnis["neu"] == 1, ergebnis
treffer = [e for e in admin.get("/api/inbox").json() if e["payee"] == "Sunrise GmbH / Yallo"]
assert treffer, "Die Werte müssen auch aus der weitergeleiteten Mail lesbar sein"
assert treffer[0]["amount_cents"] == -1490
def test_zuordnung_laesst_sich_aendern_und_loeschen(admin):
konto = _konto(admin)
zweites = _konto(admin, name="Zweitkonto")
quelle = _postfach(admin)
a = admin.post(f"/api/mail/sources/{quelle['id']}/senders",
json={"from_contains": "alt", "account_id": konto["id"]}).json()
r = admin.put(f"/api/mail/senders/{a['id']}",
json={"from_contains": "neu", "account_id": zweites["id"]})
assert r.status_code == 200, r.text
assert r.json()["from_contains"] == "neu"
assert r.json()["account_id"] == zweites["id"]
assert admin.delete(f"/api/mail/senders/{a['id']}").status_code == 204
uebrig = admin.get(f"/api/mail/sources/{quelle['id']}/senders").json()
assert not [x for x in uebrig if x["id"] == a["id"]]
def test_waehrung_entscheidet_ueber_das_konto(admin, monkeypatch):
"""Revolut schickt für Euro und Franken dieselbe Mail vom selben Absender.
Erst die erkannte Währung sagt, auf welches der beiden Revolut-Konten die
Zahlung gehört – ohne dieses Kriterium landet alles auf demselben.
"""
eur = _konto(admin, name="Revolut Euro", currency="EUR")
chf = _konto(admin, name="Revolut CHF", currency="CHF")
quelle = _postfach(admin)
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
"from_contains": "revolut", "currency": "EUR", "account_id": eur["id"]})
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
"from_contains": "revolut", "currency": "CHF", "account_id": chf["id"]})
def mail(message_id: str, betrag: str, waehrung: str) -> bytes:
return (
f"From: Revolut <no-reply@revolut.com>\n"
f"To: kundin@example.com\n"
f"Date: Sat, 01 Aug 2026 10:41:51 +0000\n"
f"Message-Id: <{message_id}@example.invalid>\n"
f"Subject: Zahlung\n"
f"Content-Type: text/plain; charset=utf-8\n"
f"\nDu hast Geld gesendet.\nGesendet\n{betrag} {waehrung}\nAn\nAlex Muster\n"
).encode("utf-8")
monkeypatch.setattr(mail_import, "_verbinden",
lambda q: FakeImap([mail("waehrung-eur", "324.00", "EUR"),
mail("waehrung-chf", "50.00", "CHF")]))
assert admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()["neu"] == 2
eintraege = admin.get("/api/inbox").json()
in_eur = [e for e in eintraege if e["amount_cents"] == -32400]
in_chf = [e for e in eintraege if e["amount_cents"] == -5000]
assert in_eur and in_eur[0]["account_id"] == eur["id"], "EUR gehört aufs Euro-Konto"
assert in_chf and in_chf[0]["account_id"] == chf["id"], "CHF gehört aufs Franken-Konto"
def test_zuordnung_mit_waehrung_schlaegt_die_allgemeine(admin, monkeypatch):
"""Wer eine genaue Regel neben eine allgemeine stellt, meint die genaue."""
eur = _konto(admin, name="Genau EUR", currency="EUR")
sammel = _konto(admin, name="Sammelkonto", currency="CHF")
quelle = _postfach(admin)
# Die allgemeine zuerst anlegen – sie darf trotzdem nicht gewinnen.
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
"from_contains": "revolut", "account_id": sammel["id"]})
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
"from_contains": "revolut", "currency": "EUR", "account_id": eur["id"]})
roh = ("From: Revolut <no-reply@revolut.com>\nTo: k@example.com\n"
"Date: Sat, 01 Aug 2026 10:41:51 +0000\n"
"Message-Id: <vorrang-eur@example.invalid>\nSubject: Zahlung\n"
"Content-Type: text/plain; charset=utf-8\n"
"\nDu hast Geld gesendet.\nGesendet\n12.00 EUR\nAn\nAlex Muster\n").encode("utf-8")
monkeypatch.setattr(mail_import, "_verbinden", lambda q: FakeImap([roh]))
admin.post(f"/api/mail/sources/{quelle['id']}/fetch")
treffer = [e for e in admin.get("/api/inbox").json() if e["amount_cents"] == -1200]
assert treffer and treffer[0]["account_id"] == eur["id"]
def test_mailvorschlag_wird_erst_auf_knopfdruck_gebucht(admin, fake_imap):
"""Eine Mail bucht nie von selbst – aber sie muss sich buchen lassen."""
konto = _konto(admin, name="Buchbar", opening_balance_cents=0)
quelle = _postfach(admin)
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
"from_contains": "raiffeisen", "account_id": konto["id"]})
roh = (MUSTER / "raiffeisen_zahlungsstatus.eml").read_bytes().replace(
b"<raiffeisen-muster-1@", b"<buchbar-1@")
import unittest.mock as _m
with _m.patch.object(mail_import, "_verbinden", lambda q: FakeImap([roh])):
admin.post(f"/api/mail/sources/{quelle['id']}/fetch")
eintrag = [e for e in admin.get("/api/inbox").json()
if e["account_id"] == konto["id"]][0]
assert eintrag["status"] == "pending"
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 0
r = admin.post(f"/api/inbox/{eintrag['id']}/book")
assert r.status_code == 200, r.text
assert r.json()["status"] == "booked"
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -1490
# Zweimal buchen darf den Saldo nicht verdoppeln.
admin.post(f"/api/inbox/{eintrag['id']}/book")
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -1490
def test_unklare_vorschlaege_werden_nicht_mitgebucht(admin, fake_imap):
"""Ohne Zielkonto oder mit unsicherer Erkennung: liegen lassen, nicht durchwinken."""
quelle = _postfach(admin)
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={"from_contains": "revolut"})
roh = (MUSTER / "revolut_ueberweisung.eml").read_bytes().replace(
b"<revolut-muster-1@", b"<unklar-bleibt@")
import unittest.mock as _m
with _m.patch.object(mail_import, "_verbinden", lambda q: FakeImap([roh])):
admin.post(f"/api/mail/sources/{quelle['id']}/fetch")
unklare = [e for e in admin.get("/api/inbox").json() if e["status"] == "unclear"]
assert unklare
admin.post("/api/inbox/book-all")
danach = admin.get("/api/inbox").json()
assert all(e["status"] == "unclear" for e in danach if e["id"] == unklare[0]["id"])