GuidoandClaude Opus 5
277fe8b0e2
Drei verschiedene Faelle sahen alle gleich aus: Die Bank schickte nichts, sie schickte Eintraege ohne Kennung (die still verworfen wurden), oder alle Konten waren schon bekannt – Letzteres meldete faelschlich "0 Konten", obwohl alles in Ordnung war. Die Antwort unterscheidet jetzt uebermittelt / neu / bekannt / ohne_kennung. Die Rueckkehr-Seite zeigt bei null nutzbaren Konten den passenden Hinweis: Im eingeschraenkten Betrieb gibt eine Bank nur heraus, was zuvor im Control Panel unter "Activate by linking accounts" verknuepft wurde. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
522 lines
20 KiB
Python
522 lines
20 KiB
Python
"""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"]
|
||
|
||
|
||
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
|