"""Bankanbindung über PSD2 gegen einen Doppelgänger der Schnittstelle. Ein echter Zugang wäre im Test weder verfügbar noch wünschenswert: Er hinge an einer registrierten Anwendung, einem privaten Schlüssel und echten Bankkonten. Der Doppelgänger unten spricht nur die vier Endpunkte, die ``bank_api`` benutzt. Was hier NICHT gefälscht wird, ist die Kryptografie: Das JWT wird echt signiert und gegen den öffentlichen Schlüssel geprüft. Genau daran scheitert die Anbindung sonst als Erstes, und der Fehler wäre von aussen nicht zu sehen. """ import base64 import json import secrets from datetime import date, timedelta import pytest from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import padding, rsa from app.services import bank_api, bank_sync SCHLUESSEL = rsa.generate_private_key(public_exponent=65537, key_size=2048) PEM = SCHLUESSEL.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption(), ).decode() class FakeBank: """Minimale Gegenstelle: nur was ``BankClient`` tatsächlich aufruft.""" def __init__(self, umsaetze=None, konten=None, seiten=1, salden=None): self.umsaetze = umsaetze or [] self.salden = salden if salden is not None else [ {"balance_amount": {"amount": "100.00", "currency": "EUR"}}, ] self.konten = konten if konten is not None else [{ "uid": "acc-1", "name": "Girokonto", "currency": "EUR", "account_id": {"iban": "DE02120300000000202051"}, }] self.seiten = seiten self.aufrufe: list[str] = [] self.date_from: str | None = None self.jwt: str | None = None def __call__(self, methode, pfad, headers=None, json=None, **kw): self.jwt = (headers or {}).get("Authorization", "")[7:] self.aufrufe.append(f"{methode} {pfad.split('?')[0]}") if pfad.startswith("/aspsps"): alle = [{"name": "N26", "country": "DE"}, {"name": "Revolut", "country": "LT"}] if "country=" in pfad: land = pfad.split("country=")[1].split("&")[0] alle = [b for b in alle if b["country"] == land] return 200, {"aspsps": alle} if pfad == "/auth": return 200, {"url": "https://bank.example/authorize?x=1", "authorization_id": "auth-1"} if pfad == "/sessions": if json.get("code") != "guter-code": return 400, {"message": "Code ungültig"} return 200, {"session_id": "sess-1", "accounts": self.konten, "access": {"valid_until": "2027-01-01T00:00:00+00:00"}} if "/balances" in pfad: return 200, {"balances": self.salden} if "/transactions" in pfad: if "date_from=" in pfad: self.date_from = pfad.split("date_from=")[1].split("&")[0] # Wie die echte Schnittstelle nur ab dem angefragten Tag liefern – # sonst prüfte kein Test, ob das Startdatum überhaupt wirkt. passend = [u for u in self.umsaetze if not self.date_from or str(u.get("booking_date", ""))[:10] >= self.date_from] # Blättern nachstellen: erst mit Fortsetzungsschlüssel, dann ohne. hat_schluessel = "continuation_key=" in pfad if self.seiten > 1 and not hat_schluessel: return 200, {"transactions": passend[:1], "continuation_key": "s2"} rest = passend[1:] if self.seiten > 1 else passend return 200, {"transactions": rest} return 404, {"message": "unbekannt"} @pytest.fixture() def fake_bank(monkeypatch): """Ersetzt den HTTP-Aufruf; gibt eine Funktion zum Bestücken zurück.""" def bestuecken(**kw): gegenstelle = FakeBank(**kw) def _anfrage(self, methode, pfad, **rest): code, daten = gegenstelle(methode, pfad, **rest) if code >= 400: raise bank_api.BankFehler(daten.get("message", "Fehler")) return daten monkeypatch.setattr(bank_api.BankClient, "_anfrage", _anfrage) return gegenstelle return bestuecken def _app(admin, **kw): body = {"application_id": "app-123", "private_key": PEM, "base_url": "https://certain.example.com"} body.update(kw) r = admin.put("/api/bank/app", json=body) assert r.status_code == 200, r.text return r.json() def _konto(admin, **kw): body = {"name": "N26 Konto", "type": "girokonto", "currency": "EUR", "opening_balance_cents": 0} body.update(kw) return admin.post("/api/accounts", json=body).json() UMSATZ = { "entry_reference": "tx-1", "booking_date": "2026-08-20", "transaction_amount": {"amount": "14.90", "currency": "EUR"}, "credit_debit_indicator": "DBIT", "creditor": {"name": "SPEEDEE RESTAURATIONS"}, "remittance_information": ["Kartenzahlung"], } # --- Zugangsdaten ------------------------------------------------------------ def test_schluessel_wird_verschluesselt_abgelegt_und_nie_ausgeliefert(admin): from app.database import SessionLocal from app.models import BankApp from app.services import geheimnis ausgabe = _app(admin) assert "private_key" not in ausgabe assert ausgabe["key_set"] is True db = SessionLocal() try: zeile = db.query(BankApp).order_by(BankApp.id.desc()).first() assert "PRIVATE KEY" not in zeile.private_key_enc, "Klartext in der Datenbank" assert geheimnis.entschluesseln(zeile.private_key_enc) == PEM finally: db.close() def test_rueckleitung_ohne_https_wird_abgelehnt(admin): """Banken leiten nicht auf eine unverschlüsselte Adresse zurück.""" r = admin.put("/api/bank/app", json={ "application_id": "app-123", "private_key": PEM, "base_url": "http://192.168.1.50:8071"}) assert r.status_code == 422 assert "https" in r.text.lower() def test_ohne_hinterlegte_anwendung_kommt_eine_klare_ansage(admin): admin.delete("/api/bank/app") r = admin.get("/api/bank/banks?country=DE") assert r.status_code == 400 assert "Anwendung" in r.json()["detail"] # --- JWT --------------------------------------------------------------------- def test_jwt_ist_echt_signiert_und_nachpruefbar(): """Die Gegenstelle prüft die Signatur – wenn sie hier stimmt, stimmt sie dort.""" jwt = bank_api.baue_jwt("app-123", PEM) kopf_b64, rumpf_b64, sig_b64 = jwt.split(".") pad = lambda s: s + "=" * (-len(s) % 4) kopf = json.loads(base64.urlsafe_b64decode(pad(kopf_b64))) assert kopf == {"typ": "JWT", "alg": "RS256", "kid": "app-123"} rumpf = json.loads(base64.urlsafe_b64decode(pad(rumpf_b64))) assert rumpf["iss"] == "enablebanking.com" assert rumpf["aud"] == "api.enablebanking.com" assert rumpf["exp"] - rumpf["iat"] == 3600 SCHLUESSEL.public_key().verify( base64.urlsafe_b64decode(pad(sig_b64)), f"{kopf_b64}.{rumpf_b64}".encode(), padding.PKCS1v15(), hashes.SHA256()) def test_unbrauchbarer_schluessel_meldet_sich_verstaendlich(): with pytest.raises(bank_api.BankFehler) as fehler: bank_api.baue_jwt("app-123", "das ist kein schluessel") assert "pem" in str(fehler.value).lower() # --- Verbinden --------------------------------------------------------------- def test_banken_eines_landes_abrufen(admin, fake_bank): """Ein Land liefert nur dessen Banken – Revolut sitzt in Litauen.""" _app(admin) fake_bank() r = admin.get("/api/bank/banks?country=DE") assert r.status_code == 200, r.text assert [b["name"] for b in r.json()["treffer"]] == ["N26"] def test_verbinden_und_konten_finden(admin, fake_bank): _app(admin) gegenstelle = fake_bank() start = admin.post("/api/bank/connect", json={"aspsp_name": "N26", "country": "DE"}) assert start.status_code == 200, start.text assert start.json()["url"].startswith("https://") state = start.json()["state"] fertig = admin.post(f"/api/bank/callback?state={state}&code=guter-code") assert fertig.status_code == 200, fertig.text assert fertig.json()["konten"] == 1 verknuepfungen = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == "acc-1"] assert len(verknuepfungen) == 1 assert verknuepfungen[0]["iban"] == "DE02120300000000202051" # Noch keinem Konto zugeordnet – das entscheidet der Nutzer. assert verknuepfungen[0]["account_id"] is None assert "POST /auth" in gegenstelle.aufrufe and "POST /sessions" in gegenstelle.aufrufe def test_falscher_code_wird_verstaendlich_abgewiesen(admin, fake_bank): _app(admin) fake_bank() state = admin.post("/api/bank/connect", json={"aspsp_name": "N26", "country": "DE"}).json()["state"] r = admin.post(f"/api/bank/callback?state={state}&code=falsch") assert r.status_code == 400 assert "ungültig" in r.json()["detail"] def test_fremder_state_findet_keine_verbindung(admin, fake_bank): _app(admin) fake_bank() r = admin.post("/api/bank/callback?state=ausgedacht&code=guter-code") assert r.status_code == 400 assert "keine offene Verbindung" in r.json()["detail"] # --- Kontozuordnung ---------------------------------------------------------- def _verbunden(admin, fake_bank_fn, **kw): """Verbindung mit EIGENER Konto-Kennung je Aufruf. Die Zuordnung eines Bankkontos ist absichtlich dauerhaft: Verbindet man dieselbe Bank neu, bleibt sie erhalten. In der gemeinsamen Testdatenbank würde derselbe uid deshalb die Zuordnung eines früheren Tests erben. """ _app(admin) kw.setdefault("konten", [{ "uid": f"acc-{secrets.token_hex(4)}", "name": "Girokonto", "currency": "EUR", "account_id": {"iban": "DE02120300000000202051"}, }]) gegenstelle = fake_bank_fn(**kw) # Eigener Bankname je Aufruf: Erneutes Verbinden derselben Bank frischt # bewusst den bestehenden Eintrag auf, sonst erbte jeder Test die # Verbindung des vorigen. bank = f"Testbank-{secrets.token_hex(3)}" state = admin.post("/api/bank/connect", json={"aspsp_name": bank, "country": "DE"}).json()["state"] antwort = admin.post(f"/api/bank/callback?state={state}&code=guter-code").json() link = [v for v in admin.get("/api/bank/links").json() if v["connection_id"] == antwort["connection_id"]][0] return antwort["connection_id"], link, gegenstelle def test_zuordnung_verlangt_dieselbe_waehrung(admin, fake_bank): """Ein EUR-Konto in ein CHF-Konto zu schreiben verfälscht jeden Saldo.""" _, link, _ = _verbunden(admin, fake_bank) chf = _konto(admin, name="Franken-Konto", currency="CHF") r = admin.post(f"/api/bank/links/{link['id']}/account/{chf['id']}") assert r.status_code == 400 assert "Währungen passen nicht" in r.json()["detail"] eur = _konto(admin, name="Euro-Konto", currency="EUR") ok = admin.post(f"/api/bank/links/{link['id']}/account/{eur['id']}") assert ok.status_code == 200, ok.text assert ok.json()["account_id"] == eur["id"] # --- Umsätze holen ----------------------------------------------------------- def test_ohne_zuordnung_wird_nichts_geholt(admin, fake_bank): verbindung, _, _ = _verbunden(admin, fake_bank, umsaetze=[UMSATZ]) r = admin.post(f"/api/bank/connections/{verbindung}/sync") assert r.status_code == 200, r.text assert r.json()["ohne_zuordnung"] is True assert r.json()["neu"] == 0 def test_umsatz_wird_sofort_gebucht(admin, fake_bank): """Ein angebundenes Konto ist die Quelle der Wahrheit: Was die Bank gebucht hat, IST gebucht. Nur so kann Certains Saldo gar nicht erst abweichen.""" verbindung, link, _ = _verbunden(admin, fake_bank, umsaetze=[UMSATZ]) konto = _konto(admin, name="N26 fuer Abruf", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") r = admin.post(f"/api/bank/connections/{verbindung}/sync") assert r.status_code == 200, r.text assert r.json()["neu"] == 1 assert r.json()["gebucht"] == 1 treffer = [e for e in admin.get("/api/inbox").json() if e["payee"] == "SPEEDEE RESTAURATIONS"] assert len(treffer) == 1 # DBIT ist eine Ausgabe – das Vorzeichen steckt in der Richtung, nicht im Betrag. assert treffer[0]["amount_cents"] == -1490 assert treffer[0]["date"] == "2026-08-20" assert treffer[0]["source"] == "bank" assert treffer[0]["status"] == "booked" assert treffer[0]["transaction_id"] is not None # Der Saldo steht sofort richtig, ohne dass jemand etwas tun musste. assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -1490 def test_zweiter_abruf_bucht_nicht_doppelt(admin, fake_bank): """Der Rueckgriff hinter den letzten Abruf holt dieselben Umsaetze noch einmal. Die Kennung des Umsatzes muss sie exakt abfangen.""" # Ein Umsatz von heute: Der zweite Abruf greift nur wenige Tage zurueck, # ein alter faende sich dort gar nicht wieder. heute = dict(UMSATZ, booking_date=date.today().isoformat()) verbindung, link, _ = _verbunden(admin, fake_bank, umsaetze=[heute]) konto = _konto(admin, name="Kein Doppel", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") admin.post(f"/api/bank/connections/{verbindung}/sync") zweiter = admin.post(f"/api/bank/connections/{verbindung}/sync").json() assert zweiter["geholt"] >= 1, "derselbe Umsatz kam wieder" assert zweiter["neu"] == 0 assert zweiter["gebucht"] == 0 assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -1490 def test_vorgemerkte_werden_nicht_gebucht_sondern_gemeldet(admin, fake_bank): """Vorgemerkte kommen spaeter noch einmal – dann mit anderer Kennung, weil die Bank beim Buchen eine neue vergibt. Beides zu nehmen hiesse doppelt.""" schwebend = dict(UMSATZ, entry_reference="tx-pdng", status="PDNG") verbindung, link, _ = _verbunden(admin, fake_bank, umsaetze=[schwebend]) konto = _konto(admin, name="Nur Vormerkung", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") r = admin.post(f"/api/bank/connections/{verbindung}/sync").json() assert r["vorgemerkt"] == 1 assert r["neu"] == 0 # Nicht stillschweigend verworfen – die Zahl steht in der Rueckmeldung. assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 0 def test_erster_abruf_greift_nicht_hinter_vorhandene_buchungen(admin, fake_bank): """Sonst holte die Bank Monate, die schon per CSV eingelesen sind. Der Duplikatschutz faende die nicht: Sein Schluessel enthaelt den Empfaengertext, und der lautet bei der Bank anders als im Auszug.""" verbindung, link, gegenstelle = _verbunden(admin, fake_bank, umsaetze=[UMSATZ]) konto = _konto(admin, name="Schon gepflegt", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") # So sieht ein per CSV eingelesener August aus. admin.post("/api/transactions", json={ "type": "expense", "date": "2026-08-20", "account_id": konto["id"], "amount_cents": 1490, "payee": "SPEEDEE RESTAURATIONS ZUERICH", }) r = admin.post(f"/api/bank/connections/{verbindung}/sync").json() assert r["ab"] == "2026-08-21", "ab dem Tag NACH der letzten Buchung" assert gegenstelle.date_from == "2026-08-21" # Der Umsatz vom 20. liegt davor und kommt gar nicht erst an. assert r["neu"] == 0 assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -1490 def test_gutschrift_kommt_als_einnahme_an(admin, fake_bank): gutschrift = dict(UMSATZ, entry_reference="tx-crdt", credit_debit_indicator="CRDT", creditor=None, debtor={"name": "Arbeitgeber AG"}) verbindung, link, _ = _verbunden(admin, fake_bank, umsaetze=[gutschrift]) konto = _konto(admin, name="N26 Einnahme", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") admin.post(f"/api/bank/connections/{verbindung}/sync") treffer = [e for e in admin.get("/api/inbox").json() if e["payee"] == "Arbeitgeber AG"] assert treffer and treffer[0]["amount_cents"] == 1490 def test_zweiter_abruf_holt_nichts_doppelt(admin, fake_bank): """Die Kennung des Umsatzes zählt – sonst wüchse der Eingang bei jedem Abruf.""" verbindung, link, _ = _verbunden(admin, fake_bank, umsaetze=[UMSATZ]) konto = _konto(admin, name="N26 zweimal", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") erst = admin.post(f"/api/bank/connections/{verbindung}/sync").json() zweit = admin.post(f"/api/bank/connections/{verbindung}/sync").json() assert erst["neu"] == 1 assert zweit["neu"] == 0 def test_blaettern_holt_alle_seiten(admin, fake_bank): """Ohne die Fortsetzungs-Schleife käme still nur die erste Seite an.""" umsaetze = [dict(UMSATZ, entry_reference=f"tx-{i}") for i in range(2)] verbindung, link, _ = _verbunden(admin, fake_bank, umsaetze=umsaetze, seiten=2) konto = _konto(admin, name="N26 blaettern", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") assert admin.post(f"/api/bank/connections/{verbindung}/sync").json()["neu"] == 2 def test_ablauf_der_zustimmung_wird_gemeldet(admin, fake_bank): """Läuft die Zustimmung ab, muss das sichtbar sein statt still zu enden.""" from datetime import datetime, timedelta, timezone from app.database import SessionLocal from app.models import BankConnection verbindung, _, _ = _verbunden(admin, fake_bank) liste = [c for c in admin.get("/api/bank/connections").json() if c["id"] == verbindung] assert liste and liste[0]["expires_soon"] is False db = SessionLocal() try: zeile = db.get(BankConnection, verbindung) zeile.valid_until = datetime.now(timezone.utc) + timedelta(days=3) db.commit() finally: db.close() liste = [c for c in admin.get("/api/bank/connections").json() if c["id"] == verbindung] assert liste[0]["expires_soon"] is True def test_suche_findet_ueber_alle_laender(admin, fake_bank): """Das Sitzland steht nicht in der IBAN. Revolut vergibt deutsche IBANs, ist aber in Litauen ansässig. Wer unter „Deutschland" sucht, findet es dort nie – deshalb muss ohne Land gesucht werden können. """ _app(admin) fake_bank() alle = admin.get("/api/bank/banks?search=revolut") assert alle.status_code == 200, alle.text daten = alle.json() assert [b["name"] for b in daten["treffer"]] == ["Revolut"] assert daten["treffer"][0]["country"] == "LT" # Die Gesamtzahl kommt mit: Sonst sieht „keine Treffer" gleich aus, egal ob # die Liste leer kam oder nur der Begriff nicht passte. assert daten["gesamt"] == 2 # Unter Deutschland gesucht liefert dieselbe Bank NICHT. nur_de = admin.get("/api/bank/banks?country=DE&search=revolut").json() assert nur_de["treffer"] == [] assert nur_de["gesamt"] == 1 def test_suche_ohne_land_und_ohne_begriff_wird_abgelehnt(admin, fake_bank): """Sonst käme die vollständige Liste aller Banken Europas ins Formular.""" _app(admin) fake_bank() r = admin.get("/api/bank/banks") assert r.status_code == 400 assert "Suchbegriff" in r.json()["detail"] def test_leere_bankliste_ist_von_keinem_treffer_unterscheidbar(admin, fake_bank): """„Keine Treffer" darf nicht beides heissen. Kam die Liste leer, liegt es an der Anwendung; passte nur der Begriff nicht, liegt es am Begriff. Ohne die Gesamtzahl sucht man den Fehler an der falschen Stelle – das hat schon einmal einen Abend gekostet. """ _app(admin) gegenstelle = fake_bank() gegenstelle.banken_leer = True def leer(methode, pfad, **kw): if pfad.startswith("/aspsps"): return 200, {"aspsps": []} return FakeBank.__call__(gegenstelle, methode, pfad, **kw) import app.services.bank_api as bank_api_modul urspruenglich = bank_api_modul.BankClient._anfrage def _anfrage(self, methode, pfad, **rest): code, daten = leer(methode, pfad, **rest) if code >= 400: raise bank_api_modul.BankFehler(daten.get("message", "Fehler")) return daten bank_api_modul.BankClient._anfrage = _anfrage try: daten = admin.get("/api/bank/banks?search=revolut").json() assert daten["gesamt"] == 0 assert daten["treffer"] == [] finally: bank_api_modul.BankClient._anfrage = urspruenglich def test_treffer_werden_nach_guete_gekuerzt_nicht_alphabetisch(admin, monkeypatch): """Der Fehler, der einen Abend gekostet hat. Die Trefferliste wird bei 200 gekürzt. Wurde vorher alphabetisch sortiert, fiel „N26" bei der Suche nach „n" hinter zweihundert Banken heraus, die irgendwo ein n im Namen tragen – die Bank war da, tauchte aber nie auf. Gekürzt wird deshalb erst NACH der Bewertung. """ _app(admin) # Dreihundert Banken, die alle ein „n" enthalten und alphabetisch vor # „N26" liegen – plus die gesuchte Bank. viele = [{"name": f"Aachener Bank {i:03d}", "country": "DE"} for i in range(300)] viele.append({"name": "N26", "country": "DE"}) viele.append({"name": "Revolut", "country": "LT"}) def _anfrage(self, methode, pfad, **rest): if pfad.startswith("/aspsps"): return {"aspsps": viele} raise AssertionError(f"unerwartet: {pfad}") monkeypatch.setattr(bank_api.BankClient, "_anfrage", _anfrage) daten = admin.get("/api/bank/banks?search=n").json() namen = [b["name"] for b in daten["treffer"]] assert "N26" in namen, "N26 muss trotz Kürzung sichtbar sein" assert namen[0] == "N26", "Der beste Treffer gehört nach oben" # Die echte Trefferzahl, nicht die gekürzte – sonst steht dort immer 200. # 300 Aachener + N26; "Revolut" enthält kein n und zählt nicht mit. assert daten["treffer_gesamt"] == 301 assert len(daten["treffer"]) == 200 assert daten["gekuerzt"] is True def test_genauer_name_schlaegt_teiltreffer(admin, monkeypatch): _app(admin) liste = [{"name": "Revolut Bank UAB", "country": "LT"}, {"name": "Revolut", "country": "LT"}, {"name": "Banca Revoluta", "country": "IT"}] def _anfrage(self, methode, pfad, **rest): return {"aspsps": liste} monkeypatch.setattr(bank_api.BankClient, "_anfrage", _anfrage) namen = [b["name"] for b in admin.get("/api/bank/banks?search=revolut").json()["treffer"]] assert namen == ["Revolut", "Revolut Bank UAB", "Banca Revoluta"] def test_bereits_bekanntes_konto_zaehlt_nicht_als_null(admin, fake_bank): """„0 Konten" darf nicht heissen, dass alles schon bekannt war. Beim erneuten Verbinden derselben Bank blieb die Zaehlung auf null stehen, obwohl die Konten da waren – das liest sich wie ein Fehlschlag. """ _app(admin) konten = [{"uid": f"acc-{secrets.token_hex(4)}", "name": "Girokonto", "currency": "EUR", "account_id": {"iban": "DE02120300000000202051"}}] fake_bank(konten=konten) def verbinde(): state = admin.post("/api/bank/connect", json={"aspsp_name": "N26", "country": "DE"}).json()["state"] return admin.post(f"/api/bank/callback?state={state}&code=guter-code").json() erst = verbinde() assert erst["neu"] == 1 and erst["konten"] == 1 zweit = verbinde() assert zweit["neu"] == 0 assert zweit["bekannt"] == 1 assert zweit["konten"] == 1, "Bekannte Konten sind trotzdem nutzbar" def test_konten_ohne_kennung_werden_gemeldet_statt_verworfen(admin, fake_bank): """Ohne diese Zahl sieht „nichts übermittelt" genauso aus wie „alles verworfen" – und man sucht den Fehler bei der falschen Seite.""" _app(admin) fake_bank(konten=[{"name": "Ohne Kennung", "currency": "EUR"}]) state = admin.post("/api/bank/connect", json={"aspsp_name": "N26", "country": "DE"}).json()["state"] antwort = admin.post(f"/api/bank/callback?state={state}&code=guter-code").json() assert antwort["uebermittelt"] == 1 assert antwort["ohne_kennung"] == 1 assert antwort["konten"] == 0 def test_erneutes_verbinden_behaelt_die_kontozuordnung(admin, fake_bank): """Ohne das müsste man vor dem Neuverbinden löschen – und verlöre dabei die Zuordnungen, die man gerade behalten will.""" _app(admin) uid = f"acc-{secrets.token_hex(4)}" konten = [{"uid": uid, "name": "Girokonto", "currency": "EUR", "account_id": {"iban": "DE02120300000000202051"}}] fake_bank(konten=konten) def verbinde(): state = admin.post("/api/bank/connect", json={"aspsp_name": "Testbank", "country": "DE"}).json()["state"] return admin.post(f"/api/bank/callback?state={state}&code=guter-code").json() erste = verbinde() link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] konto = _konto(admin, name="Zuordnung bleibt", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") zweite = verbinde() assert zweite["connection_id"] == erste["connection_id"], "kein zweiter Eintrag" danach = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid] assert len(danach) == 1 assert danach[0]["account_id"] == konto["id"], "die Zuordnung muss bleiben" # Und es steht genau eine Verbindung zu dieser Bank da. passende = [c for c in admin.get("/api/bank/connections").json() if c["aspsp_name"] == "Testbank"] assert len(passende) == 1 def test_pocket_ohne_iban_und_namen_bleibt_unterscheidbar(admin, fake_bank): """Ein Revolut-Pocket bringt weder IBAN noch eigenen Namen mit. Ohne Produktbezeichnung und Ersatzkennung sähen mehrere Konten gleich aus.""" _app(admin) haupt = f"acc-{secrets.token_hex(4)}" pocket = f"acc-{secrets.token_hex(4)}" fake_bank(konten=[ {"uid": haupt, "name": "Revolut", "currency": "CHF", "account_id": {"iban": "CH9300762011623852957"}, "product": "Current"}, # So kommt ein Pocket an: keine IBAN, kein Name, nur Produkt und # eine Kennung unter ``other``. {"uid": pocket, "currency": "CHF", "product": "Steuern", "account_id": {"other": {"identification": "POCKET-7788"}}}, ]) state = admin.post("/api/bank/connect", json={"aspsp_name": "Revolut", "country": "LT"}).json()["state"] r = admin.post(f"/api/bank/callback?state={state}&code=guter-code") assert r.status_code == 200, r.text assert r.json()["konten"] == 2 links = {v["remote_uid"]: v for v in admin.get("/api/bank/links").json()} p = links[pocket] assert p["product"] == "Steuern" assert p["other_id"] == "POCKET-7788" assert p["iban"] is None # Der Name faellt auf etwas Sprechendes zurueck, nicht auf die uid. assert p["remote_name"] == "Steuern" assert p["remote_name"] != p["remote_uid"] assert p["bank_felder"]["product"] == "Steuern" # Und die beiden sind auseinanderzuhalten. assert links[haupt]["remote_name"] != p["remote_name"] def test_merkmale_werden_beim_neuverbinden_aufgefrischt(admin, fake_bank): """Sonst muesste man loeschen, um an eine spaetere Produktbezeichnung zu kommen – und verloere dabei die Zuordnung.""" _app(admin) uid = f"acc-{secrets.token_hex(4)}" bank = f"Testbank-{secrets.token_hex(3)}" def verbinde(konten): fake_bank(konten=konten) state = admin.post("/api/bank/connect", json={"aspsp_name": bank, "country": "DE"}).json()["state"] return admin.post(f"/api/bank/callback?state={state}&code=guter-code") verbinde([{"uid": uid, "currency": "EUR"}]) assert admin.get("/api/bank/links").json()[-1]["product"] is None verbinde([{"uid": uid, "currency": "EUR", "name": "Urlaub", "product": "Savings"}]) danach = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] assert danach["product"] == "Savings" assert danach["remote_name"] == "Urlaub" def test_saldo_der_bank_wird_gefuehrt_und_gebuchter_bevorzugt(admin, fake_bank): """Banken melden mehrere Saldi nebeneinander. Den ersten zu nehmen waere Zufall – vergleichbar mit Certains eigener Rechnung ist nur der gebuchte.""" _app(admin) uid = f"acc-{secrets.token_hex(4)}" fake_bank( konten=[{"uid": uid, "name": "Girokonto", "currency": "EUR", "account_id": {"iban": "DE02120300000000202051"}}], salden=[ # Verfuegbar steht zuerst und ist hoeher – er enthaelt den # Ueberziehungsrahmen und darf gerade NICHT gewinnen. {"balance_type": "ITAV", "balance_amount": {"amount": "1500.00", "currency": "EUR"}}, {"balance_type": "CLBD", "balance_amount": {"amount": "342.15", "currency": "EUR"}, "reference_date": "2026-08-25"}, ], ) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Saldobank-{secrets.token_hex(3)}", "country": "DE"}).json()["state"] # Die Kennung aus der Rueckmeldung nehmen, nicht die zuletzt angelegte # Verbindung: Die Testdatenbank lebt ueber den ganzen Lauf. verbindung_id = admin.post( f"/api/bank/callback?state={state}&code=guter-code").json()["connection_id"] r = admin.post(f"/api/bank/connections/{verbindung_id}/salden") assert r.status_code == 200, r.text assert r.json()["saldi"] == 1 link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] assert link["balance_cents"] == 34215 assert link["balance_type"] == "CLBD" assert link["balance_at"].startswith("2026-08-25") def test_saldo_kommt_auch_ohne_zuordnung(admin, fake_bank): """Genau dann braucht man ihn: Bei mehreren Pockets ist der Betrag oft das Einzige, woran man erkennt, welches man vor sich hat.""" _app(admin) uid = f"acc-{secrets.token_hex(4)}" fake_bank( konten=[{"uid": uid, "currency": "CHF", "product": "Steuern"}], salden=[{"balance_type": "CLBD", "balance_amount": {"amount": "2480.00", "currency": "CHF"}}], ) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Revolut-{secrets.token_hex(3)}", "country": "LT"}).json()["state"] verbindung_id = admin.post( f"/api/bank/callback?state={state}&code=guter-code").json()["connection_id"] r = admin.post(f"/api/bank/connections/{verbindung_id}/sync") assert r.status_code == 200, r.text # Kein Konto zugeordnet – trotzdem ist der Saldo da. assert r.json()["ohne_zuordnung"] is True assert r.json()["saldi"] == 1 link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] assert link["balance_cents"] == 248000 def test_abruf_je_konto_ruehrt_das_andere_nicht_an(admin, fake_bank): """Ein Abruf bucht. Wer ein Konto abruft, will nicht, dass nebenbei auf einem zweiten Buchungen entstehen.""" _app(admin) eins, zwei = f"acc-{secrets.token_hex(4)}", f"acc-{secrets.token_hex(4)}" heute = date.today().isoformat() fake_bank( konten=[{"uid": eins, "name": "Erstes", "currency": "EUR"}, {"uid": zwei, "name": "Zweites", "currency": "EUR"}], umsaetze=[dict(UMSATZ, booking_date=heute)], ) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Zweikonto-{secrets.token_hex(3)}", "country": "DE"}).json()["state"] admin.post(f"/api/bank/callback?state={state}&code=guter-code") links = {v["remote_uid"]: v for v in admin.get("/api/bank/links").json()} ka = _konto(admin, name=f"Konto A {secrets.token_hex(3)}", currency="EUR") kb = _konto(admin, name=f"Konto B {secrets.token_hex(3)}", currency="EUR") admin.post(f"/api/bank/links/{links[eins]['id']}/account/{ka['id']}") admin.post(f"/api/bank/links/{links[zwei]['id']}/account/{kb['id']}") r = admin.post(f"/api/bank/links/{links[eins]['id']}/sync") assert r.status_code == 200, r.text assert r.json()["gebucht"] == 1 assert admin.get(f"/api/accounts/{ka['id']}").json()["balance_cents"] == -1490 assert admin.get(f"/api/accounts/{kb['id']}").json()["balance_cents"] == 0, \ "das zweite Konto wurde angefasst" def test_einzelabruf_verschiebt_nicht_das_startdatum_des_anderen(admin, fake_bank): """Haenge der Stempel an der Verbindung, schoebe ein Einzelabruf das Startdatum der uebrigen Konten mit vor – deren Umsaetze kaemen nie an.""" _app(admin) eins, zwei = f"acc-{secrets.token_hex(4)}", f"acc-{secrets.token_hex(4)}" alt = (date.today() - timedelta(days=30)).isoformat() gegenstelle = fake_bank( konten=[{"uid": eins, "currency": "EUR"}, {"uid": zwei, "currency": "EUR"}], umsaetze=[dict(UMSATZ, booking_date=alt)], ) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Stempel-{secrets.token_hex(3)}", "country": "DE"}).json()["state"] admin.post(f"/api/bank/callback?state={state}&code=guter-code") links = {v["remote_uid"]: v for v in admin.get("/api/bank/links").json()} ka = _konto(admin, name=f"Stempel A {secrets.token_hex(3)}", currency="EUR") kb = _konto(admin, name=f"Stempel B {secrets.token_hex(3)}", currency="EUR") admin.post(f"/api/bank/links/{links[eins]['id']}/account/{ka['id']}") admin.post(f"/api/bank/links/{links[zwei]['id']}/account/{kb['id']}") admin.post(f"/api/bank/links/{links[eins]['id']}/sync") # Das zweite Konto war noch nie dran – es muss weiterhin weit zurueckgreifen. admin.post(f"/api/bank/links/{links[zwei]['id']}/sync") assert gegenstelle.date_from < alt, f"greift nur bis {gegenstelle.date_from} zurueck" assert admin.get(f"/api/accounts/{kb['id']}").json()["balance_cents"] == -1490 def test_takt_null_laeuft_nie_und_sonst_erst_wenn_faellig(admin): """Nichts laeuft ungefragt los – ein automatischer Abruf bucht.""" from datetime import datetime, timezone from app.models import BankConnection from app.services import bank_sync jetzt = datetime.now(timezone.utc) aus = BankConnection(owner_id=1, aspsp_name="X", aspsp_country="DE", poll_minutes=0) assert bank_sync.faellig(aus, jetzt) is False nie_gelaufen = BankConnection(owner_id=1, aspsp_name="X", aspsp_country="DE", poll_minutes=720) assert bank_sync.faellig(nie_gelaufen, jetzt) is True frisch = BankConnection(owner_id=1, aspsp_name="X", aspsp_country="DE", poll_minutes=720, last_sync_at=jetzt - timedelta(minutes=30)) assert bank_sync.faellig(frisch, jetzt) is False # SQLite gibt Zeitstempel OHNE Zeitzone zurueck – das darf nicht werfen. ohne_zone = BankConnection(owner_id=1, aspsp_name="X", aspsp_country="DE", poll_minutes=720, last_sync_at=(jetzt - timedelta(hours=13)).replace(tzinfo=None)) assert bank_sync.faellig(ohne_zone, jetzt) is True def test_takt_wird_geprueft(admin, fake_bank): verbindung, _, _ = _verbunden(admin, fake_bank) assert admin.put(f"/api/bank/connections/{verbindung}/takt", params={"minutes": 15}).status_code == 400, "zu haeufig" assert admin.put(f"/api/bank/connections/{verbindung}/takt", params={"minutes": 0}).json()["poll_minutes"] == 0 assert admin.put(f"/api/bank/connections/{verbindung}/takt", params={"minutes": 720}).json()["poll_minutes"] == 720 def test_spaeter_gelieferte_gegenseite_wird_nicht_noch_einmal_gebucht(admin, fake_bank): """Der eigentliche Zweck der Gegenseite: Die Bank meldet den Zugang selbst. Ohne den Eingangs-Eintrag auf dem Zielkonto stuende das Geld zweimal da.""" _app(admin) uid = f"acc-{secrets.token_hex(4)}" heute = date.today() betrag = 30000 + secrets.randbelow(9000) fake_bank( konten=[{"uid": uid, "name": "Ziel", "currency": "CHF"}], umsaetze=[{ "entry_reference": f"tx-{secrets.token_hex(4)}", "booking_date": heute.isoformat(), "transaction_amount": {"amount": f"{betrag / 100:.2f}", "currency": "CHF"}, "credit_debit_indicator": "CRDT", "debtor": {"name": "Eigenes Konto"}, }], ) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Gegen-{secrets.token_hex(3)}", "country": "CH"}).json()["state"] admin.post(f"/api/bank/callback?state={state}&code=guter-code") link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] ziel = _konto(admin, name=f"Bank-Ziel {secrets.token_hex(3)}", currency="CHF") admin.post(f"/api/bank/links/{link['id']}/account/{ziel['id']}") quelle = _konto(admin, name=f"Quelle {secrets.token_hex(3)}", currency="CHF") # Den Uebertrag von Hand buchen – die Gegenseite entsteht dabei auf `ziel`. from tests.test_inbox import _eingang eintrag = _eingang(admin, quelle, -betrag, datum=heute.isoformat()) r = admin.post(f"/api/inbox/{eintrag['id']}/transfer", json={"to_account_id": ziel["id"]}) assert r.status_code == 200, r.text vorher = admin.get(f"/api/accounts/{ziel['id']}").json()["balance_cents"] assert vorher == betrag # Jetzt liefert die Bank denselben Zugang. abruf = admin.post(f"/api/bank/links/{link['id']}/sync").json() assert abruf["geholt"] == 1, "der Umsatz kam an" assert abruf["neu"] == 0, "und wurde als Gegenseite erkannt" assert admin.get(f"/api/accounts/{ziel['id']}").json()["balance_cents"] == vorher def test_kurs_der_bank_wird_uebernommen(admin, fake_bank): """Die Bank rechnet mit ihrem eigenen Kurs. Ihn wegzuwerfen und mit dem EZB-Kurs zu naehern, macht den Betrag der Gegenseite unbrauchbar.""" _app(admin) uid = f"acc-{secrets.token_hex(4)}" fake_bank( konten=[{"uid": uid, "currency": "CHF"}], umsaetze=[{ "entry_reference": f"tx-{secrets.token_hex(4)}", "booking_date": date.today().isoformat(), "transaction_amount": {"amount": "93.50", "currency": "CHF"}, "credit_debit_indicator": "DBIT", "creditor": {"name": "Eigenes Euro-Konto"}, "currency_exchange": { "source_currency": "CHF", "target_currency": "EUR", "exchange_rate": "1.0695", "instructed_amount": {"amount": "100.00", "currency": "EUR"}, }, }], ) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Kurs-{secrets.token_hex(3)}", "country": "CH"}).json()["state"] admin.post(f"/api/bank/callback?state={state}&code=guter-code") link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] chf = _konto(admin, name=f"Kurs CHF {secrets.token_hex(3)}", currency="CHF") admin.post(f"/api/bank/links/{link['id']}/account/{chf['id']}") admin.post(f"/api/bank/links/{link['id']}/sync") eintrag = [e for e in admin.get("/api/inbox").json() if e["payee"] == "Eigenes Euro-Konto"][0] assert eintrag["counter_amount_cents"] == 10000 assert eintrag["counter_currency"] == "EUR" assert float(eintrag["fx_rate"]) == 1.0695 def test_ohne_angabe_wird_nichts_geraten(admin, fake_bank): """Ein falscher Kurs sieht aus wie ein richtiger – lieber gar keiner.""" verbindung, link, _ = _verbunden(admin, fake_bank, umsaetze=[dict(UMSATZ, booking_date=date.today().isoformat())]) konto = _konto(admin, name=f"Ohne Kurs {secrets.token_hex(3)}", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") admin.post(f"/api/bank/connections/{verbindung}/sync") eintrag = [e for e in admin.get("/api/inbox").json() if e["payee"] == "SPEEDEE RESTAURATIONS"][0] assert eintrag["fx_rate"] is None assert eintrag["counter_amount_cents"] is None def test_geloeschte_buchung_kommt_beim_naechsten_abruf_nicht_zurueck(admin, fake_bank): """Die Frage beim Aufraeumen: Legt der naechste Abruf alles wieder an? Nein – der Umsatz wird an der Kennung der Bank wiedererkannt, unabhaengig davon, was aus der Buchung geworden ist. Das ist Absicht: Wer etwas weggeworfen hat, will es nicht beim naechsten Abruf zurueck. """ heute = date.today().isoformat() verbindung, link, _ = _verbunden(admin, fake_bank, umsaetze=[dict(UMSATZ, booking_date=heute)]) konto = _konto(admin, name=f"Aufraeumen {secrets.token_hex(3)}", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") admin.post(f"/api/bank/connections/{verbindung}/sync") eintrag = [e for e in admin.get("/api/inbox").json() if e["payee"] == "SPEEDEE RESTAURATIONS"][0] assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -1490 # So raeumt man auf: die Buchung loeschen. assert admin.delete(f"/api/transactions/{eintrag['transaction_id']}").status_code == 204 assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 0 # Und jetzt derselbe Umsatz noch einmal von der Bank. zweiter = admin.post(f"/api/bank/connections/{verbindung}/sync").json() assert zweiter["geholt"] >= 1, "der Umsatz kam wieder an" assert zweiter["neu"] == 0, "wurde aber nicht erneut angelegt" assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 0 def test_geloeschte_buchung_blockiert_auch_den_csv_import(admin, fake_bank): """Der andere Weg herein: Ein spaeterer Auszug darf sie auch nicht bringen.""" heute = date.today() verbindung, link, _ = _verbunden( admin, fake_bank, umsaetze=[dict(UMSATZ, booking_date=heute.isoformat())]) konto = _konto(admin, name=f"CSV-Schutz {secrets.token_hex(3)}", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") admin.post(f"/api/bank/connections/{verbindung}/sync") eintrag = [e for e in admin.get("/api/inbox").json() if e["payee"] == "SPEEDEE RESTAURATIONS"][0] admin.delete(f"/api/transactions/{eintrag['transaction_id']}") from app.database import SessionLocal from app.models import Account, User from app.services import csv_import db = SessionLocal() try: nutzer = db.query(User).order_by(User.id).first() acc = db.get(Account, konto["id"]) zeilen = csv_import.detect_rows(db, nutzer, acc, [{ "date": heute.isoformat(), "amount_cents": -1490, "payee": "SPEEDEE RESTAURATIONS ZUERICH", "purpose": None}]) finally: db.close() assert zeilen[0]["is_duplicate"] is True assert zeilen[0]["duplicate_reason"] == "inbox" def test_zwei_gleiche_betraege_kurz_hintereinander_kommen_beide_an(admin, fake_bank): """Revolut-Aufrundungen sind oft mehrfach derselbe Betrag in wenigen Tagen. Der Eingangs-Abgleich vergleicht nur Betrag und Datum (±3 Tage) – wenn er denselben Eintrag mehrfach verbraucht, faellt die zweite Zahlung weg und der Saldo weicht ab.""" _app(admin) heute = date.today() uid = f"acc-{secrets.token_hex(4)}" umsaetze = [ {"entry_reference": f"tx-{secrets.token_hex(4)}", "booking_date": (heute - timedelta(days=tage)).isoformat(), "transaction_amount": {"amount": "1.00", "currency": "CHF"}, "credit_debit_indicator": "DBIT", "creditor": {"name": "To CHF MB:8f33c0aa"}} for tage in (2, 1, 0) ] fake_bank(konten=[{"uid": uid, "currency": "CHF"}], umsaetze=umsaetze) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Rundung-{secrets.token_hex(3)}", "country": "CH"}).json()["state"] admin.post(f"/api/bank/callback?state={state}&code=guter-code") link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] konto = _konto(admin, name=f"Rundung {secrets.token_hex(3)}", currency="CHF") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") r = admin.post(f"/api/bank/links/{link['id']}/sync").json() assert r["geholt"] == 3 assert r["neu"] == 3, f"nur {r['neu']} von 3 angekommen" assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -300 def test_verschluckter_umsatz_laesst_sich_nachholen(admin, fake_bank): """Wurde etwas faelschlich als Dublette uebersprungen, entstand kein Eingangs-Eintrag – der naechste Abruf haette es also wieder geholt. Nur liegt es dann meist vor dem Fenster des letzten Abrufs.""" _app(admin) alt = (date.today() - timedelta(days=40)).isoformat() uid = f"acc-{secrets.token_hex(4)}" gegenstelle = fake_bank( konten=[{"uid": uid, "currency": "EUR"}], umsaetze=[dict(UMSATZ, entry_reference=f"tx-{secrets.token_hex(4)}", booking_date=alt)]) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Nachhol-{secrets.token_hex(3)}", "country": "DE"}).json()["state"] admin.post(f"/api/bank/callback?state={state}&code=guter-code") link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] konto = _konto(admin, name=f"Nachhol {secrets.token_hex(3)}", currency="EUR") # Eine vorhandene Buchung schiebt das Startdatum nach vorn. admin.post("/api/transactions", json={ "type": "expense", "date": date.today().isoformat(), "account_id": konto["id"], "amount_cents": 100, "payee": "schon da"}) admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") # Der normale Abruf greift nicht weit genug zurueck. assert admin.post(f"/api/bank/links/{link['id']}/sync").json()["neu"] == 0 assert gegenstelle.date_from > alt # Mit Zeitraum schon. r = admin.post(f"/api/bank/links/{link['id']}/sync", params={"tage": 90}).json() assert r["neu"] == 1, r assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -100 - 1490 assert admin.post(f"/api/bank/links/{link['id']}/sync", params={"tage": 400}).status_code == 400 def test_startdatum_liegt_nie_in_der_zukunft(admin, fake_bank): """Liegt die neueste Buchung auf heute, waere der Tag danach morgen – und die Schnittstelle lehnt das ab ("date_from can not be in the future"). Der Fehler faellt erst beim Umsatz-Abruf an, nachdem die Saldi schon gelesen sind: Es sieht dann aus, als kaemen nur Saldi und keine Umsaetze. """ _app(admin) heute = date.today() uid = f"acc-{secrets.token_hex(4)}" gegenstelle = fake_bank( konten=[{"uid": uid, "currency": "EUR"}], umsaetze=[dict(UMSATZ, entry_reference=f"tx-{secrets.token_hex(4)}", booking_date=heute.isoformat())]) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Zukunft-{secrets.token_hex(3)}", "country": "DE"}).json()["state"] admin.post(f"/api/bank/callback?state={state}&code=guter-code") link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] konto = _konto(admin, name=f"Zukunft {secrets.token_hex(3)}", currency="EUR") # Eine Buchung von HEUTE – der Tag danach ist morgen. admin.post("/api/transactions", json={ "type": "expense", "date": heute.isoformat(), "account_id": konto["id"], "amount_cents": 500, "payee": "von heute"}) admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") r = admin.post(f"/api/bank/links/{link['id']}/sync") assert r.status_code == 200, r.text assert gegenstelle.date_from <= heute.isoformat(), \ f"date_from liegt in der Zukunft: {gegenstelle.date_from}" assert r.json()["ab"] == heute.isoformat() def test_abruf_ueber_schon_eingelesene_monate_bucht_nicht_doppelt(admin, fake_bank): """'Aelteres nachholen' greift in Monate zurueck, die schon per CSV eingelesen sind. Der Schluessel des Duplikatschutzes enthaelt den Empfaengertext, und der lautet bei der Bank anders – ohne eigene Pruefung stuende die Zahlung zweimal da.""" _app(admin) vor_wochen = date.today() - timedelta(days=30) uid = f"acc-{secrets.token_hex(4)}" fake_bank( konten=[{"uid": uid, "currency": "CHF"}], umsaetze=[ {"entry_reference": f"tx-{secrets.token_hex(4)}", "booking_date": vor_wochen.isoformat(), "transaction_amount": {"amount": "35.00", "currency": "CHF"}, "credit_debit_indicator": "DBIT", "creditor": {"name": "Coop 6141"}}, # Und ein Umsatz, den es noch NICHT gibt – der muss ankommen. {"entry_reference": f"tx-{secrets.token_hex(4)}", "booking_date": vor_wochen.isoformat(), "transaction_amount": {"amount": "7.77", "currency": "CHF"}, "credit_debit_indicator": "DBIT", "creditor": {"name": "Neu und echt"}}, ]) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Ueberlapp-{secrets.token_hex(3)}", "country": "CH"}).json()["state"] admin.post(f"/api/bank/callback?state={state}&code=guter-code") link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] konto = _konto(admin, name=f"Ueberlapp {secrets.token_hex(3)}", currency="CHF") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") # So sieht der schon eingelesene Auszug aus – anderer Text, ein Tag daneben. admin.post("/api/transactions", json={ "type": "expense", "date": (vor_wochen + timedelta(days=1)).isoformat(), "account_id": konto["id"], "amount_cents": 3500, "payee": "COOP-6141 KILLWANGEN"}) r = admin.post(f"/api/bank/links/{link['id']}/sync", params={"tage": 90}).json() assert r["doppelt"] == 1, r assert r["neu"] == 1, "der echte Umsatz muss ankommen" # 35.00 einmal (aus der CSV) plus 7.77 neu – nicht 35.00 zweimal. assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -4277 def test_uebergangene_konten_werden_gezeigt_nicht_nur_gezaehlt(admin, fake_bank): """Ohne die Zahl der uebermittelten Konten laesst sich nicht sagen, ob ein fehlendes Konto gar nicht kam oder hier haengen blieb – und man sucht an der falschen Stelle.""" _app(admin) gut = f"acc-{secrets.token_hex(4)}" fake_bank(konten=[ {"uid": gut, "name": "Girokonto", "currency": "CHF"}, # So sieht ein Pocket aus, wenn die Kennung fehlt. {"name": "Sparen", "currency": "CHF", "account_id": {"other": {"identification": "OTHI 6bdb2e3e", "scheme_name": "OTHI"}}}, {"name": "Steuern", "currency": "CHF"}, ]) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Pockets-{secrets.token_hex(3)}", "country": "LT"}).json()["state"] r = admin.post(f"/api/bank/callback?state={state}&code=guter-code") assert r.status_code == 200, r.text daten = r.json() assert daten["konten"] == 1 assert daten["uebermittelt"] == 3, "die Bank hat drei geschickt" assert daten["ohne_kennung"] == 2 # Und zwar mitsamt dem, was ankam – sonst raet man, was ihnen fehlt. namen = {k.get("name") for k in daten["uebergangen"]} assert namen == {"Sparen", "Steuern"} def test_ein_konto_wird_nur_von_einem_bankkonto_gefuellt(admin, fake_bank): """Sonst holen zwei Zugaenge dieselben Umsaetze – mit verschiedenen Kennungen, also ohne dass der Duplikatschutz greift. Genau das passiert, wenn dieselbe Bank zweimal verbunden ist: Enable Banking fuehrt sie je Land getrennt und vergibt eigene Kennungen.""" _app(admin) konto = _konto(admin, name=f"Doppelt gefuellt {secrets.token_hex(3)}", currency="EUR") links = [] for land in ("DE", "LT"): uid = f"acc-{secrets.token_hex(4)}" fake_bank(konten=[{"uid": uid, "currency": "EUR"}]) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Revolut-{land}-{secrets.token_hex(3)}", "country": land}).json()["state"] admin.post(f"/api/bank/callback?state={state}&code=guter-code") links.append([v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0]) admin.post(f"/api/bank/links/{links[0]['id']}/account/{konto['id']}") admin.post(f"/api/bank/links/{links[1]['id']}/account/{konto['id']}") zugeordnet = [v for v in admin.get("/api/bank/links").json() if v["account_id"] == konto["id"]] assert len(zugeordnet) == 1, "zwei Bankkonten fuellen dasselbe Konto" assert zugeordnet[0]["id"] == links[1]["id"], "die neuere Zuordnung gewinnt" def test_zweiter_bankzugang_bucht_dieselbe_zahlung_nicht_noch_einmal(admin, fake_bank): """Auch wenn die Zuordnung von Hand geaendert wurde: Eine Zahlung, die ueber den einen Zugang schon gebucht ist, darf ueber den anderen nicht ein zweites Mal hereinkommen.""" _app(admin) heute = date.today().isoformat() konto = _konto(admin, name=f"Zwei Zugaenge {secrets.token_hex(3)}", currency="EUR") salden = [{"balance_type": "CLBD", "balance_amount": {"amount": "0.00", "currency": "EUR"}}] def verbinde(land): uid = f"acc-{secrets.token_hex(4)}" fake_bank(konten=[{"uid": uid, "currency": "EUR"}], salden=salden, umsaetze=[dict(UMSATZ, entry_reference=f"ref-{land}", booking_date=heute)]) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Zwei-{land}-{secrets.token_hex(3)}", "country": land}).json()["state"] admin.post(f"/api/bank/callback?state={state}&code=guter-code") return [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] eins, zwei = verbinde("DE"), verbinde("LT") admin.post(f"/api/bank/links/{eins['id']}/account/{konto['id']}") admin.post(f"/api/bank/links/{eins['id']}/sync") assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -1490 # Jetzt derselbe Umsatz ueber den zweiten Zugang – andere Kennung. admin.post(f"/api/bank/links/{zwei['id']}/account/{konto['id']}") r = admin.post(f"/api/bank/links/{zwei['id']}/sync").json() assert r["doppelt"] == 1, r assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -1490 def test_verfuegbarer_saldo_wird_nicht_als_abweichung_gemeldet(admin, fake_bank): """Ein verfuegbarer Saldo enthaelt Vorgemerktes, das Certain bewusst nicht bucht. Daraus eine Abweichung zu machen, waere eine Falschmeldung – und genau die kam staendig.""" _app(admin) uid = f"acc-{secrets.token_hex(4)}" fake_bank( konten=[{"uid": uid, "currency": "EUR"}], salden=[{"balance_type": "ITAV", "balance_amount": {"amount": "150.00", "currency": "EUR"}}]) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Verfuegbar-{secrets.token_hex(3)}", "country": "DE"}).json()["state"] verbindung = admin.post( f"/api/bank/callback?state={state}&code=guter-code").json()["connection_id"] link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] konto = _konto(admin, name=f"Verfuegbar {secrets.token_hex(3)}", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") admin.post(f"/api/bank/connections/{verbindung}/salden") aus_liste = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] assert aus_liste["balance_cents"] == 15000 assert aus_liste["balance_vergleichbar"] is False am_konto = admin.get(f"/api/accounts/{konto['id']}").json() assert am_konto["bank_balance_cents"] == 15000 assert am_konto["bank_balance_vergleichbar"] is False # Und ein gebuchter Saldo ist es sehr wohl. fake_bank(konten=[{"uid": uid, "currency": "EUR"}], salden=[{"balance_type": "CLBD", "balance_amount": {"amount": "150.00", "currency": "EUR"}}]) admin.post(f"/api/bank/connections/{verbindung}/salden") assert admin.get(f"/api/accounts/{konto['id']}").json()["bank_balance_vergleichbar"] is True def test_saldo_kommt_erst_nach_den_umsaetzen(admin, fake_bank, monkeypatch): """Vorher geholt waere er neuer als die Umsaetze: Bricht deren Abruf ab, stuende ein aktueller Saldo neben altem Bestand – und die Anzeige meldete eine Abweichung, die keine ist.""" from app.services import bank_api _app(admin) uid = f"acc-{secrets.token_hex(4)}" fake_bank(konten=[{"uid": uid, "currency": "EUR"}], salden=[{"balance_type": "CLBD", "balance_amount": {"amount": "99.00", "currency": "EUR"}}]) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Reihenfolge-{secrets.token_hex(3)}", "country": "DE"}).json()["state"] admin.post(f"/api/bank/callback?state={state}&code=guter-code") link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] konto = _konto(admin, name=f"Reihenfolge {secrets.token_hex(3)}", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") # Der Umsatz-Abruf scheitert. echte = bank_api.BankClient.umsaetze monkeypatch.setattr(bank_api.BankClient, "umsaetze", lambda *a, **kw: (_ for _ in ()).throw( bank_api.BankFehler("Schnittstelle streikt"))) assert admin.post(f"/api/bank/links/{link['id']}/sync").status_code == 400 monkeypatch.setattr(bank_api.BankClient, "umsaetze", echte) # Dann darf auch kein Saldo dastehen, der neuer waere als der Bestand. aus_liste = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] assert aus_liste["balance_cents"] is None, "Saldo ohne Umsaetze geschrieben" def test_zu_knapper_takt_wird_gebremst(): """Eine bestehende Einstellung von 15 Minuten liefe sonst weiter gegen die Grenze der Bank – und jeder Abruf scheiterte mit 'Too many requests'.""" from datetime import datetime, timezone from app.models import BankConnection from app.services import bank_sync jetzt = datetime.now(timezone.utc) knapp = BankConnection(owner_id=1, aspsp_name="N26", aspsp_country="DE", poll_minutes=15, last_sync_at=jetzt - timedelta(minutes=20)) assert bank_sync.faellig(knapp, jetzt) is False, "15 Minuten duerfen nicht greifen" # Nach sechs Stunden schon. knapp.last_sync_at = jetzt - timedelta(hours=7) assert bank_sync.faellig(knapp, jetzt) is True def test_umsatz_unter_anderer_kennung_kommt_nicht_zweimal(admin, fake_bank): """Liefert die Bank denselben Umsatz spaeter unter einer anderen Kennung – etwa weil aus vorgemerkt gebucht wurde –, waeren es fuer Certain zwei. Geprueft wird deshalb gegen ALLE Kennungen, die er mitbringt.""" _app(admin) heute = date.today().isoformat() uid = f"acc-{secrets.token_hex(4)}" stabil = f"tx-{secrets.token_hex(5)}" salden = [{"balance_type": "CLBD", "balance_amount": {"amount": "0.00", "currency": "EUR"}}] def umsatz(referenz): return {"entry_reference": referenz, "transaction_id": stabil, "booking_date": heute, "transaction_amount": {"amount": "288.00", "currency": "EUR"}, "credit_debit_indicator": "DBIT", "creditor": {"name": "Noemi Koenig"}} fake_bank(konten=[{"uid": uid, "currency": "EUR"}], salden=salden, umsaetze=[umsatz("ref-vorgemerkt")]) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Kennung-{secrets.token_hex(3)}", "country": "DE"}).json()["state"] admin.post(f"/api/bank/callback?state={state}&code=guter-code") link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] konto = _konto(admin, name=f"Kennung {secrets.token_hex(3)}", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") assert admin.post(f"/api/bank/links/{link['id']}/sync").json()["neu"] == 1 assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -28800 # Zweiter Abruf: andere entry_reference, dieselbe transaction_id. fake_bank(konten=[{"uid": uid, "currency": "EUR"}], salden=salden, umsaetze=[umsatz("ref-gebucht")]) zweiter = admin.post(f"/api/bank/links/{link['id']}/sync").json() assert zweiter["neu"] == 0, "unter neuer Referenz noch einmal gebucht" assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -28800 # --- Die Kennung steht an der Buchung --------------------------------------- def _eingaenge_loeschen(uid: str) -> int: """Alle Eingangs-Eintraege dieses Bankkontos entfernen. Stellt nach, was im Betrieb wirklich passiert: Der Eintrag ist eine Zwischenstation und faellt weg – geloeschtes Konto, aelteres Backup, neu verbundene Bank mit neuer Kennung. Die Buchung bleibt. """ from app.database import SessionLocal from app.models import InboxItem with SessionLocal() as s: eintraege = s.query(InboxItem).filter( InboxItem.external_id.like(f"{uid}:%")).all() for e in eintraege: s.delete(e) s.commit() return len(eintraege) def test_kennung_der_bank_steht_an_der_buchung(admin, fake_bank): """Nicht nur am Eingangs-Eintrag. Nur an der Buchung ueberlebt sie alles, was mit dem Eingang passieren kann.""" from app.database import SessionLocal from app.models import Transaction _app(admin) uid = f"acc-{secrets.token_hex(4)}" tx = f"tx-{secrets.token_hex(5)}" fake_bank(konten=[{"uid": uid, "currency": "EUR"}], salden=[{"balance_type": "CLBD", "balance_amount": {"amount": "0.00", "currency": "EUR"}}], umsaetze=[{"transaction_id": tx, "booking_date": date.today().isoformat(), "transaction_amount": {"amount": "31.40", "currency": "EUR"}, "credit_debit_indicator": "DBIT", "creditor": {"name": "Baecker Ohnesorg"}}]) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Steht-{secrets.token_hex(3)}", "country": "DE"}).json()["state"] admin.post(f"/api/bank/callback?state={state}&code=guter-code") link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] konto = _konto(admin, name=f"Steht {secrets.token_hex(3)}", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") assert admin.post(f"/api/bank/links/{link['id']}/sync").json()["neu"] == 1 with SessionLocal() as s: txn = s.query(Transaction).filter(Transaction.account_id == konto["id"]).one() assert txn.source_uid == f"bank:{uid}:{tx}" def test_ohne_eingangs_eintrag_kommt_der_umsatz_nicht_zurueck(admin, fake_bank): """Faellt der Eingangs-Eintrag weg, blieb bisher nur noch die unscharfe Aehnlichkeit. Jetzt erkennt die Kennung an der Buchung den Umsatz exakt wieder – daran, dass ``doppelt`` bei 0 bleibt, sieht man den Unterschied.""" _app(admin) heute = date.today().isoformat() uid = f"acc-{secrets.token_hex(4)}" salden = [{"balance_type": "CLBD", "balance_amount": {"amount": "0.00", "currency": "EUR"}}] umsatz = {"transaction_id": f"tx-{secrets.token_hex(5)}", "booking_date": heute, "transaction_amount": {"amount": "76.20", "currency": "EUR"}, "credit_debit_indicator": "DBIT", "creditor": {"name": "Werkzeug Nadolny"}} fake_bank(konten=[{"uid": uid, "currency": "EUR"}], salden=salden, umsaetze=[umsatz]) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Ohne-{secrets.token_hex(3)}", "country": "DE"}).json()["state"] admin.post(f"/api/bank/callback?state={state}&code=guter-code") link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] konto = _konto(admin, name=f"Ohne {secrets.token_hex(3)}", currency="EUR") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") assert admin.post(f"/api/bank/links/{link['id']}/sync").json()["neu"] == 1 assert _eingaenge_loeschen(uid) == 1 fake_bank(konten=[{"uid": uid, "currency": "EUR"}], salden=salden, umsaetze=[umsatz]) zweiter = admin.post(f"/api/bank/links/{link['id']}/sync").json() assert zweiter["neu"] == 0, "ohne Eingangs-Eintrag noch einmal gebucht" assert zweiter["doppelt"] == 0, "nur noch unscharf erkannt statt an der Kennung" assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -7620 def test_als_uebertrag_gebuchter_umsatz_kommt_nicht_zurueck(admin, fake_bank): """Der Fall, in dem sonst NICHTS mehr greift. Ein Uebertrag ist von der Aehnlichkeitspruefung ausgenommen (er hat kein Gegenueber in der Ausgaben-/Einnahmen-Liste), und ohne Eingangs-Eintrag gibt es auch dort nichts zu finden. Bleibt allein die Kennung. """ _app(admin) heute = date.today().isoformat() uid = f"acc-{secrets.token_hex(4)}" salden = [{"balance_type": "CLBD", "balance_amount": {"amount": "0.00", "currency": "EUR"}}] umsatz = {"transaction_id": f"tx-{secrets.token_hex(5)}", "booking_date": heute, "transaction_amount": {"amount": "500.00", "currency": "EUR"}, "credit_debit_indicator": "DBIT", "creditor": {"name": "Eigenes Sparkonto"}} fake_bank(konten=[{"uid": uid, "currency": "EUR"}], salden=salden, umsaetze=[umsatz]) state = admin.post("/api/bank/connect", json={"aspsp_name": f"Uebertrag-{secrets.token_hex(3)}", "country": "DE"}).json()["state"] admin.post(f"/api/bank/callback?state={state}&code=guter-code") link = [v for v in admin.get("/api/bank/links").json() if v["remote_uid"] == uid][0] konto = _konto(admin, name=f"Uebertrag {secrets.token_hex(3)}", currency="EUR") sparkonto = _konto(admin, name=f"Sparen {secrets.token_hex(3)}", currency="EUR", type="sparkonto") admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}") assert admin.post(f"/api/bank/links/{link['id']}/sync").json()["neu"] == 1 gebucht = admin.get(f"/api/transactions?account_id={konto['id']}").json()[0] r = admin.post(f"/api/transactions/{gebucht['id']}/transfer", json={"to_account_id": sparkonto["id"], "trotzdem": True}) assert r.status_code == 200, r.text assert _eingaenge_loeschen(uid) >= 1 fake_bank(konten=[{"uid": uid, "currency": "EUR"}], salden=salden, umsaetze=[umsatz]) zweiter = admin.post(f"/api/bank/links/{link['id']}/sync").json() assert zweiter["neu"] == 0, "als Uebertrag gebucht und trotzdem neu geholt" assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -50000