GuidoandClaude Opus 5
60b8178c57
Der Zeitstempel des letzten Abrufs wird auch dann gesetzt, wenn ein Abruf
nichts fand – und danach sucht Certain nur noch ab gestern. Wer das Postfach
einrichtet, nachdem die Mails schon da waren, findet sie nie: Der erste,
ergebnislose Abruf schiebt das Fenster an ihnen vorbei.
Jetzt gibt es "Aelteres nachholen": Es sucht die letzten 90 Tage und uebergeht
den Zeitstempel. Dass dabei nichts doppelt ankommt, sichert wie bisher die
Message-ID.
Ausserdem sagt die Meldung jetzt, AB WANN gesucht wurde. Bisher stand dort
"Stimmt der Ordner?" – die Frage war fast immer die falsche.
Und ein latenter Fehler, der im August nicht auffaellt: Das Suchdatum wurde mit
strftime("%b") gebaut, also mit dem Monatskuerzel in der Sprache des Servers.
IMAP verlangt die englischen; auf einem deutschen System haette die Suche im
Maerz, Mai, Oktober und Dezember nichts gefunden. Die Kuerzel stehen jetzt fest
im Code.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
497 lines
21 KiB
Python
497 lines
21 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"]]
|
||
|
||
|
||
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"])
|
||
|
||
|
||
def test_suchdatum_ist_sprachunabhaengig():
|
||
"""IMAP verlangt die englischen Monatskuerzel. strftime("%b") richtet sich
|
||
nach der Sprache des Servers – auf Deutsch kaeme "Mär"/"Okt"/"Dez" heraus
|
||
und die Suche faende nichts. Im August faellt das nicht auf."""
|
||
from datetime import date
|
||
|
||
from app.services.mail_import import _imap_datum
|
||
|
||
assert _imap_datum(date(2026, 3, 9)) == "09-Mar-2026"
|
||
assert _imap_datum(date(2026, 5, 1)) == "01-May-2026"
|
||
assert _imap_datum(date(2026, 10, 31)) == "31-Oct-2026"
|
||
assert _imap_datum(date(2026, 12, 24)) == "24-Dec-2026"
|
||
|
||
|
||
def test_zeitraum_uebergeht_den_letzten_abruf(admin, monkeypatch):
|
||
"""Der Zeitstempel wird auch gesetzt, wenn ein Abruf nichts fand. Aeltere
|
||
Mails waeren danach fuer immer ausserhalb des Fensters – genau deshalb
|
||
muss sich der Zeitraum uebersteuern lassen."""
|
||
from datetime import date, datetime, timedelta, timezone
|
||
|
||
from app.database import SessionLocal
|
||
from app.models import MailSource
|
||
from app.services import mail_import
|
||
|
||
konto = _konto(admin, name="Nachholkonto", currency="CHF")
|
||
quelle = _postfach(admin)
|
||
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
|
||
"from_contains": "raiffeisen", "account_id": konto["id"]})
|
||
|
||
# Ein Abruf hat schon stattgefunden – ohne Treffer.
|
||
db = SessionLocal()
|
||
try:
|
||
q = db.get(MailSource, quelle["id"])
|
||
q.last_poll_at = datetime.now(timezone.utc)
|
||
db.commit()
|
||
q = db.get(MailSource, quelle["id"])
|
||
eng, seit_eng = mail_import._suchkriterium(q)
|
||
weit, seit_weit = mail_import._suchkriterium(q, 90)
|
||
finally:
|
||
db.close()
|
||
|
||
assert seit_eng >= date.today() - timedelta(days=2), "sucht nur noch ab gestern"
|
||
assert seit_weit == date.today() - timedelta(days=90)
|
||
assert "SINCE" in eng and "SINCE" in weit
|
||
|
||
# Und der Endpunkt nimmt den Zeitraum an.
|
||
assert admin.post(f"/api/mail/sources/{quelle['id']}/fetch",
|
||
params={"tage": 400}).status_code == 400
|