516 lines
24 KiB
Python
516 lines
24 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
|
||
import secrets
|
||
|
||
|
||
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_import_verbindet_keine_konten_mehr(admin):
|
||
"""Der Import erzeugt keine Überträge – weder über Schlagwort noch über
|
||
Erkennungs-Text, weder ausgehend noch eingehend.
|
||
|
||
Das Raten des Gegenkontos hat mehr kaputt gemacht als geholfen: falsch
|
||
geratene Ziele, verdrehte Richtungen und doppelt gebuchte Bewegungen,
|
||
sobald beide Auszüge eingelesen wurden. Jede Zeile ist jetzt schlicht das,
|
||
was im Auszug steht.
|
||
"""
|
||
giro = _account(admin, name="Giro TR")
|
||
spar = _account(admin, name="Sparkonto TR", type="sparkonto",
|
||
identifiers=[{"kind": "text", "value": "Sparen"}])
|
||
admin.post("/api/categories/transfer-keywords", json={"word": "uebertrag"})
|
||
admin.post("/api/categories/transfer-keywords",
|
||
json={"word": "revolut", "target_account_id": spar["id"]})
|
||
|
||
csv = (
|
||
"Datum;Betrag;Empfaenger;Zweck\n"
|
||
"01.03.2026;-100,00;Max;Uebertrag Sparen\n" # Schlagwort + Erkennungs-Text
|
||
"02.03.2026;-25,00;Einkauf REVOLUT;Karte\n" # Schlagwort mit festem Ziel
|
||
"03.03.2026;120,00;Gutschrift;von Sparen\n" # Eingang mit Erkennungs-Text
|
||
)
|
||
rows = _preview_csv(admin, giro["id"], csv, COLMAP)
|
||
assert [r["type"] for r in rows] == ["expense", "expense", "income"]
|
||
assert all(r["to_account_id"] is None for r in rows)
|
||
assert all(r["type"] != "transfer" for r in rows)
|
||
|
||
|
||
def test_beide_auszuege_werden_beide_gebucht(admin):
|
||
"""Lies man beide Seiten einer Bewegung ein, zählt jede auf ihrem Konto.
|
||
|
||
Früher galt die zweite Zeile als Gegenbuchung und blieb ungebucht – was
|
||
schiefging, sobald die Erkennung danebenlag. Jetzt bucht jedes Konto das,
|
||
was in seinem eigenen Auszug steht, und beide Salden stimmen.
|
||
"""
|
||
a = _account(admin, name="Konto A CP", opening_balance_cents=100000,
|
||
identifiers=[{"kind": "text", "value": "KontoA"}])
|
||
b = _account(admin, name="Konto B CP", identifiers=[{"kind": "text", "value": "KontoB"}])
|
||
|
||
rowsA = _preview_csv(admin, a["id"],
|
||
"Datum;Betrag;Empfaenger;Zweck\n05.03.2026;-100,00;Uebertrag;an KontoB\n", COLMAP)
|
||
assert rowsA[0]["type"] == "expense"
|
||
r = admin.post("/api/imports/commit", json={"account_id": a["id"], "filename": "a.csv", "rows": rowsA})
|
||
assert r.json()["imported_count"] == 1
|
||
|
||
rowsB = _preview_csv(admin, b["id"],
|
||
"Datum;Betrag;Empfaenger;Zweck\n07.03.2026;100,00;Gutschrift;von KontoA\n", COLMAP)
|
||
assert rowsB[0]["type"] == "income"
|
||
assert rowsB[0]["type"] != "transfer"
|
||
assert rowsB[0]["include"] is True
|
||
assert rowsB[0]["include"] is True
|
||
r = admin.post("/api/imports/commit", json={"account_id": b["id"], "filename": "b.csv", "rows": rowsB})
|
||
assert r.json()["imported_count"] == 1
|
||
|
||
assert admin.get(f"/api/accounts/{a['id']}").json()["balance_cents"] == 90000
|
||
assert admin.get(f"/api/accounts/{b['id']}").json()["balance_cents"] == 10000
|
||
|
||
|
||
def test_same_word_across_kinds_allowed(admin):
|
||
a, b = _expense_cats(admin, 2)
|
||
inc = _income_cat(admin)
|
||
# Dasselbe Wort für eine AUSGABE- und eine EINNAHME-Kategorie -> beides erlaubt.
|
||
r1 = admin.post(f"/api/categories/{a['id']}/keywords", json={"word": "querart"})
|
||
assert r1.status_code == 201, r1.text
|
||
r2 = admin.post(f"/api/categories/{inc['id']}/keywords", json={"word": "querart"})
|
||
assert r2.status_code == 201, r2.text
|
||
assert r1.json()["id"] != r2.json()["id"]
|
||
# ... und auch als Transfer-Regel (dritte Art) erlaubt.
|
||
assert admin.post("/api/categories/transfer-keywords", json={"word": "querart"}).status_code == 201
|
||
# Aber ein zweites Ausgabe-Wort gleicher Art bleibt ein Konflikt.
|
||
assert admin.post(f"/api/categories/{b['id']}/keywords", json={"word": "querart"}).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
|
||
|
||
|
||
def test_same_word_multiple_amounts(admin):
|
||
"""Ein Wort darf mehrere Preise haben – im Personalrestaurant zahlt man mal
|
||
5.00, mal 7.50. Ohne Betrag bleibt es der Auffang für alle übrigen."""
|
||
giro = _account(admin, name="Giro Preise")
|
||
mensa, buero = _expense_cats(admin, 2)
|
||
|
||
# Zwei Preise fürs Restaurant …
|
||
for cents in (500, 750):
|
||
r = admin.post(f"/api/categories/{mensa['id']}/keywords",
|
||
json={"word": "sv gruppe", "amount_cents": cents})
|
||
assert r.status_code == 201, r.text
|
||
# … und derselbe Text ohne Betrag als Auffang in einer anderen Kategorie.
|
||
r = admin.post(f"/api/categories/{buero['id']}/keywords", json={"word": "sv gruppe"})
|
||
assert r.status_code == 201, r.text
|
||
|
||
worte = admin.get(f"/api/categories/{mensa['id']}/keywords").json()
|
||
betraege = sorted(w["amount_cents"] for w in worte if w["word"] == "sv gruppe")
|
||
assert betraege == [500, 750]
|
||
|
||
csv = (
|
||
"Datum;Betrag;Empfaenger;Zweck\n"
|
||
"01.03.2026;-5,00;SV GRUPPE;Mittag\n"
|
||
"02.03.2026;-7,50;SV GRUPPE;Mittag gross\n"
|
||
"03.03.2026;-24,00;SV GRUPPE;Material\n"
|
||
)
|
||
rows = _preview_csv(admin, giro["id"], csv, COLMAP)
|
||
assert rows[0]["category_id"] == mensa["id"], "5.00 -> Restaurant"
|
||
assert rows[1]["category_id"] == mensa["id"], "7.50 -> Restaurant"
|
||
assert rows[2]["category_id"] == buero["id"], "anderer Betrag -> Auffang"
|
||
|
||
# Derselbe Preis zweimal bleibt ein Konflikt bzw. liefert dieselbe Regel.
|
||
doppelt = admin.post(f"/api/categories/{mensa['id']}/keywords",
|
||
json={"word": "sv gruppe", "amount_cents": 500})
|
||
assert doppelt.status_code == 201
|
||
assert len(admin.get(f"/api/categories/{mensa['id']}/keywords").json()) == len(worte)
|
||
|
||
|
||
def _kto(admin, **kw):
|
||
body = {"name": f"K {secrets.token_hex(3)}", "type": "girokonto",
|
||
"currency": "CHF", "opening_balance_cents": 0}
|
||
body.update(kw)
|
||
return admin.post("/api/accounts", json=body).json()
|
||
|
||
|
||
def test_regeln_lassen_sich_nachtraeglich_anwenden(admin):
|
||
"""Regeln greifen sonst nur beim Import. Bei einer Bankanbindung kommen die
|
||
Umsaetze von selbst und die Regel entsteht spaeter – dann bleibt alles
|
||
liegen, was vorher schon da war."""
|
||
quelle = _kto(admin, name=f"Revolut {secrets.token_hex(3)}")
|
||
pocket = _kto(admin, name=f"Sparen Pocket {secrets.token_hex(3)}")
|
||
kennung = f"to chf mb:{secrets.token_hex(6)}"
|
||
|
||
# Drei Aufrundungen, wie Revolut sie liefert – ohne Kategorie.
|
||
ids = []
|
||
for cent in (100, 50, 34):
|
||
ids.append(admin.post("/api/transactions", json={
|
||
"type": "expense", "date": "2026-08-25", "account_id": quelle["id"],
|
||
"amount_cents": cent, "payee": kennung.upper()}).json()["id"])
|
||
|
||
r = admin.post("/api/categories/transfer-keywords", json={
|
||
"word": kennung, "target_account_id": pocket["id"]})
|
||
assert r.status_code == 201, r.text
|
||
|
||
# Erst zeigen, nichts tun. Die Zahlen gelten fuer ALLE offenen Buchungen
|
||
# des Nutzers, deshalb wird hier nur geprueft, dass nichts passiert ist.
|
||
vor = admin.post("/api/categories/keywords/anwenden", params={"vorschau": True}).json()
|
||
assert vor["uebertraege"] >= 3
|
||
assert admin.get(f"/api/accounts/{pocket['id']}").json()["balance_cents"] == 0
|
||
for txn_id in ids:
|
||
assert admin.get(f"/api/transactions/{txn_id}").json()["type"] == "expense"
|
||
|
||
# Dann wirklich.
|
||
admin.post("/api/categories/keywords/anwenden", params={"vorschau": False})
|
||
assert admin.get(f"/api/accounts/{pocket['id']}").json()["balance_cents"] == 184
|
||
assert admin.get(f"/api/accounts/{quelle['id']}").json()["balance_cents"] == -184
|
||
for txn_id in ids:
|
||
assert admin.get(f"/api/transactions/{txn_id}").json()["type"] == "transfer"
|
||
|
||
# Ein zweiter Lauf ruehrt sie nicht noch einmal an: Ein Uebertrag ist kein
|
||
# Kandidat mehr, sonst verschoebe jeder Klick das Geld erneut.
|
||
admin.post("/api/categories/keywords/anwenden", params={"vorschau": False})
|
||
assert admin.get(f"/api/accounts/{pocket['id']}").json()["balance_cents"] == 184
|
||
|
||
|
||
def test_nachtraegliches_anwenden_ruehrt_kategorisiertes_nicht_an(admin):
|
||
"""Nur Buchungen OHNE Kategorie – sonst überschriebe der Knopf stillschweigend
|
||
das, was man von Hand zugeordnet hat."""
|
||
konto = _kto(admin)
|
||
gruppen = admin.get("/api/categories").json()
|
||
kat = next(c["id"] for g in gruppen if g["kind"] == "expense"
|
||
for c in g["categories"])
|
||
wort = f"marke{secrets.token_hex(4)}"
|
||
txn = admin.post("/api/transactions", json={
|
||
"type": "expense", "date": "2026-08-25", "account_id": konto["id"],
|
||
"amount_cents": 999, "payee": wort, "category_id": kat}).json()
|
||
|
||
andere = next(c["id"] for g in gruppen if g["kind"] == "expense"
|
||
for c in g["categories"] if c["id"] != kat)
|
||
admin.post(f"/api/categories/{andere}/keywords",
|
||
json={"pattern": wort, "match_field": "any"})
|
||
|
||
admin.post("/api/categories/keywords/anwenden", params={"vorschau": False})
|
||
assert admin.get(f"/api/transactions/{txn['id']}").json()["category_id"] == kat
|
||
|
||
|
||
def test_verknuepfte_buchungen_landen_nicht_in_uebersprungen(admin):
|
||
"""Eine verknuepfte Buchung ist erledigt – beide Seiten stehen, beide zaehlen
|
||
in keiner Auswertung. Sie passt aber weiter auf die Uebertrags-Regel und
|
||
wurde deshalb bei JEDEM Lauf neu abgelehnt: Die Uebersprungen-Liste fuellte
|
||
sich mit Hunderten Zeilen, die nie verschwinden, und verdeckte die
|
||
wirklich offenen Buchungen."""
|
||
quelle = _kto(admin, name=f"Revolut {secrets.token_hex(3)}")
|
||
pocket = _kto(admin, name=f"Sparen Pocket {secrets.token_hex(3)}")
|
||
kennung = f"um chf sparen {secrets.token_hex(6)}"
|
||
betrag = 60 + secrets.randbelow(900)
|
||
|
||
raus = admin.post("/api/transactions", json={
|
||
"type": "expense", "date": "2026-01-22", "account_id": quelle["id"],
|
||
"amount_cents": betrag, "payee": kennung.upper()}).json()
|
||
rein = admin.post("/api/transactions", json={
|
||
"type": "income", "date": "2026-01-22", "account_id": pocket["id"],
|
||
"amount_cents": betrag, "payee": kennung.upper()}).json()
|
||
|
||
r = admin.post("/api/categories/transfer-keywords", json={
|
||
"word": kennung, "target_account_id": pocket["id"]})
|
||
assert r.status_code == 201, r.text
|
||
|
||
# Beide Seiten sind dieselbe Bewegung – verbinden, nicht uebertragen.
|
||
assert admin.post("/api/uebertraege/verbinden",
|
||
json={"a_id": raus["id"], "b_id": rein["id"]}).status_code == 200
|
||
|
||
ergebnis = admin.post("/api/categories/keywords/anwenden",
|
||
params={"vorschau": False}).json()
|
||
texte = [z["text"] for z in ergebnis["uebersprungen"]]
|
||
assert kennung.upper() not in texte, "verknuepfte Buchung gehoert nicht in die Liste"
|
||
|
||
# Und sie bleibt unangetastet: kein Uebertrag, kein verschobenes Geld.
|
||
assert admin.get(f"/api/transactions/{raus['id']}").json()["type"] == "expense"
|
||
assert admin.get(f"/api/accounts/{pocket['id']}").json()["balance_cents"] == betrag
|
||
|
||
|
||
def _tx(admin, konto, typ, datum, cents, payee):
|
||
return admin.post("/api/transactions", json={
|
||
"type": typ, "date": datum, "account_id": konto["id"],
|
||
"amount_cents": cents, "payee": payee}).json()
|
||
|
||
|
||
def test_vergebene_gegenseite_wird_als_solche_benannt(admin):
|
||
"""Die Pruefung fragt „gibt es auf dem Zielkonto eine passende Buchung?",
|
||
nicht „ist sie noch frei?". Haengt die gefundene schon an einer anderen
|
||
Buchung, ist sie NICHT die Gegenseite dieser hier – und der Rat „verbinden"
|
||
geht ins Leere, denn die Vorschlagsliste laesst vergebene Buchungen aus.
|
||
Dann muss die Meldung sagen, was wirklich los ist."""
|
||
quelle = _kto(admin, name=f"Revolut {secrets.token_hex(3)}")
|
||
pocket = _kto(admin, name=f"Sparen Pocket {secrets.token_hex(3)}")
|
||
anderes = _kto(admin, name=f"Anderes {secrets.token_hex(3)}")
|
||
kennung = f"um chf sparen {secrets.token_hex(6)}"
|
||
betrag = 62 + secrets.randbelow(900)
|
||
|
||
# Der Zugang im Pocket gehoert schon zu einer Buchung vom dritten Konto.
|
||
zugang = _tx(admin, pocket, "income", "2026-05-20", betrag, "Pocket")
|
||
fremd = _tx(admin, anderes, "expense", "2026-05-20", betrag, "Fremd")
|
||
assert admin.post("/api/uebertraege/verbinden",
|
||
json={"a_id": fremd["id"], "b_id": zugang["id"]}).status_code == 200
|
||
|
||
# Und jetzt kommt die Revolut-Ausgabe, gleicher Tag, gleicher Betrag.
|
||
raus = _tx(admin, quelle, "expense", "2026-05-20", betrag, kennung.upper())
|
||
assert admin.post("/api/categories/transfer-keywords", json={
|
||
"word": kennung, "target_account_id": pocket["id"]}).status_code == 201
|
||
|
||
r = admin.post("/api/categories/keywords/anwenden", params={"vorschau": False}).json()
|
||
grund = next(z["grund"] for z in r["uebersprungen"] if z["text"] == kennung.upper())
|
||
assert "schon mit einer anderen Buchung verbunden" in grund
|
||
assert "Richtig ist hier, die beiden zu verbinden" not in grund, \
|
||
"zum Verbinden raten waere hier falsch – die Gegenseite ist vergeben"
|
||
|
||
# Geblockt wird weiter: kein Uebertrag, kein verschobenes Geld.
|
||
assert admin.get(f"/api/transactions/{raus['id']}").json()["type"] == "expense"
|
||
assert admin.get(f"/api/accounts/{pocket['id']}").json()["balance_cents"] == betrag
|
||
|
||
|
||
def test_freie_gegenseite_raet_weiter_zum_verbinden(admin):
|
||
"""Die Gegenprobe: Ist die Gegenseite noch frei, bleibt es beim alten Rat.
|
||
Sonst schluckte der neue Zweig alle Faelle."""
|
||
quelle = _kto(admin, name=f"Revolut {secrets.token_hex(3)}")
|
||
pocket = _kto(admin, name=f"Sparen Pocket {secrets.token_hex(3)}")
|
||
kennung = f"um chf sparen {secrets.token_hex(6)}"
|
||
betrag = 62 + secrets.randbelow(900)
|
||
|
||
_tx(admin, pocket, "income", "2026-05-20", betrag, "Pocket")
|
||
_tx(admin, quelle, "expense", "2026-05-20", betrag, kennung.upper())
|
||
assert admin.post("/api/categories/transfer-keywords", json={
|
||
"word": kennung, "target_account_id": pocket["id"]}).status_code == 201
|
||
|
||
r = admin.post("/api/categories/keywords/anwenden", params={"vorschau": False}).json()
|
||
grund = next(z["grund"] for z in r["uebersprungen"] if z["text"] == kennung.upper())
|
||
assert "Richtig ist hier, die beiden zu verbinden" in grund
|