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: die IBAN gehört einem eigenen Konto – der Import verbindet # trotzdem nichts mehr, es bleibt eine gewöhnliche Ausgabe. assert rows[3]["type"] == "expense" assert rows[3]["to_account_id"] is None assert spar 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_umtausch_wird_als_ausgabe_und_einnahme_gebucht(admin): """Beide Seiten eines Umtauschs zählen einzeln – der Import verbindet nichts. Früher wurde daraus ein Übertrag mit geratenem Gegenkonto. Das ging schief, sobald beide Auszüge eingelesen wurden oder das Ziel falsch geraten war. Jetzt ist jede Zeile das, was im Auszug steht: Abgang beim einen Konto, Zugang beim anderen. Beide Salden stimmen damit ohne Zutun. """ 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) admin.post("/api/categories/transfer-keywords", json={"word": "umgetauscht in eur", "target_account_id": eur["id"]}) rows = _rev_preview(admin, chf["id"], CHF_UMTAUSCH) assert rows[0]["type"] == "expense" assert rows[0]["to_account_id"] is None 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 # Der EUR-Auszug bringt den Zugang – als Einnahme, nicht als Gegenbuchung. rows2 = _rev_preview(admin, eur["id"], EUR_UMTAUSCH) assert rows2[0]["type"] == "income" assert rows2[0]["is_counterpart"] is False assert rows2[0]["include"] is True r = admin.post("/api/imports/commit", json={"account_id": eur["id"], "filename": "eur.csv", "rows": rows2}) assert r.json()["imported_count"] == 1 assert admin.get(f"/api/accounts/{eur['id']}").json()["balance_cents"] == 10642 assert admin.get(f"/api/accounts/{chf['id']}").json()["balance_cents"] == 40000 def test_uebertrags_schlagwort_setzt_neutrale_kategorie(admin): """Ein früheres Übertrags-Schlagwort kennzeichnet die Zeile jetzt nur noch. Es verbindet keine Konten mehr, sondern hängt die neutrale Kategorie an – die Buchung verändert den Saldo, bleibt aber aus der Statistik heraus. """ grp = admin.post("/api/categories/groups", json={"name": "Neutral X", "kind": "expense"}).json() neutral = admin.post("/api/categories", json={ "group_id": grp["id"], "name": "Kontoübertrag X", "excluded_from_stats": True}).json() eur = _account(admin, name="Rev EUR Z", currency="EUR") chf = _account(admin, name="Rev CHF Z", currency="CHF", opening_balance_cents=50000) admin.post("/api/categories/transfer-keywords", json={"word": "umgetauscht in eur", "target_account_id": eur["id"]}) rows = _rev_preview(admin, chf["id"], CHF_UMTAUSCH) assert rows[0]["type"] == "expense" assert rows[0]["category_id"] == neutral["id"] assert rows[0]["needs_target"] is False def test_expense_is_not_swallowed_by_incoming_transfer(admin): """Eine echte Ausgabe bleibt eine Ausgabe – auch wenn anderswo ein gleich hoher Übertrag steht. Früher wurde sie als dessen Gegenbuchung geschluckt.""" 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"] # Keine Zeile gilt mehr als Gegenbuchung – beide werden gebucht. assert all(r["is_counterpart"] is False for r in rows) assert all(r["include"] is True for r in rows) assert rows[0]["type"] == "expense" assert rows[1]["type"] == "income" # --- 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_keine_zeile_bleibt_wegen_fehlendem_gegenkonto_liegen(admin): """Jede Zeile des Auszugs wird gebucht – der Saldo-Abgleich geht auf. Früher liess eine Übertrags-Regel ohne erkennbares Gegenkonto die Zeile ungebucht; das Geld fehlte unbemerkt und der Abgleich schlug fehl. Ohne automatische Kontoverknüpfung gibt es diesen Zustand nicht mehr. """ giro = _account(admin, name="Ziel offen", currency="CHF") admin.post("/api/categories/transfer-keywords", json={"word": "uebertrag", "target_account_id": giro["id"]}) 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() assert not [r for r in res["rows"] if r.get("needs_target")] uebertrag = res["rows"][1] assert uebertrag["type"] == "income" and uebertrag["include"] is True rec = res["reconciliation"] assert rec["ok"] is True, "der Auszug muss ohne Rest aufgehen" assert rec["difference_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()) KURZ_CSV = ( "Datum;Betrag;Empfaenger;Zweck;GegenIBAN\n" "05.03.2026;-48,97;PayPal Europe S.a.r.l.;Lastschrift;\n" "06.03.2026;-30,00;Europa-Park;Eintritt;\n" "07.03.2026;-24,50;TWINT EUROBUS AG;Einkauf;\n" "08.03.2026;-200,00;Umbuchung;Uebertrag Euro Konto;\n" ) def test_kurzer_erkennungstext_trifft_nicht_mitten_im_wort(admin): """Ein kurzer Erkennungs-Text darf nicht als blosser Teilstring greifen. Am EUR-Konto stand als Erkennungs-Text „Euro". Weil er ungeprüft als Teilstring gesucht wurde, wurden „PayPal Europe", „Europa-Park" und „EUROBUS" zu Überträgen auf dieses Konto – das Geld landete dort zusätzlich und der Saldo lief weit daneben. Nur ein eigenständiges Wort zählt. """ giro = _account(admin, name="Giro CHF", currency="CHF") _account(admin, name="Euro Konto", currency="EUR", identifiers=[{"kind": "text", "value": "Euro"}]) r = admin.post( "/api/imports/preview", files={"file": ("k.csv", io.BytesIO(KURZ_CSV.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": COLMAP}, ) assert r.status_code == 200, r.text typ = {row["payee"]: row["type"] for row in r.json()["rows"]} assert typ["PayPal Europe S.a.r.l."] == "expense" assert typ["Europa-Park"] == "expense" assert typ["TWINT EUROBUS AG"] == "expense" # Auch als eigenständiges Wort verbindet der Erkennungs-Text keine Konten mehr. assert typ["Umbuchung"] == "expense" MEHRERE_PRODUKTE = ( "Produkt;Datum;Betrag;Empfaenger;Zweck\n" "Giro;01.03.2026;-10,00;Coop;\n" "Erspartes;01.03.2026;0,50;Aufrundung;\n" "Giro;02.03.2026;-20,00;Migros;\n" "Anlagekonto;02.03.2026;-100,00;Depot;\n" "Giro;03.03.2026;-5,00;Kiosk;\n" ) def test_rohvorschau_nennt_die_waehlbaren_filterwerte(admin): """Revolut legt alle Produkte in EINE Datei. Damit man den Zeilenfilter setzen kann, ohne den Wert zu erraten, liefert die Vorschau die Werte je Spalte samt Zeilenzahl mit.""" r = admin.post( "/api/imports/raw", files={"file": ("m.csv", io.BytesIO(MEHRERE_PRODUKTE.encode("utf-8")), "text/csv")}, data={"delimiter": ";", "encoding": "utf-8", "has_header": "true", "skip_rows": "0"}, ) assert r.status_code == 200, r.text daten = r.json() assert daten["row_count"] == 5 produkte = daten["filter_values"]["0"] assert [p["value"] for p in produkte] == ["Giro", "Erspartes", "Anlagekonto"] assert produkte[0]["count"] == 3 # Spalten mit lauter verschiedenen Werten sind keine Gruppen. assert "3" not in daten["filter_values"]