Files
certain/backend/tests/test_abo_erkennung.py
T
ScarriffleandClaude Opus 5 152483707e feat: Zahlungen finden ihren Vertrag von selbst – Preisaenderungen werden gemeldet
Ein Abo stand bisher unverbunden neben seinen Zahlungen. Wer wissen wollte,
ob Netflix diesen Monat abgebucht wurde, musste es in der Buchungsliste
suchen. Jetzt verknuepft Certain Buchung und Vertrag beim Import selbst.

Drei Stufen, bewusst unterschiedlich mutig:

- sicher (Betrag und Text passen, Datum im Fenster): wird zugeordnet.
- Vorschlag: Betrag passt, Text nicht – oder der Text passt und der Betrag
  weicht ab. Der zweite Fall ist die Preiserhoehung.
- nichts: im Zweifel lieber schweigen.

Eine Preiserhoehung wird nie stillschweigend uebernommen. Sie ist eine
Nachricht an den Nutzer, kein Wartungsvorgang – er entscheidet, ob der
neue Preis der richtige ist.

Die Zuordnung ist ein Etikett, keine Umbuchung: die Buchung selbst bleibt
unangetastet, und ein geloeschter Vertrag nimmt sie nicht mit.

Beim Backup wandern Vertraege nun vor die Buchungen, sonst haenge eine
Rueckspielung Zahlungen an die falschen Vertraege.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-30 20:16:13 +02:00

332 lines
13 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Welche Buchung gehoert zu welchem Vertrag.
Die Zuordnung ist ein ETIKETT: An der Buchung aendert sich nichts. Die Tests
pruefen deshalb vor allem, was gleich bleiben MUSS – und wo das Programm
lieber fragt, statt zu raten.
"""
import secrets
from datetime import date, timedelta
def _konto(admin, name=None, currency="EUR"):
r = admin.post("/api/accounts", json={
"name": name or ("Konto " + secrets.token_hex(3)), "type": "girokonto",
"currency": currency, "opening_balance_cents": 0})
assert r.status_code == 201, r.text
return r.json()
def _abo(admin, **kw):
body = {"name": "Netflix " + secrets.token_hex(3), "amount_cents": 1990,
"currency": "EUR", "interval_months": 1}
body.update(kw)
r = admin.post("/api/subscriptions", json=body)
assert r.status_code == 201, r.text
return r.json()
def _buchung(admin, konto, betrag, datum, payee="", beschreibung=""):
r = admin.post("/api/transactions", json={
"type": "expense", "date": datum, "account_id": konto["id"],
"amount_cents": betrag, "payee": payee, "description": beschreibung})
assert r.status_code == 201, r.text
return r.json()
def _zuordnen(admin):
r = admin.post("/api/subscriptions/zuordnen")
assert r.status_code == 200, r.text
return r.json()
def _txn(admin, txn_id):
for t in admin.get("/api/transactions").json():
if t["id"] == txn_id:
return t
raise AssertionError("Buchung verschwunden")
def _aufraeumen(admin, abo=None, txns=()):
for t in txns:
admin.delete(f"/api/transactions/{t}")
if abo:
admin.delete(f"/api/subscriptions/{abo['id']}")
# --- Der Normalfall ----------------------------------------------------------
def test_abbuchung_findet_ihren_vertrag(admin):
konto = _konto(admin)
abo = _abo(admin, name="Netflix", amount_cents=1990, account_id=konto["id"],
next_due=str(date.today()))
b = _buchung(admin, konto, 1990, str(date.today()), payee="NETFLIX.COM 4711")
try:
assert _zuordnen(admin)["zugeordnet"] >= 1
verlauf = admin.get(f"/api/subscriptions/{abo['id']}/verlauf").json()
assert verlauf["anzahl"] == 1
assert verlauf["summe_cents"] == 1990
assert verlauf["buchungen"][0]["txn_id"] == b["id"]
finally:
_aufraeumen(admin, abo, [b["id"]])
def test_die_buchung_bleibt_unveraendert(admin):
"""Es ist ein Etikett. Betrag, Konto und Kategorie bleiben, wie sie sind."""
konto = _konto(admin)
abo = _abo(admin, name="Spotify", amount_cents=1299, account_id=konto["id"],
next_due=str(date.today()))
b = _buchung(admin, konto, 1299, str(date.today()), payee="SPOTIFY AB")
try:
_zuordnen(admin)
danach = _txn(admin, b["id"])
assert danach["amount_cents"] == 1299
assert danach["account_id"] == konto["id"]
assert danach["date"] == b["date"]
assert danach["type"] == "expense"
finally:
_aufraeumen(admin, abo, [b["id"]])
def test_faelligkeit_rueckt_nach_der_erkannten_abbuchung_weiter(admin):
konto = _konto(admin)
heute = date.today()
abo = _abo(admin, name="Netflix", amount_cents=1990, account_id=konto["id"],
next_due=str(heute), interval_months=1)
b = _buchung(admin, konto, 1990, str(heute), payee="NETFLIX")
try:
_zuordnen(admin)
danach = admin.get(f"/api/subscriptions/{abo['id']}").json()
assert danach["next_due"] is not None
assert danach["next_due"] > str(heute), "Faelligkeit blieb stehen"
finally:
_aufraeumen(admin, abo, [b["id"]])
def test_jede_abbuchung_wird_einzeln_erkannt(admin):
"""Drei Monate nachgetragen ergeben drei Zeilen im Verlauf, nicht eine."""
konto = _konto(admin)
heute = date.today()
abo = _abo(admin, name="Netflix", amount_cents=1990, account_id=konto["id"],
next_due=str(heute), interval_months=1)
tage = [heute - timedelta(days=62), heute - timedelta(days=31), heute]
ids = [_buchung(admin, konto, 1990, str(t), payee="NETFLIX")["id"] for t in tage]
try:
_zuordnen(admin)
verlauf = admin.get(f"/api/subscriptions/{abo['id']}/verlauf").json()
assert verlauf["anzahl"] == 3, verlauf
assert verlauf["summe_cents"] == 3 * 1990
finally:
_aufraeumen(admin, abo, ids)
# --- Wo das Programm lieber fragt --------------------------------------------
def test_fremder_betrag_wird_nicht_zugeordnet(admin):
"""Lieber nichts sagen als daneben raten."""
konto = _konto(admin)
abo = _abo(admin, name="Netflix", amount_cents=1990, account_id=konto["id"],
next_due=str(date.today()))
b = _buchung(admin, konto, 7350, str(date.today()), payee="BAUMARKT")
try:
_zuordnen(admin)
assert _txn(admin, b["id"]).get("subscription_id") in (None, 0)
assert admin.get(f"/api/subscriptions/{abo['id']}/verlauf").json()["anzahl"] == 0
finally:
_aufraeumen(admin, abo, [b["id"]])
def test_richtiger_betrag_fremder_text_kommt_als_vorschlag(admin):
"""Auf dem Auszug steht oft der Abwickler, nicht die Marke. Fragen statt
raten – aber die Frage muss gestellt werden."""
konto = _konto(admin)
abo = _abo(admin, name="Netflix", amount_cents=1990, account_id=konto["id"],
next_due=str(date.today()))
b = _buchung(admin, konto, 1990, str(date.today()), payee="STRIPE PAYMENTS")
try:
antwort = _zuordnen(admin)
assert _txn(admin, b["id"]).get("subscription_id") in (None, 0), \
"ungefragt zugeordnet"
passend = [v for v in antwort["vorschlaege"] if v["txn_id"] == b["id"]]
assert passend, antwort["vorschlaege"]
assert passend[0]["subscription_id"] == abo["id"]
finally:
_aufraeumen(admin, abo, [b["id"]])
def test_buchung_auf_fremdem_konto_zaehlt_nicht(admin):
"""Der Vertrag sagt, ueber welches Konto er laeuft – das gilt."""
konto = _konto(admin)
anderes = _konto(admin)
abo = _abo(admin, name="Netflix", amount_cents=1990, account_id=konto["id"],
next_due=str(date.today()))
b = _buchung(admin, anderes, 1990, str(date.today()), payee="NETFLIX")
try:
_zuordnen(admin)
assert _txn(admin, b["id"]).get("subscription_id") in (None, 0)
finally:
_aufraeumen(admin, abo, [b["id"]])
def test_gekuendigte_vertraege_ziehen_nichts_mehr_an(admin):
konto = _konto(admin)
abo = _abo(admin, name="Altes Abo", amount_cents=1990, account_id=konto["id"],
next_due=str(date.today()), is_active=False)
b = _buchung(admin, konto, 1990, str(date.today()), payee="ALTES ABO")
try:
_zuordnen(admin)
assert _txn(admin, b["id"]).get("subscription_id") in (None, 0)
finally:
_aufraeumen(admin, abo, [b["id"]])
# --- Die Preiserhoehung ------------------------------------------------------
def test_preiserhoehung_wird_gemeldet_statt_verbucht(admin):
"""Der wertvollste Fall – und einer, den man nie automatisch nachziehen
darf. Eine Preiserhoehung ist eine Nachricht an den Nutzer."""
konto = _konto(admin)
abo = _abo(admin, name="Netflix", amount_cents=1790, account_id=konto["id"],
next_due=str(date.today()))
b = _buchung(admin, konto, 1990, str(date.today()), payee="NETFLIX.COM")
try:
antwort = _zuordnen(admin)
meldungen = [p for p in antwort["preisaenderungen"] if p["txn_id"] == b["id"]]
assert meldungen, antwort
assert meldungen[0]["erwartet_cents"] == 1790
assert meldungen[0]["amount_cents"] == 1990
# Ungefragt geaendert hat sich nichts.
assert admin.get(f"/api/subscriptions/{abo['id']}").json()["amount_cents"] == 1790
assert _txn(admin, b["id"]).get("subscription_id") in (None, 0)
finally:
_aufraeumen(admin, abo, [b["id"]])
def test_neuen_preis_uebernehmen_und_zuordnen(admin):
"""Der Weg, den der Nutzer danach geht."""
konto = _konto(admin)
abo = _abo(admin, name="Netflix", amount_cents=1790, account_id=konto["id"],
next_due=str(date.today()))
b = _buchung(admin, konto, 1990, str(date.today()), payee="NETFLIX.COM")
try:
r = admin.post(f"/api/subscriptions/{abo['id']}/preis",
params={"amount_cents": 1990})
assert r.status_code == 200, r.text
assert r.json()["amount_cents"] == 1990
r = admin.post(f"/api/subscriptions/{abo['id']}/buchungen/{b['id']}")
assert r.status_code == 200, r.text
assert admin.get(f"/api/subscriptions/{abo['id']}/verlauf").json()["anzahl"] == 1
finally:
_aufraeumen(admin, abo, [b["id"]])
def test_ein_vielfaches_gilt_nicht_als_preisaenderung(admin):
"""Ein Abo, das sich verdreifacht, ist kein Abo mehr, sondern etwas
anderes – sonst zieht jeder groessere Einkauf beim selben Haendler an."""
konto = _konto(admin)
abo = _abo(admin, name="Netflix", amount_cents=1990, account_id=konto["id"],
next_due=str(date.today()))
b = _buchung(admin, konto, 19900, str(date.today()), payee="NETFLIX.COM")
try:
antwort = _zuordnen(admin)
assert not [p for p in antwort["preisaenderungen"] if p["txn_id"] == b["id"]]
assert not [v for v in antwort["vorschlaege"] if v["txn_id"] == b["id"]]
finally:
_aufraeumen(admin, abo, [b["id"]])
# --- Das Etikett wieder abziehen ---------------------------------------------
def test_zuordnung_laesst_sich_loesen_und_die_buchung_bleibt(admin):
konto = _konto(admin)
abo = _abo(admin, name="Netflix", amount_cents=1990, account_id=konto["id"],
next_due=str(date.today()))
b = _buchung(admin, konto, 1990, str(date.today()), payee="NETFLIX")
try:
_zuordnen(admin)
assert admin.get(f"/api/subscriptions/{abo['id']}/verlauf").json()["anzahl"] == 1
r = admin.delete(f"/api/subscriptions/buchungen/{b['id']}")
assert r.status_code == 204, r.text
assert admin.get(f"/api/subscriptions/{abo['id']}/verlauf").json()["anzahl"] == 0
# Die Buchung selbst steht unveraendert da.
assert _txn(admin, b["id"])["amount_cents"] == 1990
finally:
_aufraeumen(admin, abo, [b["id"]])
def test_geloeschter_vertrag_nimmt_die_buchung_nicht_mit(admin):
"""Wer ein Abo kuendigt und loescht, will nicht seine Zahlungshistorie
mitloeschen."""
konto = _konto(admin)
abo = _abo(admin, name="Kurzlebig", amount_cents=1990, account_id=konto["id"],
next_due=str(date.today()))
b = _buchung(admin, konto, 1990, str(date.today()), payee="KURZLEBIG")
try:
_zuordnen(admin)
assert admin.delete(f"/api/subscriptions/{abo['id']}").status_code == 204
danach = _txn(admin, b["id"])
assert danach["amount_cents"] == 1990, "Buchung mitgeloescht"
finally:
admin.delete(f"/api/transactions/{b['id']}")
def test_vorschlaege_veraendern_nichts(admin):
"""Eine Ansicht, die bloss hinsieht, darf nichts zuordnen."""
konto = _konto(admin)
abo = _abo(admin, name="Netflix", amount_cents=1990, account_id=konto["id"],
next_due=str(date.today()))
b = _buchung(admin, konto, 1990, str(date.today()), payee="NETFLIX")
try:
r = admin.get("/api/subscriptions/vorschlaege")
assert r.status_code == 200, r.text
assert r.json()["zugeordnet"] == 0
assert _txn(admin, b["id"]).get("subscription_id") in (None, 0)
finally:
_aufraeumen(admin, abo, [b["id"]])
def test_zuordnung_ueberlebt_sichern_und_einspielen(admin):
"""Ohne den Verweis in der Sicherung zeigte das Etikett danach auf einen
FREMDEN Vertrag – die Nummern werden beim Einspielen neu vergeben."""
import json
konto = _konto(admin)
name = "Backup-Abo " + secrets.token_hex(3)
abo = _abo(admin, name=name, amount_cents=1990, account_id=konto["id"],
next_due=str(date.today()))
b = _buchung(admin, konto, 1990, str(date.today()), payee=name)
try:
_zuordnen(admin)
assert _txn(admin, b["id"])["subscription_id"] == abo["id"]
roh = admin.get("/api/export/backup.json").content
haufen = json.loads(roh)
haufen = haufen.get("data", haufen)
# Der Verweis muss ueberhaupt in der Sicherung stehen.
gesichert = [t for t in haufen["transactions"]
if t.get("subscription_id") is not None]
assert gesichert, "kein einziges Etikett gesichert"
r = admin.post("/api/backup/restore",
files={"file": ("backup.json", roh, "application/json")},
data={"mode": "replace"})
assert r.status_code == 200, r.text
# Nach dem Einspielen sind die Nummern andere – die Zuordnung muss
# trotzdem auf DENSELBEN Vertrag zeigen.
abos = admin.get("/api/subscriptions").json()["subscriptions"]
wieder = next(s for s in abos if s["name"] == name)
verlauf = admin.get(f"/api/subscriptions/{wieder['id']}/verlauf").json()
assert verlauf["anzahl"] == 1, "Zuordnung nach dem Einspielen verloren"
assert verlauf["summe_cents"] == 1990
finally:
pass