GuidoandClaude Opus 4.8
204a24d0df
Regeln: - Schlagwörter mit optionaler Betragsbedingung (z. B. „SV = 5.00 → Restaurant"); Betrags-Regeln gewinnen vor betragslosen - Transfer-Schlagwörter (Text → Transfer) auf der Konten-Seite - Konto-Identifikator-Art „Text": erkennt das Zielkonto eines Transfers aus dem Buchungstext (längster passender Text gewinnt) - Migration 0007, Endpunkte, Tests (Betrag/Transfer/Konflikt), gesamt 30 grün Optik: - App-Icon: Münzhaufen (aus Nutzer-SVG) weiß auf grünem Grund, klein gehalten - Farbschema auf pastelliges Grün (Akzent hell/dunkel + Favicon) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
209 lines
9.0 KiB
Python
209 lines
9.0 KiB
Python
"""Schlagwörter je Kategorie: Anlegen/Konflikt, Treffer (Empfänger ODER Zweck),
|
|
längstes gewinnt, Kind-Filter, Korrektur-Erkennung/Flagging, keine Rückwirkung."""
|
|
|
|
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 _expense_cats(admin, n=2):
|
|
out = []
|
|
for g in admin.get("/api/categories").json():
|
|
if g["kind"] == "expense":
|
|
out += g["categories"]
|
|
assert len(out) >= n, "zu wenige Ausgabe-Kategorien im Seed"
|
|
return out[:n]
|
|
|
|
|
|
def _income_cat(admin):
|
|
for g in admin.get("/api/categories").json():
|
|
if g["kind"] == "income" and g["categories"]:
|
|
return g["categories"][0]
|
|
raise AssertionError("keine Einnahme-Kategorie im Seed")
|
|
|
|
|
|
def _preview_csv(admin, account_id, csv, colmap):
|
|
r = admin.post(
|
|
"/api/imports/preview",
|
|
files={"file": ("k.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,
|
|
},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
return r.json()["rows"]
|
|
|
|
|
|
COLMAP = json.dumps({"date": 0, "amount": 1, "payee": 2, "purpose": 3})
|
|
|
|
|
|
def test_keyword_crud_and_conflict(admin):
|
|
a, b = _expense_cats(admin, 2)
|
|
r = admin.post(f"/api/categories/{a['id']}/keywords", json={"word": " REWE "})
|
|
assert r.status_code == 201, r.text
|
|
assert r.json()["word"] == "rewe" # getrimmt + kleingeschrieben
|
|
kid = r.json()["id"]
|
|
|
|
assert [k["word"] for k in admin.get(f"/api/categories/{a['id']}/keywords").json()] == ["rewe"]
|
|
|
|
# Gleiches Wort, gleiche Kategorie -> idempotent, keine Dublette.
|
|
assert admin.post(f"/api/categories/{a['id']}/keywords", json={"word": "rewe"}).status_code == 201
|
|
assert len(admin.get(f"/api/categories/{a['id']}/keywords").json()) == 1
|
|
|
|
# Gleiches Wort, andere Kategorie -> Konflikt.
|
|
assert admin.post(f"/api/categories/{b['id']}/keywords", json={"word": "rewe"}).status_code == 409
|
|
|
|
assert admin.delete(f"/api/categories/keywords/{kid}").status_code == 204
|
|
assert admin.get(f"/api/categories/{a['id']}/keywords").json() == []
|
|
|
|
|
|
def test_matching_payee_purpose_and_kind_filter(admin):
|
|
giro = _account(admin, name="Giro KW")
|
|
a, _ = _expense_cats(admin, 2)
|
|
admin.post(f"/api/categories/{a['id']}/keywords", json={"word": "netflix"})
|
|
|
|
csv = (
|
|
"Datum;Betrag;Empfaenger;Zweck\n"
|
|
"01.03.2026;-9,99;NETFLIX;Abo\n" # Empfänger-Treffer
|
|
"02.03.2026;-9,99;Kartenzahlung;NETFLIX Monatsabo\n" # Zweck-Treffer (any)
|
|
"03.03.2026;9,99;NETFLIX;Erstattung\n" # Einnahme -> darf NICHT greifen
|
|
)
|
|
rows = _preview_csv(admin, giro["id"], csv, COLMAP)
|
|
assert rows[0]["category_id"] == a["id"]
|
|
assert rows[0]["matched_rule_id"] is not None
|
|
assert rows[1]["category_id"] == a["id"] # über den Zweck getroffen
|
|
assert rows[2]["category_id"] is None # Ausgabe-Schlagwort, Einnahme -> gefiltert
|
|
|
|
|
|
def test_longest_keyword_wins(admin):
|
|
giro = _account(admin, name="Giro LW")
|
|
a, b = _expense_cats(admin, 2)
|
|
admin.post(f"/api/categories/{a['id']}/keywords", json={"word": "rewe"})
|
|
admin.post(f"/api/categories/{b['id']}/keywords", json={"word": "rewe getraenke"})
|
|
|
|
csv = "Datum;Betrag;Empfaenger;Zweck\n01.03.2026;-20,00;REWE Getraenke Markt;Wasser\n"
|
|
rows = _preview_csv(admin, giro["id"], csv, COLMAP)
|
|
assert rows[0]["category_id"] == b["id"] # spezifischeres (längeres) Schlagwort gewinnt
|
|
|
|
|
|
def test_correction_flagging_confirm(admin):
|
|
giro = _account(admin, name="Giro Flag")
|
|
a, b = _expense_cats(admin, 2)
|
|
kw = admin.post(f"/api/categories/{a['id']}/keywords", json={"word": "spotify"}).json()
|
|
|
|
lines = "".join(f"0{i}.03.2026;-9,99;SPOTIFY;Abo\n" for i in range(1, 7)) # 6 Zeilen
|
|
rows = _preview_csv(admin, giro["id"], "Datum;Betrag;Empfaenger;Zweck\n" + lines, COLMAP)
|
|
assert len(rows) == 6 and all(r["category_id"] == a["id"] for r in rows)
|
|
|
|
for r in rows[:2]: # zwei wegkorrigieren
|
|
r["category_id"] = b["id"]
|
|
res = admin.post("/api/imports/commit", json={"account_id": giro["id"], "filename": "s.csv", "rows": rows})
|
|
assert res.status_code == 200, res.text
|
|
|
|
flagged = admin.get("/api/categories/keywords/flagged").json()
|
|
hit = next((f for f in flagged if f["id"] == kw["id"]), None)
|
|
assert hit is not None, flagged
|
|
assert hit["applied_count"] == 6 and hit["overridden_count"] == 2
|
|
|
|
# „Behalten" unterdrückt die Meldung künftig.
|
|
assert admin.post(f"/api/categories/keywords/{kw['id']}/confirm").status_code == 204
|
|
assert kw["id"] not in [f["id"] for f in admin.get("/api/categories/keywords/flagged").json()]
|
|
|
|
|
|
def test_keyword_update_and_conflict(admin):
|
|
a, b = _expense_cats(admin, 2)
|
|
kw = admin.post(f"/api/categories/{a['id']}/keywords", json={"word": "lidil"}).json()
|
|
|
|
r = admin.patch(f"/api/categories/keywords/{kw['id']}", json={"word": " LIDL "})
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["word"] == "lidl" # getrimmt + kleingeschrieben
|
|
words = admin.get(f"/api/categories/{a['id']}/keywords").json()
|
|
assert next(k for k in words if k["id"] == kw["id"])["word"] == "lidl"
|
|
assert "lidil" not in [k["word"] for k in words] # altes Wort ist weg
|
|
|
|
# Umbenennen auf ein Wort, das schon einer anderen Kategorie gehört -> Konflikt.
|
|
admin.post(f"/api/categories/{b['id']}/keywords", json={"word": "coop"})
|
|
assert admin.patch(f"/api/categories/keywords/{kw['id']}", json={"word": "coop"}).status_code == 409
|
|
|
|
|
|
def test_below_threshold_not_flagged(admin):
|
|
giro = _account(admin, name="Giro NoFlag")
|
|
a, b = _expense_cats(admin, 2)
|
|
kw = admin.post(f"/api/categories/{a['id']}/keywords", json={"word": "denner"}).json()
|
|
|
|
lines = "".join(f"0{i}.03.2026;-4,00;DENNER;Einkauf\n" for i in range(1, 7)) # 6 Zeilen
|
|
rows = _preview_csv(admin, giro["id"], "Datum;Betrag;Empfaenger;Zweck\n" + lines, COLMAP)
|
|
rows[0]["category_id"] = b["id"] # nur 1 von 6 korrigiert (< 1/3)
|
|
admin.post("/api/imports/commit", json={"account_id": giro["id"], "filename": "d.csv", "rows": rows})
|
|
|
|
assert kw["id"] not in [f["id"] for f in admin.get("/api/categories/keywords/flagged").json()]
|
|
|
|
|
|
def test_amount_condition_rule(admin):
|
|
giro = _account(admin, name="Giro Amt")
|
|
a, _ = _expense_cats(admin, 2)
|
|
# „sv" nur bei genau 5.00 -> Kategorie a
|
|
r = admin.post(f"/api/categories/{a['id']}/keywords", json={"word": "sv", "amount_cents": 500})
|
|
assert r.status_code == 201, r.text
|
|
assert r.json()["amount_cents"] == 500
|
|
|
|
csv = (
|
|
"Datum;Betrag;Empfaenger;Zweck\n"
|
|
"01.03.2026;-5,00;SV Restaurant;Mittag\n" # 5.00 -> a
|
|
"02.03.2026;-8,00;SV Restaurant;Produkt\n" # anderer Betrag -> keine Regel
|
|
)
|
|
rows = _preview_csv(admin, giro["id"], csv, COLMAP)
|
|
assert rows[0]["category_id"] == a["id"]
|
|
assert rows[1]["category_id"] is None
|
|
|
|
|
|
def test_transfer_rule_and_target_guess(admin):
|
|
giro = _account(admin, name="Giro TR")
|
|
spar = _account(admin, name="Sparkonto TR", type="sparkonto",
|
|
identifiers=[{"kind": "text", "value": "Sparen"}])
|
|
assert any(i["kind"] == "text" for i in spar["identifiers"])
|
|
|
|
r = admin.post("/api/categories/transfer-keywords", json={"word": "uebertrag"})
|
|
assert r.status_code == 201, r.text
|
|
|
|
csv = (
|
|
"Datum;Betrag;Empfaenger;Zweck\n"
|
|
"01.03.2026;-100,00;Max;Uebertrag Sparen\n" # -> Transfer, Ziel = Sparkonto
|
|
"02.03.2026;-50,00;Max;Uebertrag woandershin\n" # -> Transfer, kein Ziel erkannt
|
|
)
|
|
rows = _preview_csv(admin, giro["id"], csv, COLMAP)
|
|
assert rows[0]["type"] == "transfer"
|
|
assert rows[0]["to_account_id"] == spar["id"]
|
|
assert rows[1]["type"] == "transfer"
|
|
assert rows[1]["to_account_id"] is None
|
|
|
|
|
|
def test_transfer_keyword_conflicts_with_category_keyword(admin):
|
|
a, _ = _expense_cats(admin, 2)
|
|
admin.post(f"/api/categories/{a['id']}/keywords", json={"word": "doppelwort"})
|
|
# gleiches Wort als Transfer-Regel -> Konflikt
|
|
assert admin.post("/api/categories/transfer-keywords", json={"word": "doppelwort"}).status_code == 409
|
|
|
|
|
|
def test_no_retroactive_apply(admin):
|
|
giro = _account(admin, name="Giro Retro")
|
|
a, _ = _expense_cats(admin, 2)
|
|
rows = _preview_csv(admin, giro["id"], "Datum;Betrag;Empfaenger;Zweck\n01.03.2026;-5,00;ALDI;Einkauf\n", COLMAP)
|
|
assert rows[0]["category_id"] is None
|
|
admin.post("/api/imports/commit", json={"account_id": giro["id"], "filename": "aldi.csv", "rows": rows})
|
|
|
|
admin.post(f"/api/categories/{a['id']}/keywords", json={"word": "aldi"}) # erst danach angelegt
|
|
|
|
txns = admin.get("/api/transactions").json()
|
|
aldi = [t for t in txns if (t.get("payee") or "").upper() == "ALDI"]
|
|
assert aldi and aldi[0]["category_id"] is None # bestehende Buchung bleibt unkategorisiert
|