diff --git a/backend/app/routers/transactions.py b/backend/app/routers/transactions.py index bb0fcf3..6a2d22f 100644 --- a/backend/app/routers/transactions.py +++ b/backend/app/routers/transactions.py @@ -9,7 +9,7 @@ from datetime import date from fastapi import APIRouter, Depends, HTTPException, Query, status from pydantic import BaseModel -from sqlalchemy import select, union +from sqlalchemy import or_, select, union from sqlalchemy.orm import Session from ..database import get_db @@ -194,18 +194,24 @@ def list_transactions( type: TxnType | None = None, date_from: date | None = None, date_to: date | None = None, + # Suche und die zwei Filter, die es bisher nur im Browser gab. Die + # Weboberfläche lädt alle Buchungen und siebt sie dort; ein Telefon soll + # dafür nicht 5000 Zeilen über die Leitung ziehen müssen. + q: str | None = None, + tag_id: int | None = None, + subscription_id: int | None = None, # Großzügiges Limit: die Buchungsliste (DataTable) filtert/sortiert im Browser, # deshalb sollen standardmäßig ALLE Buchungen geladen werden, nicht nur 200. limit: int = Query(default=5000, le=50000), offset: int = 0, ): - q = db.query(Transaction).filter(Transaction.owner_id == user.id) + abfrage = db.query(Transaction).filter(Transaction.owner_id == user.id) if account_id is not None: - q = q.filter( + abfrage = abfrage.filter( (Transaction.account_id == account_id) | (Transaction.to_account_id == account_id) ) if category_id is not None: - q = q.filter(Transaction.category_id == category_id) + abfrage = abfrage.filter(Transaction.category_id == category_id) if uncategorized: # Verknüpfte Buchungen sind erledigt: Sie gehören zu einem Übertrag und # zählen in keiner Auswertung. Sie hier nach einer Kategorie zu fragen, @@ -214,16 +220,39 @@ def list_transactions( select(TransferLink.from_txn_id).where(TransferLink.owner_id == user.id), select(TransferLink.to_txn_id).where(TransferLink.owner_id == user.id), ) - q = q.filter(Transaction.category_id.is_(None), - Transaction.type != TxnType.transfer, - Transaction.id.notin_(verknuepft)) + abfrage = abfrage.filter(Transaction.category_id.is_(None), + Transaction.type != TxnType.transfer, + Transaction.id.notin_(verknuepft)) if type is not None: - q = q.filter(Transaction.type == type) + abfrage = abfrage.filter(Transaction.type == type) if date_from is not None: - q = q.filter(Transaction.date >= date_from) + abfrage = abfrage.filter(Transaction.date >= date_from) if date_to is not None: - q = q.filter(Transaction.date <= date_to) - rows = q.order_by(Transaction.date.desc(), Transaction.id.desc()).limit(limit).offset(offset).all() + abfrage = abfrage.filter(Transaction.date <= date_to) + if q: + # LIKE statt Volltextindex: Ein Haushalt bleibt im Zehntausenderbereich, + # da kostet die Suche nichts Spürbares. Was SQLite dabei nicht kann: + # Umlaute gross/klein gleichsetzen – „Müller" findet „müller" nicht. + # Die Prozent- und Unterstrich-Zeichen des Suchtexts werden entwertet, + # sonst wäre jede Eingabe mit „%" ein Platzhalter statt eines Zeichens. + muster = "%" + q.translate( + str.maketrans({"\\": "\\\\", "%": "\\%", "_": "\\_"}) + ) + "%" + abfrage = abfrage.filter(or_( + Transaction.payee.ilike(muster, escape="\\"), + Transaction.description.ilike(muster, escape="\\"), + Transaction.notes.ilike(muster, escape="\\"), + )) + if tag_id is not None: + # Unterabfrage statt Join, wie im uncategorized-Zweig: Ein Join würde + # jede Buchung so oft liefern, wie sie Tags trägt. + mit_tag = select(TransactionTag.transaction_id).where( + TransactionTag.owner_id == user.id, TransactionTag.tag_id == tag_id + ) + abfrage = abfrage.filter(Transaction.id.in_(mit_tag)) + if subscription_id is not None: + abfrage = abfrage.filter(Transaction.subscription_id == subscription_id) + rows = abfrage.order_by(Transaction.date.desc(), Transaction.id.desc()).limit(limit).offset(offset).all() base = base_currency(db, user) ccy = _currency_map(db, user) woher = _herkunft(db, user, rows) diff --git a/backend/tests/test_buchungssuche.py b/backend/tests/test_buchungssuche.py new file mode 100644 index 0000000..ef7c879 --- /dev/null +++ b/backend/tests/test_buchungssuche.py @@ -0,0 +1,126 @@ +"""Suche und Filter auf der Buchungsliste: ``q``, ``tag_id``, ``subscription_id``. + +Bisher siebte die Weboberflaeche im Browser und lud dafuer alles, was es gab - +das Limit steht nicht umsonst auf 5000. Am Telefon ist das der Unterschied +zwischen "laedt" und "benutzbar", deshalb sitzen die drei Filter jetzt im +Server. + +Die Datenbank lebt ueber den ganzen Testlauf. Texte deshalb eindeutig +wuerfeln, sonst findet die Suche die Daten anderer Tests mit. +""" + +import secrets + +from conftest import login_ready, new_user + + +def _konto(admin): + return admin.post("/api/accounts", json={ + "name": f"Konto {secrets.token_hex(3)}", "type": "girokonto", + "currency": "EUR", "opening_balance_cents": 0}).json() + + +def _buchung(admin, konto, **felder): + r = admin.post("/api/transactions", json={ + "type": "expense", "date": "2026-09-15", "account_id": konto["id"], + "amount_cents": 100 + secrets.randbelow(9000), **felder}) + assert r.status_code == 201, r.text + return r.json() + + +def _suche(admin, **filter_): + r = admin.get("/api/transactions", params=filter_) + assert r.status_code == 200, r.text + return [t["id"] for t in r.json()] + + +def test_suche_findet_empfaenger_verwendungszweck_und_notiz(admin): + konto = _konto(admin) + wort = f"Zebra{secrets.token_hex(4)}" + ueber_empfaenger = _buchung(admin, konto, payee=f"Laden {wort}") + ueber_zweck = _buchung(admin, konto, payee="Egal", description=f"Rechnung {wort}") + ueber_notiz = _buchung(admin, konto, payee="Egal", notes=f"wegen {wort}") + daneben = _buchung(admin, konto, payee="Etwas voellig anderes") + + gefunden = _suche(admin, q=wort) + assert set(gefunden) == {ueber_empfaenger["id"], ueber_zweck["id"], ueber_notiz["id"]} + assert daneben["id"] not in gefunden + + +def test_suche_achtet_nicht_auf_gross_und_kleinschreibung(admin): + konto = _konto(admin) + wort = f"Kiosk{secrets.token_hex(4)}" + txn = _buchung(admin, konto, payee=wort.upper()) + + assert txn["id"] in _suche(admin, q=wort.lower()) + + +def test_prozentzeichen_im_suchtext_bleibt_ein_zeichen(admin): + """Sonst waere jede Eingabe mit % ein Platzhalter und faende alles.""" + konto = _konto(admin) + marke = secrets.token_hex(4) + mit_prozent = _buchung(admin, konto, payee=f"Rabatt%{marke}") + ohne = _buchung(admin, konto, payee=f"Rabatt {marke}") + + gefunden = _suche(admin, q=f"Rabatt%{marke}") + assert gefunden == [mit_prozent["id"]] + assert ohne["id"] not in gefunden + + +def test_tag_filter_zeigt_nur_buchungen_mit_diesem_tag(admin): + konto = _konto(admin) + tag = admin.post("/api/tags", json={ + "name": f"Tag {secrets.token_hex(4)}", "color": "#3f51b5"}).json() + mit = _buchung(admin, konto, payee="Mit Etikett") + ohne = _buchung(admin, konto, payee="Ohne Etikett") + admin.patch(f"/api/transactions/{mit['id']}", json={"tag_ids": [tag["id"]]}) + + gefunden = _suche(admin, tag_id=tag["id"]) + assert gefunden == [mit["id"]] + assert ohne["id"] not in gefunden + + +def test_tag_eines_fremden_nutzers_zeigt_keine_buchungen(admin): + """Mandanten-Isolation: Eine geratene fremde ID darf nichts aufschliessen.""" + konto = _konto(admin) + meine = _buchung(admin, konto, payee="Meine Buchung") + + name, passwort = new_user(admin) + fremd, _ = login_ready(name, passwort) + fremder_tag = fremd.post("/api/tags", json={ + "name": f"Fremd {secrets.token_hex(4)}", "color": "#eb6834"}).json() + + gefunden = _suche(admin, tag_id=fremder_tag["id"]) + assert gefunden == [] + assert meine["id"] not in gefunden + + +def test_abo_filter_zeigt_nur_buchungen_dieses_vertrags(admin): + konto = _konto(admin) + abo = admin.post("/api/subscriptions", json={ + "name": f"Dienst {secrets.token_hex(4)}", "amount_cents": 999, + "currency": "EUR", "interval_months": 1}).json() + zugeordnet = _buchung(admin, konto, payee="Monatsbeitrag") + frei = _buchung(admin, konto, payee="Einmalkauf") + r = admin.post(f"/api/subscriptions/{abo['id']}/buchungen/{zugeordnet['id']}") + assert r.status_code == 200, r.text + + gefunden = _suche(admin, subscription_id=abo["id"]) + assert gefunden == [zugeordnet["id"]] + assert frei["id"] not in gefunden + + +def test_suche_und_tag_filter_greifen_zusammen(admin): + konto = _konto(admin) + wort = f"Baumarkt{secrets.token_hex(4)}" + tag = admin.post("/api/tags", json={ + "name": f"Tag {secrets.token_hex(4)}", "color": "#2a78d6"}).json() + beides = _buchung(admin, konto, payee=f"{wort} Farbe") + nur_text = _buchung(admin, konto, payee=f"{wort} Schrauben") + nur_tag = _buchung(admin, konto, payee="Etwas anderes") + for txn in (beides, nur_tag): + admin.patch(f"/api/transactions/{txn['id']}", json={"tag_ids": [tag["id"]]}) + + gefunden = _suche(admin, q=wort, tag_id=tag["id"]) + assert gefunden == [beides["id"]] + assert nur_text["id"] not in gefunden and nur_tag["id"] not in gefunden