import io import json def _account(admin, **kw): body = {"name": "Giro", "type": "girokonto", "currency": "EUR", "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 _leb(admin): for g in admin.get("/api/categories").json(): for c in g["categories"]: if c["name"] == "Lebensmittel": return c["id"] raise AssertionError("Lebensmittel fehlt") CSV = ( "Datum;Betrag;Empfaenger;Zweck;GegenIBAN\n" "15.03.2026;-50,00;REWE;Einkauf;\n" "25.03.2026;3.000,00;Arbeitgeber;Gehalt;\n" "15.03.2026;-50,00;REWE;Einkauf;\n" # gleich aussehend, aber echte 2. Buchung "20.03.2026;-200,00;Sparen;Ueberweisung;CH9300762011623852957\n" # interner Transfer ) COLMAP = json.dumps({"date": 0, "amount": 1, "payee": 2, "purpose": 3, "counterparty_iban": 4}) def _preview(admin, account_id): return admin.post( "/api/imports/preview", files={"file": ("konto.csv", io.BytesIO(CSV.encode("utf-8")), "text/csv")}, data={ "account_id": str(account_id), "delimiter": ";", "decimal_sep": ",", "date_format": "%d.%m.%Y", "encoding": "utf-8", "has_header": "true", "skip_rows": "0", "amount_mode": "signed", "column_map": COLMAP, }, ) def test_preview_parsing_and_detection(admin): giro = _account(admin, name="Giro Import") # Zweites Konto mit passender IBAN für die Transfer-Erkennung. spar = _account(admin, name="Spar", type="sparkonto", currency="EUR", identifiers=[{"kind": "iban", "value": "CH93 0076 2011 6238 5295 7"}]) r = _preview(admin, giro["id"]) assert r.status_code == 200, r.text rows = r.json()["rows"] assert len(rows) == 4 # Zeile 0: Ausgabe 50 EUR assert rows[0]["type"] == "expense" assert rows[0]["amount_cents"] == -5000 assert rows[0]["is_duplicate"] is False # Zeile 1: Einnahme 3000 (Tausenderpunkt korrekt) assert rows[1]["type"] == "income" assert rows[1]["amount_cents"] == 300000 # Zeile 2: gleich aussehend wie Zeile 0, aber im selben Auszug KEIN Duplikat. assert rows[2]["is_duplicate"] is False assert rows[2]["include"] is True # Zeile 3: interner Transfer (IBAN passt zu 'spar') assert rows[3]["type"] == "transfer" assert rows[3]["to_account_id"] == spar["id"] def test_commit_and_undo_and_learning(admin): giro = _account(admin, name="Giro C") leb = _leb(admin) rows = _preview(admin, giro["id"]).json()["rows"] # Nutzer kategorisiert die erste (REWE) als Lebensmittel -> soll gelernt werden. for row in rows: if row["payee"] == "REWE" and not row["is_duplicate"]: row["category_id"] = leb r = admin.post("/api/imports/commit", json={ "account_id": giro["id"], "filename": "konto.csv", "rows": rows, }) assert r.status_code == 200, r.text batch = r.json() # Beide REWE + Einnahme + Transfer = 4 gebucht (nichts fälschlich als Duplikat verworfen). assert batch["imported_count"] == 4 # Saldo: 0 + 3000 - 50 - 50 - 200(Transfer raus) = 2700,00 bal = admin.get(f"/api/accounts/{giro['id']}").json()["balance_cents"] assert bal == 270000, bal # Regel gelernt: erneute Vorschau schlägt für REWE Lebensmittel vor. rows2 = _preview(admin, giro["id"]).json()["rows"] rewe = next(r for r in rows2 if r["payee"] == "REWE") assert rewe["category_id"] == leb # Undo entfernt die Buchungen dieses Batches wieder. r = admin.delete(f"/api/imports/batches/{batch['id']}") assert r.status_code == 200 assert r.json()["deleted"] == 4 bal2 = admin.get(f"/api/accounts/{giro['id']}").json()["balance_cents"] assert bal2 == 0, bal2 def test_reimport_same_file_is_deduped(admin): """Erneuter Import derselben Datei erkennt ALLE Zeilen als Duplikat – auch die zwei gleich aussehenden REWE-Buchungen (Multiset-Zählung).""" giro = _account(admin, name="Giro Re") _account(admin, name="Spar Re", type="sparkonto", currency="EUR", identifiers=[{"kind": "iban", "value": "CH93 0076 2011 6238 5295 7"}]) rows = _preview(admin, giro["id"]).json()["rows"] r = admin.post("/api/imports/commit", json={"account_id": giro["id"], "filename": "k.csv", "rows": rows}) assert r.json()["imported_count"] == 4 # Zweiter Vorschau-Lauf derselben Datei: nichts mehr buchbar. again = _preview(admin, giro["id"]).json()["rows"] assert all(r["is_duplicate"] for r in again) assert sum(1 for r in again if r["include"]) == 0 # --- Revolut-Besonderheiten: Gebühr, Storno, Fremdwährung ------------------ REVOLUT_CSV = ( "Type,Product,Started Date,Completed Date,Description,Amount,Fee,Currency,State,Balance\n" "TOPUP,Current,2026-01-06 10:00:00,2026-01-06 10:00:00,Von Raiffeisen,200.00,0.00,CHF,COMPLETED,300.23\n" "CARD_PAYMENT,Current,2026-01-07 10:00:00,2026-01-07 10:00:00,Coop,-20.00,0.50,CHF,COMPLETED,279.73\n" "CARD_PAYMENT,Current,2026-01-08 10:00:00,2026-01-08 10:00:00,Storno Laden,-15.00,0.00,CHF,REVERTED,279.73\n" "CARD_PAYMENT,Current,2026-01-09 10:00:00,2026-01-09 10:00:00,Amazon DE,-10.00,0.00,EUR,COMPLETED,90.00\n" ) REVOLUT_MAP = json.dumps({ "date": 3, "amount": 5, "payee": 4, "fee": 6, "currency": 7, "state": 8, }) def _preview_revolut(admin, account_id): return admin.post( "/api/imports/preview", files={"file": ("revolut.csv", io.BytesIO(REVOLUT_CSV.encode("utf-8")), "text/csv")}, data={ "account_id": str(account_id), "delimiter": ",", "decimal_sep": ".", "date_format": "%Y-%m-%d %H:%M:%S", "encoding": "utf-8", "has_header": "true", "skip_rows": "0", "amount_mode": "signed", "column_map": REVOLUT_MAP, }, ).json()["rows"] def test_revolut_fee_state_and_currency(admin): acc = _account(admin, name="Revolut CHF", currency="CHF", opening_balance_cents=10023) rows = _preview_revolut(admin, acc["id"]) assert len(rows) == 4 # Aufladung ohne Gebühr. assert rows[0]["amount_cents"] == 20000 # Gebühr 0.50 mindert die Ausgabe: -20.00 - 0.50 = -20.50 assert rows[1]["amount_cents"] == -2050 # Stornierte Buchung zählt nicht. assert rows[2]["valid"] is False and rows[2]["include"] is False assert rows[2]["skip_reason"] == "state" # Fremdwährungs-Zeile gehört nicht in das CHF-Konto. assert rows[3]["valid"] is False and rows[3]["include"] is False assert rows[3]["skip_reason"] == "currency" r = admin.post("/api/imports/commit", json={ "account_id": acc["id"], "filename": "revolut.csv", "rows": rows, }) assert r.json()["imported_count"] == 2 # 100.23 + 200.00 - 20.50 = 279.73 (= Balance-Spalte der Bank) bal = admin.get(f"/api/accounts/{acc['id']}").json()["balance_cents"] assert bal == 27973, bal # Deutscher Revolut-Export: Status heißt „ABGESCHLOSSEN", Gebühren stehen in # einer eigenen Spalte (bei Abo-Gebühren ist der Betrag sogar 0.00). REVOLUT_DE_CSV = ( "Art,Produkt,Datum des Beginns,Datum des Abschlusses,Beschreibung,Betrag,Gebühr,Währung,Status,Kontostand\n" "Einzahlung,Giro,2026-01-06 23:04:52,2026-01-06 23:05:01,Einzahlung von *5264,100.00,0.00,CHF,ABGESCHLOSSEN,100.23\n" "Einzahlung,Giro,2026-01-06 23:08:15,2026-01-06 23:08:36,Einzahlung von *5264,150.00,0.00,CHF,ABGESCHLOSSEN,250.23\n" "Umtausch,Giro,2026-01-06 23:08:58,2026-01-06 23:08:58,Revolut Ltd,-100.00,0.00,CHF,ABGESCHLOSSEN,150.23\n" "Transfer,Giro,2026-01-06 23:09:35,2026-01-06 23:09:35,Auf Anlagekonto,-31.90,0.00,CHF,ABGESCHLOSSEN,118.33\n" "Transfer,Giro,2026-01-06 23:10:22,2026-01-06 23:10:22,Auf Anlagekonto,-118.00,0.00,CHF,ABGESCHLOSSEN,0.33\n" "Einzahlung,Giro,2026-01-08 15:52:44,2026-01-08 15:53:25,Einzahlung von *4469,150.00,0.00,CHF,ABGESCHLOSSEN,150.33\n" "Transfer,Giro,2026-01-09 08:34:11,2026-01-09 08:34:12,Auf Anlagekonto,-50.00,0.00,CHF,ABGESCHLOSSEN,100.33\n" "Belastung,Giro,2026-01-11 03:17:01,2026-01-11 03:17:01,Premium Abo-Gebühr,0.00,10.99,CHF,ABGESCHLOSSEN,89.34\n" "Kartenzahlung,Giro,2026-01-12 10:00:00,2026-01-12 10:00:00,Storno Laden,-25.00,0.00,CHF,STORNIERT,89.34\n" ) def test_revolut_german_export(admin): """Deutscher Export: ABGESCHLOSSEN ist buchbar, STORNIERT nicht, und die Gebührenspalte wird auch dann verrechnet, wenn der Betrag 0.00 ist.""" acc = _account(admin, name="Revolut DE", currency="CHF", opening_balance_cents=23) rows = admin.post( "/api/imports/preview", files={"file": ("r.csv", io.BytesIO(REVOLUT_DE_CSV.encode("utf-8")), "text/csv")}, data={ "account_id": str(acc["id"]), "delimiter": ",", "decimal_sep": ".", "date_format": "%Y-%m-%d %H:%M:%S", "encoding": "utf-8", "has_header": "true", "skip_rows": "0", "amount_mode": "signed", "column_map": REVOLUT_MAP, }, ).json()["rows"] assert len(rows) == 9 # ABGESCHLOSSEN wird gebucht (früher scheiterte das an der COMPLETED-Prüfung). assert [r["include"] for r in rows[:8]] == [True] * 8 # Reine Gebührenzeile: Betrag 0.00, Gebühr 10.99 -> -10.99 assert rows[7]["amount_cents"] == -1099 # STORNIERT bleibt außen vor. assert rows[8]["include"] is False and rows[8]["skip_reason"] == "state" r = admin.post("/api/imports/commit", json={ "account_id": acc["id"], "filename": "r.csv", "rows": rows, }) assert r.json()["imported_count"] == 8 # 0.23 + 100 + 150 - 100 - 31.90 - 118 + 150 - 50 - 10.99 = 89.34 (= Kontostand-Spalte) bal = admin.get(f"/api/accounts/{acc['id']}").json()["balance_cents"] assert bal == 8934, bal # --- Saldo-Abgleich gegen die Kontostand-Spalte --------------------------- def _preview_full(admin, account_id, csv_text, colmap): return admin.post( "/api/imports/preview", files={"file": ("r.csv", io.BytesIO(csv_text.encode("utf-8")), "text/csv")}, data={ "account_id": str(account_id), "delimiter": ",", "decimal_sep": ".", "date_format": "%Y-%m-%d %H:%M:%S", "encoding": "utf-8", "has_header": "true", "skip_rows": "0", "amount_mode": "signed", "column_map": colmap, }, ).json() REVOLUT_MAP_BAL = json.dumps({ "date": 3, "amount": 5, "payee": 4, "fee": 6, "currency": 7, "state": 8, "balance": 9, }) def test_reconciliation_ok(admin): """Stimmiger Auszug: Abgleich meldet ok und den Anfangssaldo aus der Datei.""" acc = _account(admin, name="Rec OK", currency="CHF") res = _preview_full(admin, acc["id"], REVOLUT_DE_CSV, REVOLUT_MAP_BAL) rec = res["reconciliation"] assert rec["available"] is True assert rec["ok"] is True, rec assert rec["mismatches"] == 0 # Anfangssaldo laut Datei: 100.23 - 100.00 = 0.23 assert rec["opening_balance_cents"] == 23 assert rec["final_balance_cents"] == 8934 assert rec["computed_final_cents"] == 8934 assert rec["difference_cents"] == 0 def test_reconciliation_detects_wrong_amount(admin): """Wird die Gebühr NICHT verrechnet, zeigt der Abgleich genau die Zeile.""" acc = _account(admin, name="Rec Bad", currency="CHF") ohne_gebuehr = json.dumps({ "date": 3, "amount": 5, "payee": 4, "currency": 7, "state": 8, "balance": 9, }) # fee bewusst nicht zugeordnet res = _preview_full(admin, acc["id"], REVOLUT_DE_CSV, ohne_gebuehr) rec = res["reconciliation"] assert rec["ok"] is False assert rec["difference_cents"] == -1099 # 10.99 fehlen mm = rec["first_mismatch"] assert mm["payee"] == "Premium Abo-Gebühr" assert mm["expected_cents"] == -1099 # laut Kontostand-Spalte assert mm["actual_cents"] == 0 # ohne Gebührenspalte erkannt def test_date_format_tolerates_time(admin): """Wählt man ein reines Datumsformat, obwohl die Spalte eine Uhrzeit enthält, werden die Zeilen trotzdem erkannt (früher: alles ungültig).""" acc = _account(admin, name="Datum Zeit", currency="CHF") nur_datum = json.dumps({ "date": 3, "amount": 5, "payee": 4, "fee": 6, "currency": 7, "state": 8, "balance": 9, }) res = admin.post( "/api/imports/preview", files={"file": ("r.csv", io.BytesIO(REVOLUT_DE_CSV.encode("utf-8")), "text/csv")}, data={ "account_id": str(acc["id"]), "delimiter": ",", "decimal_sep": ".", "date_format": "%Y-%m-%d", # ohne Uhrzeit, Datei hat aber welche "encoding": "utf-8", "has_header": "true", "skip_rows": "0", "amount_mode": "signed", "column_map": nur_datum, }, ).json() assert all(r["date"] for r in res["rows"]), "Datum trotz Uhrzeit erkannt" assert res["reconciliation"]["ok"] is True # --- Währungstausch: beide Auszüge ergeben EINEN Transfer ------------------- def _rev_preview(admin, account_id, csv_text): return admin.post( "/api/imports/preview", files={"file": ("x.csv", io.BytesIO(csv_text.encode("utf-8")), "text/csv")}, data={ "account_id": str(account_id), "delimiter": ",", "decimal_sep": ".", "date_format": "%Y-%m-%d %H:%M:%S", "encoding": "utf-8", "has_header": "true", "skip_rows": "0", "amount_mode": "signed", "column_map": json.dumps({"date": 3, "amount": 5, "payee": 4, "fee": 6, "currency": 7, "state": 8, "balance": 9}), }, ).json()["rows"] CHF_UMTAUSCH = ( "Art,Produkt,Beginn,Abschluss,Beschreibung,Betrag,Gebühr,Währung,Status,Kontostand\n" "Umtausch,Giro,2026-02-01 10:00:00,2026-02-01 10:00:00,Umgetauscht in EUR,-100.00,0.00,CHF,ABGESCHLOSSEN,400.00\n" ) EUR_UMTAUSCH = ( "Art,Produkt,Beginn,Abschluss,Beschreibung,Betrag,Gebühr,Währung,Status,Kontostand\n" "Umtausch,Giro,2026-02-01 10:00:00,2026-02-01 10:00:00,Umgetauscht in EUR,106.42,0.00,EUR,ABGESCHLOSSEN,106.42\n" ) def test_currency_exchange_pairs_across_statements(admin): chf = _account(admin, name="Rev CHF X", currency="CHF", opening_balance_cents=50000) eur = _account(admin, name="Rev EUR X", currency="EUR", opening_balance_cents=0) # EINE Regel genügt: „Umgetauscht in EUR" -> Gegenkonto ist das EUR-Konto. admin.post("/api/categories/transfer-keywords", json={"word": "umgetauscht in eur", "target_account_id": eur["id"]}) # 1) CHF-Auszug: Umtausch geht als Transfer CHF -> EUR raus. rows = _rev_preview(admin, chf["id"], CHF_UMTAUSCH) assert rows[0]["type"] == "transfer" assert rows[0]["to_account_id"] == eur["id"] assert rows[0]["is_counterpart"] is False admin.post("/api/imports/commit", json={"account_id": chf["id"], "filename": "chf.csv", "rows": rows}) assert admin.get(f"/api/accounts/{chf['id']}").json()["balance_cents"] == 40000 # 2) EUR-Auszug: OHNE weitere Regel – die Gegenseite wird über den bereits # gebuchten Transfer erkannt, obwohl der Betrag in EUR ein anderer ist. rows2 = _rev_preview(admin, eur["id"], EUR_UMTAUSCH) assert rows2[0]["type"] == "transfer" assert rows2[0]["is_counterpart"] is True, "Umtausch-Gegenseite muss erkannt werden" assert rows2[0]["include"] is False, "darf nicht doppelt zählen" r = admin.post("/api/imports/commit", json={"account_id": eur["id"], "filename": "eur.csv", "rows": rows2}) assert r.json()["imported_count"] == 0 # Der echte Eingang korrigiert die EZB-Schätzung: EUR-Konto steht auf 106.42. assert admin.get(f"/api/accounts/{eur['id']}").json()["balance_cents"] == 10642 # CHF-Seite bleibt unverändert. assert admin.get(f"/api/accounts/{chf['id']}").json()["balance_cents"] == 40000 def test_exchange_arrival_without_source_asks_instead_of_booking(admin): """Wird der Empfangs-Auszug zuerst eingelesen, ist die Quelle noch unbekannt. Die Zeile darf dann NICHT still als Einnahme durchgehen, sondern fragt das Konto ab – sonst zählt die Bewegung später doppelt.""" chf = _account(admin, name="Rev CHF Y", currency="CHF", opening_balance_cents=50000) eur = _account(admin, name="Rev EUR Y", currency="EUR") admin.post("/api/categories/transfer-keywords", json={"word": "umgetauscht in eur", "target_account_id": eur["id"]}) rows = _rev_preview(admin, eur["id"], EUR_UMTAUSCH) assert rows[0]["type"] == "transfer" assert rows[0]["to_account_id"] is None assert rows[0]["needs_target"] is True assert rows[0]["include"] is False, "ohne Gegenkonto nicht automatisch buchen" r = admin.post("/api/imports/commit", json={"account_id": eur["id"], "filename": "e.csv", "rows": rows}) assert r.json()["imported_count"] == 0 assert admin.get(f"/api/accounts/{eur['id']}").json()["balance_cents"] == 0 assert chf # Quellkonto bleibt unberührt def test_expense_is_not_swallowed_by_incoming_transfer(admin): """Eine echte Ausgabe über denselben Betrag wie ein erwarteter Transfer-Eingang darf NICHT als Gegenbuchung gelten – sonst fehlt sie und der Saldo läuft zu hoch.""" quelle = _account(admin, name="Quelle Z", currency="EUR", opening_balance_cents=100000) ziel = _account(admin, name="Ziel Z", currency="EUR", opening_balance_cents=10000) # Transfer 50,00 von Quelle -> Ziel buchen. r = admin.post("/api/transactions", json={ "type": "transfer", "date": "2026-03-10", "account_id": quelle["id"], "to_account_id": ziel["id"], "amount_cents": 5000, }) assert r.status_code == 201, r.text csv_text = ( "Datum;Betrag;Empfaenger;Zweck;GegenIBAN\n" "11.03.2026;-50,00;Supermarkt;Einkauf;\n" # echte Ausgabe, gleicher Betrag "12.03.2026;50,00;Von Quelle;Uebertrag;\n" # das ist die echte Gegenbuchung ) rows = admin.post( "/api/imports/preview", files={"file": ("z.csv", io.BytesIO(csv_text.encode("utf-8")), "text/csv")}, data={ "account_id": str(ziel["id"]), "delimiter": ";", "decimal_sep": ",", "date_format": "%d.%m.%Y", "encoding": "utf-8", "has_header": "true", "skip_rows": "0", "amount_mode": "signed", "column_map": COLMAP, }, ).json()["rows"] assert rows[0]["is_counterpart"] is False, "Ausgabe ist keine Gegenbuchung" assert rows[0]["include"] is True assert rows[1]["is_counterpart"] is True, "der Eingang schon" assert rows[1]["include"] is False admin.post("/api/imports/commit", json={"account_id": ziel["id"], "filename": "z.csv", "rows": rows}) # 100.00 + 50.00 (Transfer) - 50.00 (Einkauf) = 100.00 assert admin.get(f"/api/accounts/{ziel['id']}").json()["balance_cents"] == 10000 # --- Zeilenfilter: mehrere Pockets in einer Datei --------------------------- MULTI_POCKET_CSV = ( "Art,Produkt,Beginn,Abschluss,Beschreibung,Betrag,Gebühr,Währung,Status,Kontostand\n" "Einzahlung,Giro,2026-01-01 10:00:00,2026-01-01 10:00:00,Von Bank,100.00,0.00,CHF,ABGESCHLOSSEN,100.00\n" "Transfer,Erspartes,2026-01-02 10:00:00,2026-01-02 10:00:00,Ins Erspartes,50.00,0.00,CHF,ABGESCHLOSSEN,50.00\n" "Kartenbezahlung,Giro,2026-01-03 10:00:00,2026-01-03 10:00:00,Coop,-20.00,0.00,CHF,ABGESCHLOSSEN,80.00\n" "Zinsen,Erspartes,2026-01-04 10:00:00,2026-01-04 10:00:00,Zins,1.00,0.00,CHF,ABGESCHLOSSEN,51.00\n" ) POCKET_MAP = json.dumps({"date": 3, "amount": 5, "payee": 4, "fee": 6, "currency": 7, "state": 8, "balance": 9}) def _preview_filtered(admin, account_id, fcol=None, fval=None): data = { "account_id": str(account_id), "delimiter": ",", "decimal_sep": ".", "date_format": "%Y-%m-%d %H:%M:%S", "encoding": "utf-8", "has_header": "true", "skip_rows": "0", "amount_mode": "signed", "column_map": POCKET_MAP, } if fcol is not None: data["filter_column"] = str(fcol) data["filter_value"] = fval return admin.post( "/api/imports/preview", files={"file": ("p.csv", io.BytesIO(MULTI_POCKET_CSV.encode("utf-8")), "text/csv")}, data=data, ).json() def test_row_filter_splits_pockets(admin): """Ohne Filter landen fremde Pocket-Zeilen im Konto und der Saldo-Abgleich schlägt an; mit Filter stimmt jedes Konto für sich.""" giro = _account(admin, name="Pocket Giro", currency="CHF") spar = _account(admin, name="Pocket Spar", currency="CHF") ohne = _preview_filtered(admin, giro["id"]) assert len(ohne["rows"]) == 4 assert ohne["reconciliation"]["ok"] is False, "vermischte Ströme müssen auffallen" mit = _preview_filtered(admin, giro["id"], 1, "Giro") assert len(mit["rows"]) == 2 assert mit["reconciliation"]["ok"] is True assert mit["reconciliation"]["opening_balance_cents"] == 0 admin.post("/api/imports/commit", json={"account_id": giro["id"], "filename": "p.csv", "rows": mit["rows"]}) assert admin.get(f"/api/accounts/{giro['id']}").json()["balance_cents"] == 8000 ers = _preview_filtered(admin, spar["id"], 1, "Erspartes") assert len(ers["rows"]) == 2 assert ers["reconciliation"]["ok"] is True admin.post("/api/imports/commit", json={"account_id": spar["id"], "filename": "p.csv", "rows": ers["rows"]}) assert admin.get(f"/api/accounts/{spar['id']}").json()["balance_cents"] == 5100 REFUND_CSV = ( "Art,Produkt,Beginn,Abschluss,Beschreibung,Betrag,Gebühr,Währung,Status,Kontostand\n" "Einzahlung,Giro,2026-01-01 10:00:00,2026-01-01 10:00:00,Von Bank,100.00,0.00,CHF,ABGESCHLOSSEN,100.00\n" "Belastung,Giro,2026-01-02 10:00:00,2026-01-02 10:00:00,Premium Abo-Gebühr,0.00,10.99,CHF,ABGESCHLOSSEN,89.01\n" "Erstattung,Giro,2026-01-03 10:00:00,2026-01-03 10:00:00,Erstattung Abo-Kündigung,0.00,10.99,CHF,ABGESCHLOSSEN,100.00\n" ) def test_fee_row_direction_from_balance(admin): """Gebührenzeilen haben in beide Richtungen Betrag 0.00 und dieselbe Gebühr – Belastung und Erstattung unterscheidet nur die Kontostand-Spalte.""" acc = _account(admin, name="Gebuehren", currency="CHF") rows = admin.post( "/api/imports/preview", files={"file": ("f.csv", io.BytesIO(REFUND_CSV.encode("utf-8")), "text/csv")}, data={ "account_id": str(acc["id"]), "delimiter": ",", "decimal_sep": ".", "date_format": "%Y-%m-%d %H:%M:%S", "encoding": "utf-8", "has_header": "true", "skip_rows": "0", "amount_mode": "signed", "column_map": POCKET_MAP, }, ).json() assert rows["rows"][1]["amount_cents"] == -1099, "Gebühr belastet" assert rows["rows"][2]["amount_cents"] == 1099, "Gebühr erstattet" assert rows["reconciliation"]["ok"] is True def test_reconciliation_flags_rows_waiting_for_target(admin): """Eine Transfer-Regel ohne Gegenkonto lässt die Zeile ungebucht. Der Saldo-Abgleich muss das melden – sonst fehlt das Geld unbemerkt.""" giro = _account(admin, name="Ziel offen", currency="CHF") admin.post("/api/categories/transfer-keywords", json={"word": "uebertrag", "target_account_id": giro["id"]}) # zeigt auf sich selbst csv_text = ( "Datum;Betrag;Empfaenger;Zweck;Saldo\n" "01.03.2026;100,00;Lohn;Gehalt;100,00\n" "02.03.2026;92,00;Uebertrag von Sparkonto;;192,00\n" "03.03.2026;-10,00;Kiosk;;182,00\n" ) res = admin.post( "/api/imports/preview", files={"file": ("t.csv", io.BytesIO(csv_text.encode("utf-8")), "text/csv")}, data={ "account_id": str(giro["id"]), "delimiter": ";", "decimal_sep": ",", "date_format": "%d.%m.%Y", "encoding": "utf-8", "has_header": "true", "skip_rows": "0", "amount_mode": "signed", "column_map": json.dumps({"date": 0, "amount": 1, "payee": 2, "purpose": 3, "balance": 4}), }, ).json() wartend = [r for r in res["rows"] if r.get("needs_target")] assert len(wartend) == 1 and wartend[0]["amount_cents"] == 9200 rec = res["reconciliation"] assert rec["ok"] is False, "die wartende Zeile muss auffallen" assert rec["difference_cents"] == 9200 assert rec["first_mismatch"]["expected_cents"] == 9200 assert rec["first_mismatch"]["actual_cents"] == 0 def test_remove_undone_batch_from_history(admin): """Rückgängig gemachte Importe lassen sich aus dem Verlauf entfernen – laufende nicht, sonst verlöre man die Möglichkeit, sie zurückzunehmen.""" giro = _account(admin, name="Verlauf", currency="EUR") _account(admin, name="Spar Verlauf", type="sparkonto", currency="EUR", identifiers=[{"kind": "iban", "value": "CH93 0076 2011 6238 5295 7"}]) rows = _preview(admin, giro["id"]).json()["rows"] batch = admin.post("/api/imports/commit", json={"account_id": giro["id"], "filename": "v.csv", "rows": rows}).json() # Solange der Import steht, bleibt der Eintrag bestehen. r = admin.delete(f"/api/imports/batches/{batch['id']}/record") assert r.status_code == 400 assert any(b["id"] == batch["id"] for b in admin.get("/api/imports/batches").json()) admin.delete(f"/api/imports/batches/{batch['id']}") # rückgängig machen assert admin.delete(f"/api/imports/batches/{batch['id']}/record").status_code == 204 assert not any(b["id"] == batch["id"] for b in admin.get("/api/imports/batches").json())