"""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_zielkonto_wird_abgefangen(admin): """Der teuerste Fehler an dieser Stelle: Das Zielkonto haengt selbst an einer Bank und meldet den Zugang von allein – eine Gegenbuchung hier bedeutet das Geld zweimal.""" 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 == 409, r.text assert "angebunden" in r.json()["detail"] # Nichts gebucht, nichts verschoben. 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"] == 10000 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"