GuidoandClaude Opus 5
7b087247a5
Der Loesch-Endpunkt war da, aber ohne Knopf in der Oberflaeche, und
Bearbeiten fehlte ganz – eine falsch eingetippte Zuordnung liess sich also
nicht mehr korrigieren.
Jede Zuordnung ist jetzt eine eigene Zeile mit Erkennungstext und Konto;
der Speichern-Knopf erscheint erst bei einer Aenderung. Dazu ein PUT auf
/mail/senders/{id} im Backend.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
342 lines
13 KiB
Python
342 lines
13 KiB
Python
"""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"]]
|