Files
certain/backend/tests/test_inbox.py
T
GuidoandClaude Opus 5 6cd1c100c0 fix: Uebertrag pruefen, ob die Gegenseite schon gebucht ist
Ein Franken-in-Euro-Tausch stand zweimal da: die CHF-Seite als Uebertrag nach
Revolut Euro, die EUR-Seite weiterhin als eigene Einnahme. Revolut Euro war
damit doppelt gutgeschrieben. Drei Fehler auf einmal.

Erstens prueft "zum Uebertrag machen" jetzt, ob die Buchung bereits mit ihrer
Gegenseite VERBUNDEN ist. Dann sind beide Seiten gebucht und zaehlen schon in
keiner Auswertung; ein Uebertrag bewegt das Geld auf dem Zielkonto ein zweites
Mal. Das laesst sich auch nicht mit "trotzdem" wegdruecken – es ist nicht
Ansichtssache.

Zweitens wird auf dem Zielkonto nach einer passenden gebuchten Gegenseite
gesehen, auch ohne Verknuepfung: gleiche Tage, passender Betrag, ueber
Waehrungen hinweg mit Spielraum. Findet sich eine, kommt die Warnung samt dem
richtigen Weg – die beiden verbinden statt einen Uebertrag anzulegen. Der
Treffer ist unscharf, deshalb hier mit "trotzdem" uebergehbar.

Drittens der beobachtete Punkt: Der nachtraegliche Weg reichte den
Eingangs-Eintrag nicht durch und rechnete deshalb IMMER mit dem EZB-Kurs, auch
wenn die Bank ihren eigenen genannt hatte. Jetzt wird er nachgeschlagen, und
der Kurs der Bank landet an der Buchung.

Und Loeschen raeumt die Eingangs-Eintraege mit. Der Fremdschluessel setzt sonst
nur ihre Verweisspalte auf NULL, und sie blieben als "gebucht" ohne Buchung
stehen – bei einem Uebertrag sind das zwei.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-27 11:22:13 +02:00

436 lines
20 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.
"""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
def test_bank_saldo_laesst_sich_uebernehmen(admin):
"""Die Differenz gehoert in den Anfangsbestand: Sie steht fuer Bewegungen,
die Certain nie gesehen hat. Eine Korrekturbuchung liefe durch jede
Auswertung und bliebe fuer immer in der Liste."""
from app.database import SessionLocal
from app.models import BankAccountLink
konto = _konto(admin, name=f"Saldo {secrets.token_hex(3)}", currency="CHF")
_binde_an(konto)
admin.post("/api/transactions", json={
"type": "expense", "date": "2026-08-14", "account_id": konto["id"],
"amount_cents": 5000, "payee": "Etwas"})
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == -5000
db = SessionLocal()
try:
v = db.query(BankAccountLink).filter(
BankAccountLink.account_id == konto["id"]).first()
v.balance_cents = 120000 # die Bank sagt: 1'200.00
db.commit()
finally:
db.close()
r = admin.post(f"/api/accounts/{konto['id']}/saldo-uebernehmen")
assert r.status_code == 200, r.text
assert r.json()["balance_cents"] == 120000
assert r.json()["opening_balance_cents"] == 125000, "Differenz im Anfangsbestand"
# Die Buchung selbst ist unangetastet.
offen = admin.get("/api/transactions", params={"account_id": konto["id"]}).json()
assert len(offen) == 1 and offen[0]["amount_cents"] == 5000
def test_uebernehmen_ohne_bank_saldo_wird_erklaert(admin):
konto = _konto(admin, name=f"Ohne Saldo {secrets.token_hex(3)}", currency="CHF")
r = admin.post(f"/api/accounts/{konto['id']}/saldo-uebernehmen")
assert r.status_code == 400
assert "abrufen" in r.json()["detail"]
def test_api_antworten_werden_nicht_zwischengespeichert(admin):
"""Ohne das darf ein Browser selbst entscheiden – und der Zurueck-Knopf
zeigt Salden von vorhin."""
r = admin.get("/api/inbox")
assert r.headers.get("cache-control") == "no-store"
r = admin.get("/api/accounts")
assert r.headers.get("cache-control") == "no-store"
def test_verknuepfte_buchung_wird_nicht_zum_uebertrag(admin):
"""Sind beide Seiten gebucht und verbunden, waere ein Uebertrag falsch: Er
bewegt das Geld auf dem Zielkonto ein zweites Mal."""
a = _konto(admin, name=f"Verk-A {secrets.token_hex(3)}", currency="CHF")
b = _konto(admin, name=f"Verk-B {secrets.token_hex(3)}", currency="CHF")
betrag = 11000 + secrets.randbelow(900)
raus = admin.post("/api/transactions", json={
"type": "expense", "date": "2026-08-15", "account_id": a["id"],
"amount_cents": betrag, "payee": "Umbuchung"}).json()
rein = admin.post("/api/transactions", json={
"type": "income", "date": "2026-08-15", "account_id": b["id"],
"amount_cents": betrag, "payee": "Umbuchung"}).json()
assert admin.post("/api/uebertraege/verbinden",
json={"a_id": raus["id"], "b_id": rein["id"]}).status_code == 200
r = admin.post(f"/api/transactions/{raus['id']}/transfer",
json={"to_account_id": b["id"]})
assert r.status_code == 400, r.text
assert "verbunden" in r.json()["detail"]
# Auch mit "trotzdem" nicht – es ist nachweislich falsch, keine Ansichtssache.
r = admin.post(f"/api/transactions/{raus['id']}/transfer",
json={"to_account_id": b["id"], "trotzdem": True})
assert r.status_code == 400
def test_schon_gebuchte_gegenseite_wird_gemeldet(admin):
"""Ohne diese Warnung stuende das Geld auf dem Zielkonto zweimal."""
a = _konto(admin, name=f"Doppel-A {secrets.token_hex(3)}", currency="CHF")
b = _konto(admin, name=f"Doppel-B {secrets.token_hex(3)}", currency="CHF")
betrag = 12000 + secrets.randbelow(900)
raus = admin.post("/api/transactions", json={
"type": "expense", "date": "2026-08-16", "account_id": a["id"],
"amount_cents": betrag}).json()
# Die Gegenseite liegt schon gebucht auf B – nur eben unverbunden.
admin.post("/api/transactions", json={
"type": "income", "date": "2026-08-17", "account_id": b["id"],
"amount_cents": betrag})
r = admin.post(f"/api/transactions/{raus['id']}/transfer",
json={"to_account_id": b["id"]})
assert r.status_code == 409, r.text
assert "verbinden" in r.json()["detail"]
assert admin.get(f"/api/accounts/{b['id']}").json()["balance_cents"] == betrag
# Der Treffer ist unscharf (Datum, Betrag) – wer es besser weiss, kommt durch.
r = admin.post(f"/api/transactions/{raus['id']}/transfer",
json={"to_account_id": b["id"], "trotzdem": True})
assert r.status_code == 200, r.text
def test_uebertrag_nimmt_den_kurs_der_bank_auch_nachtraeglich(admin):
"""Der Eingangs-Eintrag traegt den Kurs. Ohne ihn bliebe hier nur die
EZB-Naeherung – und die passt nicht auf den Rappen."""
from app.database import SessionLocal
from app.models import InboxItem
chf = _konto(admin, name=f"Kurs-CHF {secrets.token_hex(3)}", currency="CHF")
eur = _konto(admin, name=f"Kurs-EUR {secrets.token_hex(3)}", currency="EUR")
txn = admin.post("/api/transactions", json={
"type": "expense", "date": "2026-08-18", "account_id": chf["id"],
"amount_cents": 40000, "payee": "Exchanged to EUR"}).json()
# So sieht es aus, wenn die Bank ihren Kurs mitgeliefert hat.
db = SessionLocal()
try:
db.add(InboxItem(
owner_id=admin.get("/api/auth/me").json()["id"],
source="bank", status="booked", account_id=chf["id"],
date=__import__("datetime").date(2026, 8, 18), amount_cents=-40000,
currency="CHF", payee="Exchanged to EUR", transaction_id=txn["id"],
counter_amount_cents=42532, counter_currency="EUR"))
db.commit()
finally:
db.close()
r = admin.post(f"/api/transactions/{txn['id']}/transfer",
json={"to_account_id": eur["id"]})
assert r.status_code == 200, r.text
assert r.json()["to_amount_cents"] == 42532, "der Betrag der Bank, nicht der gerechnete"
assert admin.get(f"/api/accounts/{eur['id']}").json()["balance_cents"] == 42532