Files
certain/backend/tests/test_import.py
T
GuidoandClaude Opus 4.8 1381967aae feat: Phase 4 – CSV-Import (Templates, Prüfansicht, Duplikat/Transfer-Erkennung, Undo)
Backend:
- Modelle + Migration 0005: import_templates, import_batches, categorization_rules;
  transactions um import_batch_id + import_hash erweitert.
- services/csv_import: Datei parsen (Trennzeichen, Dezimal-/Datumsformat, Kodierung,
  Header/Skip, Vorzeichen vs. Soll/Haben), Vorschau mit Duplikat-Erkennung (Hash),
  interner-Transfer-Erkennung via IBAN, lernender Auto-Kategorisierung; Buchen mit
  Batch und Undo.
- routers/imports: Templates (CRUD), Roh-Vorschau, Prüf-Vorschau, Buchen, Verlauf,
  Rückgängig. pytest: Parsing, Duplikate, Transfer, Buchen, Undo, Lernen.

Frontend:
- Import-Seite: Konto+Datei, Format/Spaltenzuordnung (mit Roh-Vorschau),
  editierbare Prüfansicht (Kategorie/Typ/Zielkonto, Duplikate markiert),
  Buchen, Import-Verlauf mit Undo; Vorlagen speichern; i18n DE/EN.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-19 20:06:50 +02:00

104 lines
3.7 KiB
Python

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" # Duplikat der ersten Zeile
"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: Duplikat der Zeile 0
assert rows[2]["is_duplicate"] is True
assert rows[2]["include"] is False
# 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()
# Duplikat wird übersprungen -> 3 gebucht (Ausgabe, Einnahme, Transfer).
assert batch["imported_count"] == 3
# Saldo: 0 + 3000 - 50 - 200(Transfer raus) = 2750,00
bal = admin.get(f"/api/accounts/{giro['id']}").json()["balance_cents"]
assert bal == 275000, 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"] == 3
bal2 = admin.get(f"/api/accounts/{giro['id']}").json()["balance_cents"]
assert bal2 == 0, bal2