GuidoandClaude Opus 5
34be575eb3
Revolut und Raiffeisen lassen die CSV auch am Handy exportieren; nur N26 verlangt die Web-Version. Ein zweiter Kurzbefehl nimmt die Datei aus dem Teilen-Menue entgegen und schickt sie mit dem gewaehlten Konto an /quick/statement. Am Telefon laesst sich keine Spaltenzuordnung vornehmen. Der Endpunkt verwendet deshalb die Einstellung, mit der dieses Konto zuletzt erfolgreich eingelesen wurde – und sagt klar Bescheid, wenn es noch keine gibt, statt mit falschen Spalten zu raten. Gebucht wird sofort: Ein Auszug hat hunderte Zeilen, die einzeln abzunicken waere keine Erleichterung. Sicher ist das durch die Duplikaterkennung und weil der Stapel im Import-Verlauf mit einem Klick zurueckzunehmen ist. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
308 lines
13 KiB
Python
308 lines
13 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_nutzt_die_zuletzt_benutzte_vorlage(admin):
|
||
"""Am Telefon kann man keine Spalten zuordnen. Also wird die Einstellung
|
||
wiederverwendet, mit der dieses Konto zuletzt eingelesen wurde."""
|
||
client = _fremd()
|
||
konto = _konto(admin, name="Handykonto", currency="CHF")
|
||
marke = _marke(admin, konto["id"])
|
||
kopf = {"X-Certain-Token": marke["token"]}
|
||
import io
|
||
|
||
csv_text = KOPF + "01.04.2026;-12,00;Bäckerei;Brot;\n02.04.2026;-8,00;Kiosk;Zeitung;\n"
|
||
|
||
# Ohne vorherigen Import am Rechner: klare Ansage statt Rätselraten.
|
||
r = client.post("/api/quick/statement",
|
||
files={"file": ("a.csv", io.BytesIO(csv_text.encode()), "text/csv")},
|
||
data={"account_id": str(konto["id"])}, headers=kopf)
|
||
assert r.status_code == 400
|
||
assert "am Rechner" in r.json()["detail"]
|
||
|
||
# Einmal am Rechner einlesen legt die Einstellung fest.
|
||
import json as _json
|
||
vorlage = admin.post("/api/imports/templates", json={
|
||
"name": "Handytest " + secrets.token_hex(3), "delimiter": ";", "decimal_sep": ",",
|
||
"date_format": "%d.%m.%Y", "encoding": "utf-8", "has_header": True, "skip_rows": 0,
|
||
"amount_mode": "signed",
|
||
"column_map": {"date": 0, "amount": 1, "payee": 2, "purpose": 3},
|
||
})
|
||
assert vorlage.status_code == 201, vorlage.text
|
||
v = _import(admin, konto["id"], csv_text)
|
||
admin.post("/api/imports/commit", json={
|
||
"account_id": konto["id"], "filename": "april.csv",
|
||
"template_id": vorlage.json()["id"],
|
||
"rows": [x for x in v["rows"] if x["include"]]})
|
||
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000 - 2000
|
||
|
||
# Ab jetzt geht es vom Telefon: neue Zeilen kommen an, alte nicht doppelt.
|
||
mehr = csv_text + "03.04.2026;-5,50;Coop;Milch;\n"
|
||
r = client.post("/api/quick/statement",
|
||
files={"file": ("april.csv", io.BytesIO(mehr.encode()), "text/csv")},
|
||
data={"account_id": str(konto["id"])}, headers=kopf)
|
||
assert r.status_code == 201, r.text
|
||
assert r.json()["gebucht"] == 1
|
||
assert r.json()["duplikate"] == 2
|
||
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 100000 - 2550
|
||
|
||
# Und der Stapel steht im Verlauf, also mit einem Klick ruecknehmbar.
|
||
stapel = admin.get("/api/imports/batches").json()
|
||
assert any(b["filename"] == "april.csv" and b["imported_count"] == 1 for b in stapel)
|
||
|
||
|
||
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
|