GuidoandClaude Opus 5
8ceee5c5b1
Ein Uebertrag legte auf dem Zielkonto nichts an. Der Duplikatschutz konnte die Gegenseite deshalb nicht sehen – er vergleicht import_hash, den eine Uebertrags-Buchung nicht hat, und Eingangs-Eintraege, die es dort nicht gab. Die schon vorliegende Gegenseite liess sich ein zweites Mal buchen, die spaeter von der Bank gelieferte kam ungeprueft an: das Geld zweimal auf dem Zielkonto. Jetzt entsteht auf dem Zielkonto ein Eingangs-Eintrag, der auf DIESELBE Buchung zeigt. Liegt dort schon ein passender offener Eintrag, wird dieser verbraucht statt einen zweiten anzulegen – ueber denselben Abgleich wie beim Import, nicht ueber eine zweite Vergleichslogik. Damit greift auch der vorhandene Schutz gegen die spaeter eintreffende Gegenseite, ohne dass etwas Neues dafuer gebaut werden musste. Die Uebertrags-Logik liegt dafuer in einem eigenen Dienst. Zwei Wege fuehren hinein – ein offener Vorschlag und eine schon gebuchte Buchung – und beide brauchen dieselben vier Regeln. Der PATCH-Weg des Buchungs-Routers taete es nicht: Er dreht die Richtung bei einer Einnahme nicht um und legt keine Gegenseite an. Die Sperre richtet sich nicht mehr nach der Waehrung, sondern danach, ob der Betrag der Gegenseite BEKANNT oder geschaetzt ist. Bei gleicher Waehrung ist er exakt, die Gegenseite wird sicher wiedererkannt – dann geht ein Uebertrag auf ein angebundenes Konto ohne Nachfrage durch. Ueber Waehrungen hinweg bleibt die Warnung, weil der EZB-Kurs nicht der Kurs der Bank ist. Und Verwerfen raeumt beide Seiten: Bei einem Uebertrag zeigen zwei Eintraege auf dieselbe Buchung, und der Fremdschluessel haette die andere als 'gebucht' ohne Buchung zurueckgelassen. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
305 lines
14 KiB
Python
305 lines
14 KiB
Python
"""Eingang: einen Vorschlag als Kontoübertrag buchen.
|
||
|
||
Ohne diesen Weg bleibt nur, eine Umbuchung als Ausgabe zu kategorisieren – sie
|
||
zählt dann als Ausgabe, obwohl kein Geld weg ist.
|
||
"""
|
||
|
||
import secrets
|
||
|
||
|
||
def _konto(admin, **kw):
|
||
body = {"name": "Girokonto", "type": "girokonto", "currency": "CHF",
|
||
"opening_balance_cents": 0}
|
||
body.update(kw)
|
||
r = admin.post("/api/accounts", json=body)
|
||
assert r.status_code == 201, r.text
|
||
return r.json()
|
||
|
||
|
||
def _kategorie(admin):
|
||
"""Irgendeine Ausgabe-Kategorie aus den Vorgaben."""
|
||
for gruppe in admin.get("/api/categories").json():
|
||
if gruppe["kind"] == "expense" and gruppe["categories"]:
|
||
return gruppe["categories"][0]["id"]
|
||
raise AssertionError("keine Ausgabe-Kategorie vorhanden")
|
||
|
||
|
||
def _eingang(admin, konto, betrag_cents, datum="2026-08-20", payee="Umbuchung"):
|
||
"""Ein OFFENER Eintrag – so wie ihn Mail und Bankabruf hinterlassen.
|
||
|
||
Nicht ueber die Schnellerfassung: Die bucht sofort, und dann gaebe es
|
||
nichts mehr als Uebertrag zu buchen.
|
||
"""
|
||
from datetime import date
|
||
|
||
from app.database import SessionLocal
|
||
from app.models import InboxItem, InboxSource, InboxStatus, User
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
besitzer = db.query(User).order_by(User.id).first()
|
||
eintrag = InboxItem(
|
||
owner_id=besitzer.id,
|
||
source=InboxSource.mail,
|
||
status=InboxStatus.pending,
|
||
account_id=konto["id"],
|
||
date=date.fromisoformat(datum),
|
||
amount_cents=betrag_cents,
|
||
currency=konto["currency"],
|
||
payee=payee,
|
||
)
|
||
db.add(eintrag)
|
||
db.commit()
|
||
return {"id": eintrag.id}
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def test_eingang_laesst_sich_als_uebertrag_buchen(admin):
|
||
von = _konto(admin, name=f"Von {secrets.token_hex(3)}", currency="CHF")
|
||
nach = _konto(admin, name=f"Nach {secrets.token_hex(3)}", currency="CHF")
|
||
eintrag = _eingang(admin, von, -50000)
|
||
|
||
r = admin.post(f"/api/inbox/{eintrag['id']}/transfer",
|
||
json={"to_account_id": nach["id"]})
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["status"] == "booked"
|
||
assert r.json()["category_id"] is None
|
||
|
||
txn = admin.get(f"/api/transactions/{r.json()['transaction_id']}").json()
|
||
assert txn["type"] == "transfer"
|
||
assert txn["account_id"] == von["id"]
|
||
assert txn["to_account_id"] == nach["id"]
|
||
assert txn["amount_cents"] == 50000
|
||
|
||
# Das Geld ist verschoben, nicht ausgegeben.
|
||
assert admin.get(f"/api/accounts/{von['id']}").json()["balance_cents"] == -50000
|
||
assert admin.get(f"/api/accounts/{nach['id']}").json()["balance_cents"] == 50000
|
||
|
||
|
||
def test_zugang_dreht_die_richtung_um(admin):
|
||
"""Ein positiver Betrag heisst: Auf DIESEM Konto kam das Geld an. Die
|
||
Buchung muss vom Gegenkonto hierher gehen, nicht umgekehrt."""
|
||
hier = _konto(admin, name=f"Hier {secrets.token_hex(3)}", currency="CHF")
|
||
dort = _konto(admin, name=f"Dort {secrets.token_hex(3)}", currency="CHF")
|
||
eintrag = _eingang(admin, hier, 25000)
|
||
|
||
r = admin.post(f"/api/inbox/{eintrag['id']}/transfer", json={"to_account_id": dort["id"]})
|
||
assert r.status_code == 200, r.text
|
||
txn = admin.get(f"/api/transactions/{r.json()['transaction_id']}").json()
|
||
assert txn["account_id"] == dort["id"], "der Abgang gehört zum Gegenkonto"
|
||
assert txn["to_account_id"] == hier["id"]
|
||
assert admin.get(f"/api/accounts/{hier['id']}").json()["balance_cents"] == 25000
|
||
assert admin.get(f"/api/accounts/{dort['id']}").json()["balance_cents"] == -25000
|
||
|
||
|
||
def test_uebertrag_zaehlt_in_keiner_auswertung(admin):
|
||
"""Der eigentliche Zweck: Ein Übertrag ist keine Ausgabe."""
|
||
von = _konto(admin, name=f"Budget-Von {secrets.token_hex(3)}", currency="CHF")
|
||
nach = _konto(admin, name=f"Budget-Nach {secrets.token_hex(3)}", currency="CHF")
|
||
eintrag = _eingang(admin, von, -30000, datum="2026-08-05")
|
||
admin.post(f"/api/inbox/{eintrag['id']}/transfer", json={"to_account_id": nach["id"]})
|
||
|
||
v = admin.get("/api/budget/2026-08").json()
|
||
gesamt = sum(z.get("actual_cents") or 0 for z in v.get("rows", []))
|
||
assert gesamt == 0, "ein Übertrag darf im Budget nicht auftauchen"
|
||
|
||
|
||
def _binde_an(konto):
|
||
"""Konto so markieren, als haenge es an einer Bank."""
|
||
from app.database import SessionLocal
|
||
from app.models import BankAccountLink, BankApp, BankConnection, User
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
besitzer = db.query(User).order_by(User.id).first()
|
||
app = BankApp(owner_id=besitzer.id, application_id="app-x",
|
||
private_key_enc="x", base_url="https://x.example")
|
||
db.add(app)
|
||
verbindung = BankConnection(owner_id=besitzer.id, aspsp_name="Zielbank",
|
||
aspsp_country="CH")
|
||
db.add(verbindung)
|
||
db.flush()
|
||
db.add(BankAccountLink(
|
||
owner_id=besitzer.id, connection_id=verbindung.id,
|
||
remote_uid=f"acc-{secrets.token_hex(4)}", currency=konto["currency"],
|
||
account_id=konto["id"]))
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def test_angebundenes_ziel_gleicher_waehrung_geht_durch(admin):
|
||
"""Bei gleicher Waehrung ist der Betrag exakt – die Gegenseite wird beim
|
||
Bankabruf sicher als dieselbe erkannt. Dann muss man nicht fragen."""
|
||
ziel = _konto(admin, name=f"Angebunden {secrets.token_hex(3)}", currency="CHF")
|
||
_binde_an(ziel)
|
||
quelle = _konto(admin, name=f"Quelle {secrets.token_hex(3)}", currency="CHF")
|
||
eintrag = _eingang(admin, quelle, -10000)
|
||
|
||
r = admin.post(f"/api/inbox/{eintrag['id']}/transfer", json={"to_account_id": ziel["id"]})
|
||
assert r.status_code == 200, r.text
|
||
assert admin.get(f"/api/accounts/{ziel['id']}").json()["balance_cents"] == 10000
|
||
|
||
|
||
def test_angebundenes_ziel_anderer_waehrung_fragt_nach(admin):
|
||
"""Ueber Waehrungen hinweg rechnet Certain mit dem EZB-Kurs, die Bank mit
|
||
ihrem eigenen. Der geschaetzte Betrag laesst sich beim Abruf nicht
|
||
wiedererkennen – dann kaeme der Zugang ein zweites Mal."""
|
||
ziel = _konto(admin, name=f"Angebunden EUR {secrets.token_hex(3)}", currency="EUR")
|
||
_binde_an(ziel)
|
||
quelle = _konto(admin, name=f"Quelle CHF {secrets.token_hex(3)}", currency="CHF")
|
||
eintrag = _eingang(admin, quelle, -10000)
|
||
|
||
r = admin.post(f"/api/inbox/{eintrag['id']}/transfer", json={"to_account_id": ziel["id"]})
|
||
assert r.status_code == 409, r.text
|
||
assert "EZB" in r.json()["detail"]
|
||
assert admin.get(f"/api/accounts/{ziel['id']}").json()["balance_cents"] == 0
|
||
|
||
# Wer es besser weiss, kommt trotzdem durch.
|
||
r = admin.post(f"/api/inbox/{eintrag['id']}/transfer",
|
||
json={"to_account_id": ziel["id"], "trotzdem": True})
|
||
assert r.status_code == 200, r.text
|
||
assert admin.get(f"/api/accounts/{ziel['id']}").json()["balance_cents"] > 0
|
||
|
||
|
||
def test_uebertrag_nennt_das_gegenkonto_und_nimmt_keine_kategorie(admin):
|
||
"""Ohne das Gegenkonto kann die Oberfläche einen Übertrag nicht von einer
|
||
Ausgabe unterscheiden – und fragt weiter nach einer Kategorie."""
|
||
von = _konto(admin, name=f"Nenn-Von {secrets.token_hex(3)}", currency="CHF")
|
||
nach = _konto(admin, name=f"Nenn-Nach {secrets.token_hex(3)}", currency="CHF")
|
||
eintrag = _eingang(admin, von, -8000)
|
||
|
||
r = admin.post(f"/api/inbox/{eintrag['id']}/transfer", json={"to_account_id": nach["id"]})
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["transfer_account_id"] == nach["id"]
|
||
|
||
# Auch in der Liste, und aus Sicht des Eintrags das jeweils ANDERE Konto.
|
||
aus_liste = [e for e in admin.get("/api/inbox").json() if e["id"] == eintrag["id"]][0]
|
||
assert aus_liste["transfer_account_id"] == nach["id"]
|
||
|
||
r = admin.post(f"/api/inbox/{eintrag['id']}/category/{_kategorie(admin)}")
|
||
assert r.status_code == 400, r.text
|
||
assert "Kategorie" in r.json()["detail"]
|
||
|
||
# Und die Buchung hat weiterhin keine.
|
||
txn = admin.get(f"/api/transactions/{aus_liste['transaction_id']}").json()
|
||
assert txn["category_id"] is None
|
||
|
||
|
||
def test_normale_buchung_nimmt_weiterhin_eine_kategorie(admin):
|
||
"""Die Sperre darf nur Überträge treffen."""
|
||
konto = _konto(admin, name=f"Normal {secrets.token_hex(3)}", currency="CHF")
|
||
# Eigener Betrag: Die Testdatenbank lebt ueber den ganzen Lauf, und ein
|
||
# runder Betrag kollidiert mit dem, wonach ein anderer Test sucht.
|
||
eintrag = _eingang(admin, konto, -(700000 + secrets.randbelow(90000)))
|
||
admin.post(f"/api/inbox/{eintrag['id']}/book")
|
||
|
||
kat = _kategorie(admin)
|
||
r = admin.post(f"/api/inbox/{eintrag['id']}/category/{kat}")
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["category_id"] == kat
|
||
assert r.json()["transfer_account_id"] is None
|
||
txn = admin.get(f"/api/transactions/{r.json()['transaction_id']}").json()
|
||
assert txn["category_id"] == kat, "die Kategorie muss auch an der Buchung stehen"
|
||
|
||
|
||
def test_gegenseite_wird_verbraucht_statt_gespiegelt(admin):
|
||
"""Liegt die Gegenseite schon im Eingang, wuerde sie sonst ein zweites Mal
|
||
gebucht – das Geld staende zweimal auf dem Zielkonto."""
|
||
von = _konto(admin, name=f"Paar-Von {secrets.token_hex(3)}", currency="CHF")
|
||
nach = _konto(admin, name=f"Paar-Nach {secrets.token_hex(3)}", currency="CHF")
|
||
betrag = 40000 + secrets.randbelow(9000)
|
||
ab = _eingang(admin, von, -betrag, datum="2026-08-10")
|
||
an = _eingang(admin, nach, betrag, datum="2026-08-11") # einen Tag spaeter
|
||
|
||
r = admin.post(f"/api/inbox/{ab['id']}/transfer", json={"to_account_id": nach["id"]})
|
||
assert r.status_code == 200, r.text
|
||
txn_id = r.json()["transaction_id"]
|
||
|
||
eintraege = {e["id"]: e for e in admin.get("/api/inbox").json()}
|
||
assert eintraege[an["id"]]["status"] == "booked", "die Gegenseite blieb offen"
|
||
assert eintraege[an["id"]]["transaction_id"] == txn_id, "muss dieselbe Buchung sein"
|
||
|
||
# Genau eine Bewegung, kein doppeltes Geld.
|
||
assert admin.get(f"/api/accounts/{von['id']}").json()["balance_cents"] == -betrag
|
||
assert admin.get(f"/api/accounts/{nach['id']}").json()["balance_cents"] == betrag
|
||
|
||
|
||
def test_ohne_gegenseite_entsteht_genau_eine(admin):
|
||
von = _konto(admin, name=f"Spiegel-Von {secrets.token_hex(3)}", currency="CHF")
|
||
nach = _konto(admin, name=f"Spiegel-Nach {secrets.token_hex(3)}", currency="CHF")
|
||
betrag = 50000 + secrets.randbelow(9000)
|
||
eintrag = _eingang(admin, von, -betrag)
|
||
|
||
r = admin.post(f"/api/inbox/{eintrag['id']}/transfer", json={"to_account_id": nach["id"]})
|
||
txn_id = r.json()["transaction_id"]
|
||
|
||
gegen = [e for e in admin.get("/api/inbox").json()
|
||
if e["transaction_id"] == txn_id and e["id"] != eintrag["id"]]
|
||
assert len(gegen) == 1
|
||
assert gegen[0]["source"] == "transfer"
|
||
assert gegen[0]["account_id"] == nach["id"]
|
||
assert gegen[0]["amount_cents"] == betrag
|
||
assert gegen[0]["transfer_account_id"] == von["id"], "aus ihrer Sicht das andere Konto"
|
||
|
||
|
||
def test_verwerfen_raeumt_beide_seiten(admin):
|
||
"""Sonst bliebe die andere Seite als „gebucht" ohne Buchung stehen."""
|
||
von = _konto(admin, name=f"Weg-Von {secrets.token_hex(3)}", currency="CHF")
|
||
nach = _konto(admin, name=f"Weg-Nach {secrets.token_hex(3)}", currency="CHF")
|
||
betrag = 60000 + secrets.randbelow(9000)
|
||
eintrag = _eingang(admin, von, -betrag)
|
||
r = admin.post(f"/api/inbox/{eintrag['id']}/transfer", json={"to_account_id": nach["id"]})
|
||
txn_id = r.json()["transaction_id"]
|
||
|
||
assert admin.delete(f"/api/inbox/{eintrag['id']}").status_code == 204
|
||
|
||
offen = admin.get("/api/inbox").json()
|
||
assert not [e for e in offen if e["transaction_id"] == txn_id], \
|
||
"ein Eintrag zeigt noch auf die geloeschte Buchung"
|
||
assert admin.get(f"/api/accounts/{von['id']}").json()["balance_cents"] == 0
|
||
assert admin.get(f"/api/accounts/{nach['id']}").json()["balance_cents"] == 0
|
||
|
||
|
||
def test_gebuchte_ausgabe_laesst_sich_nachtraeglich_zum_uebertrag_machen(admin):
|
||
"""Ohne diesen Weg bliebe eine per CSV eingelesene Umbuchung fuer immer
|
||
eine Ausgabe."""
|
||
von = _konto(admin, name=f"Nachtraeglich {secrets.token_hex(3)}", currency="CHF")
|
||
nach = _konto(admin, name=f"Nachziel {secrets.token_hex(3)}", currency="CHF")
|
||
betrag = 80000 + secrets.randbelow(9000)
|
||
txn = admin.post("/api/transactions", json={
|
||
"type": "expense", "date": "2026-08-12", "account_id": von["id"],
|
||
"amount_cents": betrag, "payee": "Umbuchung"}).json()
|
||
|
||
r = admin.post(f"/api/transactions/{txn['id']}/transfer",
|
||
json={"to_account_id": nach["id"]})
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["type"] == "transfer"
|
||
assert r.json()["category_id"] is None
|
||
assert admin.get(f"/api/accounts/{von['id']}").json()["balance_cents"] == -betrag
|
||
assert admin.get(f"/api/accounts/{nach['id']}").json()["balance_cents"] == betrag
|
||
|
||
# Und die Gegenseite ist da, damit die Bank sie nicht noch einmal liefert.
|
||
gegen = [e for e in admin.get("/api/inbox").json() if e["transaction_id"] == txn["id"]]
|
||
assert len(gegen) == 1 and gegen[0]["account_id"] == nach["id"]
|
||
|
||
|
||
def test_nachtraeglicher_uebertrag_dreht_eine_einnahme_um(admin):
|
||
"""Der PATCH-Weg des Buchungs-Routers taete das nicht – die Buchung liefe
|
||
falsch herum und beide Salden waeren verkehrt."""
|
||
hier = _konto(admin, name=f"Ein-Hier {secrets.token_hex(3)}", currency="CHF")
|
||
dort = _konto(admin, name=f"Ein-Dort {secrets.token_hex(3)}", currency="CHF")
|
||
betrag = 90000 + secrets.randbelow(9000)
|
||
txn = admin.post("/api/transactions", json={
|
||
"type": "income", "date": "2026-08-13", "account_id": hier["id"],
|
||
"amount_cents": betrag}).json()
|
||
|
||
r = admin.post(f"/api/transactions/{txn['id']}/transfer",
|
||
json={"to_account_id": dort["id"]})
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["account_id"] == dort["id"], "der Abgang gehoert zum Gegenkonto"
|
||
assert r.json()["to_account_id"] == hier["id"]
|
||
assert admin.get(f"/api/accounts/{hier['id']}").json()["balance_cents"] == betrag
|
||
assert admin.get(f"/api/accounts/{dort['id']}").json()["balance_cents"] == -betrag
|