Bisher brauchte es je Konto einen eigenen Kurzbefehl, weil account_id Pflicht war. Jetzt ist sie optional: Raiffeisen fuehrt die eigene IBAN in der Datei mit (Treffer gegen die Erkennungs-Texte der Konten), Revolut die Waehrung – zusammen mit der erkannten Bank und dem Import-Verlauf reicht das. Bleibt es mehrdeutig, wird NICHT geraten: Ein Auszug im falschen Konto verfaelscht zwei Salden auf einmal. Stattdessen nennt die Fehlermeldung die Konten samt Nummern, sodass man sie im Kurzbefehl nachtragen kann. Hat der Schluessel ein Standardkonto, gilt das. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
385 lines
16 KiB
Python
385 lines
16 KiB
Python
"""Schnellerfassung per Gerätemarke (iOS-Kurzbefehl)."""
|
||
|
||
import secrets
|
||
|
||
from fastapi.testclient import TestClient
|
||
|
||
|
||
def _fremd():
|
||
"""Client OHNE Sitzung. Die Fixture ``client`` ist derselbe Client wie
|
||
``admin`` und traegt dessen Anmeldung – damit liesse sich nicht pruefen,
|
||
ob eine Marke fuer sich allein etwas darf."""
|
||
from app.main import app
|
||
|
||
return TestClient(app)
|
||
|
||
|
||
def _konto(admin, **kw):
|
||
body = {"name": "Girokonto", "type": "girokonto", "currency": "CHF",
|
||
"opening_balance_cents": 100000}
|
||
body.update(kw)
|
||
r = admin.post("/api/accounts", json=body)
|
||
assert r.status_code == 201, r.text
|
||
return r.json()
|
||
|
||
|
||
def _marke(admin, konto_id=None, name="iPhone"):
|
||
r = admin.post("/api/inbox/tokens", json={"name": name, "default_account_id": konto_id})
|
||
assert r.status_code == 201, r.text
|
||
return r.json()
|
||
|
||
|
||
def _kategorie(admin, name):
|
||
for g in admin.get("/api/categories").json():
|
||
for c in g["categories"]:
|
||
if c["name"] == name:
|
||
return c["id"]
|
||
raise AssertionError(f"{name} fehlt")
|
||
|
||
|
||
def test_marke_bucht_und_taucht_im_eingang_auf(admin):
|
||
client = _fremd()
|
||
konto = _konto(admin)
|
||
marke = _marke(admin, konto["id"])
|
||
|
||
r = client.post("/api/quick/booking", json={"amount": "12.50", "payee": "Kantine SV"},
|
||
headers={"X-Certain-Token": marke["token"]})
|
||
assert r.status_code == 201, r.text
|
||
assert "12.50 CHF" in r.json()["message"]
|
||
|
||
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000 - 1250
|
||
meine = [e for e in admin.get("/api/inbox").json() if e["payee"] == "Kantine SV"]
|
||
assert len(meine) == 1
|
||
assert meine[0]["status"] == "booked"
|
||
|
||
|
||
def test_marke_darf_nur_erfassen_nichts_lesen(admin):
|
||
"""Die Marke liegt auf dem Telefon. Sie darf keine Daten herausgeben."""
|
||
client = _fremd()
|
||
konto = _konto(admin)
|
||
marke = _marke(admin, konto["id"])
|
||
kopf = {"X-Certain-Token": marke["token"]}
|
||
|
||
for pfad in ("/api/transactions", "/api/accounts", "/api/inbox", "/api/export/all.json", "/api/backup/db", "/api/categories"):
|
||
r = client.get(pfad, headers=kopf)
|
||
assert r.status_code == 401, f"{pfad} gab {r.status_code} statt 401"
|
||
|
||
|
||
def test_unbekannte_oder_fehlende_marke_wird_abgewiesen(admin):
|
||
client = _fremd()
|
||
_konto(admin)
|
||
assert client.post("/api/quick/booking", json={"amount": "5"}).status_code == 401
|
||
r = client.post("/api/quick/booking", json={"amount": "5"},
|
||
headers={"X-Certain-Token": "ausgedacht"})
|
||
assert r.status_code == 401
|
||
|
||
|
||
def test_widerrufene_marke_bucht_nicht_mehr(admin):
|
||
client = _fremd()
|
||
konto = _konto(admin)
|
||
marke = _marke(admin, konto["id"])
|
||
kopf = {"X-Certain-Token": marke["token"]}
|
||
assert client.post("/api/quick/booking", json={"amount": "1"}, headers=kopf).status_code == 201
|
||
|
||
assert admin.delete(f"/api/inbox/tokens/{marke['id']}").status_code == 204
|
||
assert client.post("/api/quick/booking", json={"amount": "1"}, headers=kopf).status_code == 401
|
||
|
||
|
||
def test_klartext_der_marke_ist_nur_einmal_zu_sehen(admin):
|
||
konto = _konto(admin)
|
||
marke = _marke(admin, konto["id"])
|
||
assert marke["token"]
|
||
liste = admin.get("/api/inbox/tokens").json()
|
||
meine = [t for t in liste if t["id"] == marke["id"]]
|
||
assert len(meine) == 1
|
||
assert "token" not in meine[0], "Der Klartext darf nie wieder ausgeliefert werden"
|
||
|
||
|
||
def test_doppelt_ausgeloester_kurzbefehl_bucht_nur_einmal(admin):
|
||
"""Schlechtes Netz oder Doppeltipp dürfen nicht zweimal abbuchen."""
|
||
client = _fremd()
|
||
konto = _konto(admin)
|
||
marke = _marke(admin, konto["id"])
|
||
kopf = {"X-Certain-Token": marke["token"]}
|
||
daten = {"amount": "9.90", "payee": "Kiosk", "external_id": "abc-123"}
|
||
|
||
a = client.post("/api/quick/booking", json=daten, headers=kopf)
|
||
b = client.post("/api/quick/booking", json=daten, headers=kopf)
|
||
assert a.status_code == 201 and b.status_code == 201
|
||
assert b.json()["message"] == "Bereits erfasst"
|
||
assert a.json()["item"]["id"] == b.json()["item"]["id"]
|
||
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000 - 990
|
||
|
||
|
||
def test_schlagwort_schlaegt_die_kategorie_vor(admin):
|
||
"""Was der CSV-Import kann, kann der Kurzbefehl auch."""
|
||
client = _fremd()
|
||
konto = _konto(admin)
|
||
leb = _kategorie(admin, "Lebensmittel")
|
||
# Eindeutiges Wort: Die Testdatenbank lebt über den ganzen Lauf, und
|
||
# Schlagwörter sind je Nutzer eindeutig.
|
||
wort = "laden" + secrets.token_hex(3)
|
||
assert admin.post(f"/api/categories/{leb}/keywords", json={"word": wort}).status_code == 201
|
||
|
||
marke = _marke(admin, konto["id"])
|
||
r = client.post("/api/quick/booking",
|
||
json={"amount": "23.40", "payee": f"{wort.upper()} Bahnhof"},
|
||
headers={"X-Certain-Token": marke["token"]})
|
||
assert r.status_code == 201, r.text
|
||
assert r.json()["item"]["category_id"] == leb
|
||
|
||
|
||
def test_einnahme_und_fremdes_konto(admin):
|
||
client = _fremd()
|
||
konto = _konto(admin)
|
||
zweites = _konto(admin, name="Sparkonto", opening_balance_cents=0)
|
||
marke = _marke(admin, konto["id"])
|
||
kopf = {"X-Certain-Token": marke["token"]}
|
||
|
||
r = client.post("/api/quick/booking",
|
||
json={"amount": "50", "kind": "income", "account_id": zweites["id"]},
|
||
headers=kopf)
|
||
assert r.status_code == 201, r.text
|
||
assert admin.get(f"/api/accounts/{zweites['id']}").json()["balance_cents"] == 5000
|
||
# Das Standardkonto bleibt unberührt.
|
||
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000
|
||
|
||
|
||
def test_verwerfen_loescht_die_buchung_behaelt_aber_die_spur(admin):
|
||
client = _fremd()
|
||
konto = _konto(admin)
|
||
marke = _marke(admin, konto["id"])
|
||
client.post("/api/quick/booking", json={"amount": "7.00", "payee": "Irrtum"},
|
||
headers={"X-Certain-Token": marke["token"]})
|
||
eintrag = admin.get("/api/inbox").json()[0]
|
||
|
||
assert admin.delete(f"/api/inbox/{eintrag['id']}").status_code == 204
|
||
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000
|
||
assert not [e for e in admin.get("/api/inbox").json() if e["id"] == eintrag["id"]]
|
||
|
||
|
||
def _import(admin, konto_id, csv_text):
|
||
import io
|
||
import json
|
||
return admin.post(
|
||
"/api/imports/preview",
|
||
files={"file": ("a.csv", io.BytesIO(csv_text.encode("utf-8")), "text/csv")},
|
||
data={"account_id": str(konto_id), "delimiter": ";", "decimal_sep": ",",
|
||
"date_format": "%d.%m.%Y", "encoding": "utf-8", "has_header": "true",
|
||
"skip_rows": "0", "amount_mode": "signed",
|
||
"column_map": json.dumps({"date": 0, "amount": 1, "payee": 2, "purpose": 3})},
|
||
).json()
|
||
|
||
|
||
KOPF = "Datum;Betrag;Empfaenger;Zweck\n"
|
||
|
||
|
||
def test_auszug_erkennt_die_bereits_per_kurzbefehl_erfasste_zahlung(admin):
|
||
"""Der Kern: dieselbe Zahlung darf nicht zweimal in den Saldo laufen.
|
||
|
||
Der Import-Schlüssel greift hier nicht – auf dem Auszug steht ein anderer
|
||
Text als das, was man am Telefon eingetippt hat, und die Bank bucht ein
|
||
bis zwei Tage später.
|
||
"""
|
||
client = _fremd()
|
||
konto = _konto(admin)
|
||
marke = _marke(admin, konto["id"])
|
||
client.post("/api/quick/booking",
|
||
json={"amount": "12.50", "payee": "Kantine", "date": "2026-05-04"},
|
||
headers={"X-Certain-Token": marke["token"]})
|
||
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000 - 1250
|
||
|
||
# Der Auszug nennt sie anders und datiert sie zwei Tage später.
|
||
v = _import(admin, konto["id"], KOPF + "06.05.2026;-12,50;SV GROUP SCHWEIZ AG;Kartenzahlung\n")
|
||
zeile = v["rows"][0]
|
||
assert zeile["is_duplicate"] is True
|
||
assert zeile["duplicate_reason"] == "inbox"
|
||
assert zeile["include"] is False
|
||
|
||
admin.post("/api/imports/commit", json={
|
||
"account_id": konto["id"], "filename": "mai.csv",
|
||
"rows": [r for r in v["rows"] if r["include"]]})
|
||
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000 - 1250
|
||
|
||
|
||
def test_abweichender_betrag_gilt_nicht_als_dieselbe_zahlung(admin):
|
||
"""Bei Geld ist Toleranz gefährlich: nur der exakte Betrag zählt."""
|
||
client = _fremd()
|
||
konto = _konto(admin)
|
||
marke = _marke(admin, konto["id"])
|
||
client.post("/api/quick/booking",
|
||
json={"amount": "12.50", "payee": "Kantine", "date": "2026-06-04"},
|
||
headers={"X-Certain-Token": marke["token"]})
|
||
|
||
v = _import(admin, konto["id"], KOPF + "05.06.2026;-12,60;SV GROUP;Kartenzahlung\n")
|
||
assert v["rows"][0]["is_duplicate"] is False
|
||
|
||
|
||
def test_zu_weit_entferntes_datum_gilt_nicht(admin):
|
||
client = _fremd()
|
||
konto = _konto(admin)
|
||
marke = _marke(admin, konto["id"])
|
||
client.post("/api/quick/booking",
|
||
json={"amount": "31.00", "payee": "Baumarkt", "date": "2026-07-01"},
|
||
headers={"X-Certain-Token": marke["token"]})
|
||
|
||
v = _import(admin, konto["id"], KOPF + "20.07.2026;-31,00;OBI;Einkauf\n")
|
||
assert v["rows"][0]["is_duplicate"] is False
|
||
|
||
|
||
def test_zwei_gleiche_zahlungen_verbrauchen_je_einen_eintrag(admin):
|
||
"""Zweimal derselbe Kaffee am selben Tag: beide Seiten müssen aufgehen."""
|
||
client = _fremd()
|
||
konto = _konto(admin)
|
||
marke = _marke(admin, konto["id"])
|
||
kopf = {"X-Certain-Token": marke["token"]}
|
||
for lauf in ("a", "b"):
|
||
client.post("/api/quick/booking",
|
||
json={"amount": "4.20", "payee": "Kaffee", "date": "2026-09-02",
|
||
"external_id": f"kaffee-{lauf}"}, headers=kopf)
|
||
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000 - 840
|
||
|
||
v = _import(admin, konto["id"], KOPF
|
||
+ "02.09.2026;-4,20;CAFE BAR;Karte\n"
|
||
+ "02.09.2026;-4,20;CAFE BAR;Karte\n"
|
||
+ "02.09.2026;-4,20;CAFE BAR;Karte\n")
|
||
gruende = [r["duplicate_reason"] for r in v["rows"]]
|
||
assert gruende.count("inbox") == 2, gruende
|
||
# Die dritte ist echt und wird gebucht.
|
||
assert sum(1 for r in v["rows"] if r["include"]) == 1
|
||
|
||
|
||
def test_auszug_vom_telefon_braucht_eine_gueltige_marke(admin):
|
||
import io
|
||
konto = _konto(admin, name="Fremdes Konto")
|
||
r = _fremd().post("/api/quick/statement",
|
||
files={"file": ("a.csv", io.BytesIO(b"x"), "text/csv")},
|
||
data={"account_id": str(konto["id"])})
|
||
assert r.status_code == 401
|
||
|
||
|
||
def test_auszug_vom_telefon_wird_nur_abgelegt_nicht_gebucht(admin):
|
||
"""Am Telefon prüft man keine 500 Zeilen. Die Datei wartet in Certain."""
|
||
client = _fremd()
|
||
konto = _konto(admin, name="Ablagekonto", currency="CHF")
|
||
marke = _marke(admin, konto["id"])
|
||
import io
|
||
|
||
csv_text = KOPF + "01.04.2026;-12,00;Bäckerei;Brot;\n"
|
||
r = client.post("/api/quick/statement",
|
||
files={"file": ("april.csv", io.BytesIO(csv_text.encode()), "text/csv")},
|
||
data={"account_id": str(konto["id"])},
|
||
headers={"X-Certain-Token": marke["token"]})
|
||
assert r.status_code == 201, r.text
|
||
assert "Empfangen" in r.json()["message"]
|
||
# Nichts gebucht.
|
||
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000
|
||
|
||
wartend = [u for u in admin.get("/api/imports/uploads").json()
|
||
if u["account_id"] == konto["id"]]
|
||
assert len(wartend) == 1
|
||
assert wartend[0]["filename"] == "april.csv"
|
||
|
||
# Der Inhalt kommt unverändert zurück – die Import-Ansicht liest ihn wie
|
||
# eine lokale Datei.
|
||
inhalt = admin.get(f"/api/imports/uploads/{wartend[0]['id']}/content")
|
||
assert inhalt.status_code == 200
|
||
assert "Bäckerei" in inhalt.content.decode("utf-8")
|
||
|
||
assert admin.delete(f"/api/imports/uploads/{wartend[0]['id']}").status_code == 204
|
||
assert not [u for u in admin.get("/api/imports/uploads").json()
|
||
if u["id"] == wartend[0]["id"]]
|
||
|
||
|
||
def test_leere_und_zu_grosse_dateien_werden_abgewiesen(admin):
|
||
client = _fremd()
|
||
konto = _konto(admin, name="Grenzkonto")
|
||
marke = _marke(admin, konto["id"])
|
||
kopf = {"X-Certain-Token": marke["token"]}
|
||
import io
|
||
|
||
leer = client.post("/api/quick/statement",
|
||
files={"file": ("leer.csv", io.BytesIO(b""), "text/csv")},
|
||
data={"account_id": str(konto["id"])}, headers=kopf)
|
||
assert leer.status_code == 400
|
||
|
||
riesig = client.post("/api/quick/statement",
|
||
files={"file": ("gross.csv", io.BytesIO(b"x" * (6 * 1024 * 1024)), "text/csv")},
|
||
data={"account_id": str(konto["id"])}, headers=kopf)
|
||
assert riesig.status_code == 413
|
||
|
||
|
||
def test_fehler_am_kurzbefehl_sind_im_feld_message_lesbar(admin):
|
||
"""Der Kurzbefehl zeigt „message" an. Steht dort bei einem Fehler nichts,
|
||
sieht man auf dem Telefon nur eine leere Mitteilung – und sucht den Fehler
|
||
dann auf dem Server statt im Kurzbefehl."""
|
||
client = _fremd()
|
||
konto = _konto(admin, name="Meldungskonto")
|
||
marke = _marke(admin, konto["id"])
|
||
kopf = {"X-Certain-Token": marke["token"]}
|
||
import io
|
||
|
||
# Gar keine Datei mitgeschickt: Pflichtfeld fehlt.
|
||
r = client.post("/api/quick/statement", data={"account_id": str(konto["id"])}, headers=kopf)
|
||
assert r.status_code == 422
|
||
assert "file" in r.json()["message"]
|
||
assert "Formular" in r.json()["message"]
|
||
|
||
# Falsche Marke.
|
||
r = client.post("/api/quick/booking", json={"amount": "1"},
|
||
headers={"X-Certain-Token": "falsch"})
|
||
assert r.status_code == 401
|
||
assert "message" in r.json()
|
||
|
||
# Leere Datei.
|
||
r = client.post("/api/quick/statement",
|
||
files={"file": ("leer.csv", io.BytesIO(b""), "text/csv")},
|
||
data={"account_id": str(konto["id"])}, headers=kopf)
|
||
assert r.status_code == 400
|
||
assert "leer" in r.json()["message"].lower()
|
||
|
||
|
||
def test_konto_wird_aus_der_datei_erkannt(admin):
|
||
"""Ein Kurzbefehl statt einer je Konto: Raiffeisen fuehrt die eigene IBAN
|
||
in der Datei mit, damit ist die Zuordnung eindeutig."""
|
||
client = _fremd()
|
||
iban = "CH43 8080 8007 9160 7106 9"
|
||
konto = _konto(admin, name="Lohnkonto IBAN", currency="CHF",
|
||
identifiers=[{"kind": "iban", "value": iban}])
|
||
# Zweites Konto derselben Waehrung, damit es etwas zu verwechseln gibt.
|
||
_konto(admin, name="Anderes CHF-Konto", currency="CHF")
|
||
marke = _marke(admin, None, name="Ohne Standardkonto " + secrets.token_hex(3))
|
||
import io
|
||
|
||
csv_text = (
|
||
"IBAN;Booked At;Text;Credit/Debit Amount;Balance;Valuta Date\n"
|
||
"CH4380808007916071069;2026-01-05 00:00:00.0;LSV Viseca;-1718.7;702.47;2026-01-05 00:00:00.0\n"
|
||
"CH4380808007916071069;2026-01-06 00:00:00.0;Coop;-42.10;660.37;2026-01-06 00:00:00.0\n"
|
||
)
|
||
r = client.post("/api/quick/statement",
|
||
files={"file": ("raiff.csv", io.BytesIO(csv_text.encode()), "text/csv")},
|
||
headers={"X-Certain-Token": marke["token"]})
|
||
assert r.status_code == 201, r.text
|
||
assert "IBAN" in r.json()["erkannt_als"]
|
||
|
||
wartend = [u for u in admin.get("/api/imports/uploads").json()
|
||
if u["filename"] == "raiff.csv"]
|
||
assert wartend and wartend[0]["account_id"] == konto["id"]
|
||
|
||
|
||
def test_unklares_konto_nennt_die_moeglichkeiten(admin):
|
||
"""Raten waere teuer: Ein Auszug im falschen Konto verfaelscht zwei Salden."""
|
||
client = _fremd()
|
||
_konto(admin, name="Erstes Konto", currency="CHF")
|
||
_konto(admin, name="Zweites Konto", currency="CHF")
|
||
marke = _marke(admin, None, name="Ratlos " + secrets.token_hex(3))
|
||
import io
|
||
|
||
r = client.post("/api/quick/statement",
|
||
files={"file": ("x.csv", io.BytesIO(
|
||
b"Datum;Betrag;Text\n01.02.2026;-5,00;Kiosk\n"), "text/csv")},
|
||
headers={"X-Certain-Token": marke["token"]})
|
||
assert r.status_code == 400
|
||
assert "account_id" in r.json()["message"]
|
||
assert "Erstes Konto" in r.json()["message"]
|