Files
certain/backend/tests/test_bank.py
T
GuidoandClaude Opus 5 beb1bb3419 fix: Bank-Suche verlor die besten Treffer beim Kuerzen
Der Fehler, den der Nutzer gefunden hat: Die Suche nach "n" lieferte
"genau 200" Treffer und darunter nie N26. Ursache war die Reihenfolge –
alphabetisch sortiert und ERST DANN bei 200 gekuerzt. "N26" liegt hinter
"Aachener Bank", fiel also hinter zweihundert Banken heraus, die irgendwo
ein n im Namen tragen. Die Bank war die ganze Zeit da.

Jetzt wird zuerst bewertet (genauer Name > Namensanfang > Wortanfang >
Teiltreffer > nur BIC/Gruppe) und erst danach gekuerzt. Ausserdem wird die
ECHTE Trefferzahl gemeldet statt der gekuerzten – "genau 200" sah aus wie ein
Suchergebnis, war aber die Obergrenze.

Ein Regressionstest stellt den Fall nach: 300 Banken mit "n" im Namen,
alphabetisch vor N26. Ohne den Fix faellt er um.

Nebenbei geprueft und ausgeschlossen: /aspsps blaettert nicht, die Liste ist
also vollstaendig; Sandbox und alphabetische Kuerzung waren es auch nicht.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-26 22:10:41 +02:00

483 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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
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):
self.umsaetze = umsaetze or []
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.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": [{"balance_amount": {"amount": "100.00",
"currency": "EUR"}}]}
if "/transactions" in pfad:
# 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": self.umsaetze[:1], "continuation_key": "s2"}
rest = self.umsaetze[1:] if self.seiten > 1 else self.umsaetze
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)
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()
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_landet_als_vorschlag_im_eingang(admin, fake_bank):
"""Nichts wird gebucht – die Bank liefert Vorschläge wie Mail und Kurzbefehl."""
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
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"] == "pending"
assert treffer[0]["transaction_id"] is None
# Und der Saldo hat sich nicht bewegt.
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 0
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"]