Der Schlagwort-Knopf sah bei jeder Kategorie gleich aus - man musste jede einzeln oeffnen, um zu sehen, wo ueberhaupt etwas hinterlegt ist. Belegte Kategorien tragen den Knopf jetzt in Akzentfarbe und mit der Anzahl daneben, der Tooltip nennt sie ausgeschrieben. Die Zahl kommt als keyword_count an der Kategorie mit, ermittelt in einer Aggregat-Abfrage fuer die ganze Liste. Ein Abruf je Kategorie waere dafuer unverhaeltnismaessig gewesen. Gefuellt wird das Feld nur beim Auflisten; beim Anlegen und Aendern einer Gruppe bleibt es bei 0, wo es ohnehin nichts aussagt. Das Popup laedt beim Schliessen die Liste neu, damit die Zahl nach dem Bearbeiten stimmt. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
271 lines
12 KiB
Python
271 lines
12 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_keyword_count_in_category_list(admin):
|
||
"""Die Übersicht markiert Kategorien mit Schlagwörtern – dafür die Anzahl.
|
||
|
||
Gemessen wird relativ: die Testdatenbank lebt über den ganzen Lauf, in den
|
||
Kategorien liegen also schon Wörter aus vorherigen Tests."""
|
||
a, b = _expense_cats(admin, 2)
|
||
|
||
def anzahl(cat_id):
|
||
for g in admin.get("/api/categories").json():
|
||
for c in g["categories"]:
|
||
if c["id"] == cat_id:
|
||
return c["keyword_count"]
|
||
raise AssertionError("Kategorie nicht gefunden")
|
||
|
||
vorher_a, vorher_b = anzahl(a["id"]), anzahl(b["id"])
|
||
# Gegenprobe: die Zahl entspricht dem, was der Einzelabruf liefert.
|
||
assert vorher_a == len(admin.get(f"/api/categories/{a['id']}/keywords").json())
|
||
|
||
kw = admin.post(f"/api/categories/{a['id']}/keywords", json={"word": "zaehltest-eins"}).json()
|
||
admin.post(f"/api/categories/{a['id']}/keywords", json={"word": "zaehltest-zwei"})
|
||
assert anzahl(a["id"]) == vorher_a + 2
|
||
assert anzahl(b["id"]) == vorher_b # andere Kategorie bleibt unberührt
|
||
|
||
admin.delete(f"/api/categories/keywords/{kw['id']}")
|
||
assert anzahl(a["id"]) == vorher_a + 1
|
||
|
||
|
||
def test_keyword_update_amount(admin):
|
||
"""Wort und Betrag gemeinsam ändern – das Popup lädt beides in die Zeile."""
|
||
(a,) = _expense_cats(admin, 1)
|
||
kw = admin.post(
|
||
f"/api/categories/{a['id']}/keywords", json={"word": "sbb", "amount_cents": 500}
|
||
).json()
|
||
assert kw["amount_cents"] == 500
|
||
|
||
r = admin.patch(
|
||
f"/api/categories/keywords/{kw['id']}", json={"word": "sbb", "amount_cents": 750}
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["amount_cents"] == 750
|
||
|
||
# Nur das Wort schicken lässt den Betrag stehen.
|
||
r = admin.patch(f"/api/categories/keywords/{kw['id']}", json={"word": "sbb cff"})
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["amount_cents"] == 750
|
||
|
||
# Ausdrücklich null löscht ihn (geleertes Betragsfeld im Popup).
|
||
r = admin.patch(
|
||
f"/api/categories/keywords/{kw['id']}", json={"word": "sbb cff", "amount_cents": None}
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["amount_cents"] is None
|
||
gespeichert = admin.get(f"/api/categories/{a['id']}/keywords").json()
|
||
assert next(k for k in gespeichert if k["id"] == kw["id"])["amount_cents"] is None
|
||
|
||
# Null ist erlaubt, null-als-Betrag nicht.
|
||
r = admin.patch(
|
||
f"/api/categories/keywords/{kw['id']}", json={"word": "sbb cff", "amount_cents": 0}
|
||
)
|
||
assert r.status_code == 422
|
||
|
||
|
||
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
|