GuidoandClaude Opus 4.8
14b3e53685
Bisher wurde eine zweite Buchung mit gleichem Tag/Betrag/Text (z. B. zwei Gutschriften à 324 am selben Tag) fälschlich als Duplikat verworfen. Jetzt gilt eine Zeile nur als Duplikat, wenn bereits GENAU SO VIELE identische Buchungen in der DB stehen – ein erneuter Import derselben Datei bleibt weiterhin Duplikat-sicher. - csv_import.detect_rows: existing/seen als Counter (Multiset) statt Set - Tests angepasst + Re-Import-Test ergänzt; an echtem Raiffeisen-Auszug verifiziert (254 Zeilen, 0 fälschliche Duplikate, Saldo == Bank-Endsaldo) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
121 lines
4.6 KiB
Python
121 lines
4.6 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" # 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
|