"""Abos & Vertraege: verzeichnen und zusammenrechnen. Der Nutzen dieser Stufe ist eine einzige Zahl – was die Vertraege zusammen im Monat kosten. Alles andere haengt daran, dass sie stimmt: auch ueber Waehrungen und ueber verschiedene Intervalle hinweg. """ import json import secrets from datetime import date, timedelta from conftest import login_ready, new_user def _abo(admin, **kw): body = {"name": "Netflix " + secrets.token_hex(3), "amount_cents": 1790, "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 _liste(admin): r = admin.get("/api/subscriptions") assert r.status_code == 200, 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"Kategorie '{name}' fehlt") def test_monatswert_wird_aus_dem_intervall_gerechnet(admin): """Nicht gespeichert, sondern gerechnet – sonst veraltet er beim naechsten Preisaufschlag stillschweigend.""" jahr = _abo(admin, amount_cents=40000, interval_months=12) assert jahr["monatlich_cents"] == 3333 quartal = _abo(admin, amount_cents=3000, interval_months=3) assert quartal["monatlich_cents"] == 1000 monat = _abo(admin, amount_cents=1790, interval_months=1) assert monat["monatlich_cents"] == 1790 def test_preisaenderung_zieht_den_monatswert_mit(admin): abo = _abo(admin, amount_cents=1790, interval_months=1) r = admin.patch(f"/api/subscriptions/{abo['id']}", json={"amount_cents": 1990}) assert r.status_code == 200, r.text assert r.json()["monatlich_cents"] == 1990 def test_summe_zaehlt_gekuendigte_nicht_mit(admin): vorher = _liste(admin)["summary"]["monatlich_cents"] abo = _abo(admin, amount_cents=2500, interval_months=1, currency="EUR") assert _liste(admin)["summary"]["monatlich_cents"] == vorher + 2500 admin.patch(f"/api/subscriptions/{abo['id']}", json={"is_active": False}) assert _liste(admin)["summary"]["monatlich_cents"] == vorher def test_fremde_waehrung_wird_umgerechnet(admin): """Ein deutscher Handyvertrag laeuft in Euro, auch wenn das Konto Franken fuehrt. Ohne Umrechnung liessen sich die Zeilen nicht addieren.""" from app.database import SessionLocal from app.models import ExchangeRate with SessionLocal() as s: if s.get(ExchangeRate, (date.today(), "CHF")) is None: s.add(ExchangeRate(date=date.today(), currency="CHF", rate_per_eur=0.95, source="test")) s.commit() abo = _abo(admin, amount_cents=1000, interval_months=1, currency="CHF") # Hauptwaehrung ist EUR: 10.00 CHF bei 0.95 CHF je EUR sind 10.53 EUR. assert abo["monatlich_base_cents"] == 1053 assert abo["monatlich_cents"] == 1000, "die eigene Waehrung bleibt unangetastet" def test_waehrung_ohne_kurs_wird_gemeldet_statt_verschluckt(admin): """Lieber sagen, dass eine Zeile fehlt, als die Summe zu klein anzuzeigen.""" abo = _abo(admin, amount_cents=999, interval_months=1, currency="ZAR") summary = _liste(admin)["summary"] assert "ZAR" in summary["ohne_kurs"], summary treffer = next(z for z in _liste(admin)["subscriptions"] if z["id"] == abo["id"]) assert treffer["monatlich_base_cents"] is None def test_dollar_ist_jetzt_erlaubt(admin): """Manche Abos werden in Dollar abgerechnet. Frueher lehnte das Formular alles ausser EUR und CHF ab.""" abo = _abo(admin, currency="USD", amount_cents=1200) assert abo["currency"] == "USD" def test_erfundene_waehrung_wird_abgelehnt(admin): """Ohne Kurs liesse sich der Betrag nie umrechnen – er stuende in keiner Summe. Das faellt spaeter auf und ist dann schwer zu erklaeren.""" r = admin.post("/api/subscriptions", json={ "name": "Fantasie", "amount_cents": 100, "currency": "XYZ", "interval_months": 1}) assert r.status_code == 422, r.text def test_krummes_intervall_wird_abgelehnt(admin): r = admin.post("/api/subscriptions", json={ "name": "Krumm", "amount_cents": 100, "currency": "EUR", "interval_months": 37}) assert r.status_code == 422, r.text def test_kategorie_muss_eine_ausgabe_sein(admin): gehalt = _kategorie(admin, "Gehalt") r = admin.post("/api/subscriptions", json={ "name": "Falsch", "amount_cents": 100, "currency": "EUR", "interval_months": 1, "category_id": gehalt}) assert r.status_code == 400, r.text assert "Ausgabe" in r.json()["detail"] def test_erkennungstext_wird_aus_dem_namen_vorbelegt(admin): abo = _abo(admin, name="Spotify Familie") assert abo["payee_pattern"] == "Spotify Familie" def test_faelligkeit_zaehlt_die_tage(admin): abo = _abo(admin, next_due=str(date.today() + timedelta(days=9))) assert abo["tage_bis_faellig"] == 9 ueberfaellig = _abo(admin, next_due=str(date.today() - timedelta(days=2))) assert ueberfaellig["tage_bis_faellig"] == -2 def test_sparkonto_laesst_sich_wieder_loesen(admin): """Sonst wird man ein einmal gesetztes Pocket nie wieder los.""" konto = admin.post("/api/accounts", json={ "name": "Pocket " + secrets.token_hex(3), "type": "sparkonto", "currency": "EUR", "opening_balance_cents": 0}).json() abo = _abo(admin, interval_months=12, amount_cents=40000, savings_account_id=konto["id"]) assert abo["savings_account_id"] == konto["id"] r = admin.patch(f"/api/subscriptions/{abo['id']}", json={"savings_account_id": None}) assert r.status_code == 200, r.text assert r.json()["savings_account_id"] is None def test_fremdes_konto_wird_nicht_angenommen(admin): """Sonst schriebe ein Vertrag seine Zahlungen spaeter in ein Konto, das einem anderen gehoert.""" fremd = admin.post("/api/accounts", json={ "name": "Fremd " + secrets.token_hex(3), "type": "girokonto", "currency": "EUR", "opening_balance_cents": 0}).json() username, password = new_user(admin) anderer, _ = login_ready(username, password) r = anderer.post("/api/subscriptions", json={ "name": "Fremdzugriff", "amount_cents": 100, "currency": "EUR", "interval_months": 1, "account_id": fremd["id"]}) assert r.status_code == 404, r.text def test_angebotene_waehrungen_sind_eine_einstellung(admin): """Die Liste beschraenkt nur die ANZEIGE. Gueltig bleibt alles mit Kurs – sonst verloere ein Konto seine Waehrung, weil sie gerade nicht angehakt ist.""" rows = {r["key"]: r["value"] for r in admin.get("/api/settings").json()} assert rows["active_currencies"] == "EUR,CHF,USD", "Vorgabe passt nicht" r = admin.put("/api/settings/active_currencies", params={"value": "EUR,CHF"}) assert r.status_code == 200, r.text try: # Trotz Abwahl weiter anlegbar: Die Einstellung ist keine Sperre. abo = _abo(admin, currency="USD", amount_cents=500) assert abo["currency"] == "USD" finally: admin.put("/api/settings/active_currencies", params={"value": "EUR,CHF,USD"}) # --- Markenlogo -------------------------------------------------------------- def _bild(groesse=(600, 400), farbe=(200, 30, 30, 255)) -> bytes: """Ein echtes PNG, absichtlich zu gross – so kommt es aus dem Alltag.""" import io as _io from PIL import Image puffer = _io.BytesIO() Image.new("RGBA", groesse, farbe).save(puffer, format="PNG") return puffer.getvalue() def _hochladen(admin, sub_id, daten=None, name="logo.png", typ="image/png"): return admin.put(f"/api/subscriptions/{sub_id}/logo", files={"file": (name, daten if daten is not None else _bild(), typ)}) def test_logo_wird_klein_gerechnet_und_ausgeliefert(admin): """Hochgeladen wird, was das Telefon hergibt – gespeichert ein Bildchen.""" from app.services import logo abo = _abo(admin) assert abo["has_logo"] is False r = _hochladen(admin, abo["id"]) assert r.status_code == 200, r.text assert r.json()["has_logo"] is True assert r.json()["logo_version"] is not None bild = admin.get(f"/api/subscriptions/{abo['id']}/logo") assert bild.status_code == 200 assert bild.headers["content-type"] == "image/png" assert len(bild.content) < 20_000, "Logo nicht verkleinert" from PIL import Image import io as _io gespeichert = Image.open(_io.BytesIO(bild.content)) assert max(gespeichert.size) <= logo.MAX_KANTE # Seitenverhaeltnis bleibt: ein verzerrtes Logo sieht falscher aus als ein # kleines. 600x400 ist 3:2. assert gespeichert.size == (128, 85), gespeichert.size def test_ohne_logo_gibt_es_kein_bild(admin): abo = _abo(admin) assert admin.get(f"/api/subscriptions/{abo['id']}/logo").status_code == 404 def test_logo_laesst_sich_austauschen_und_entfernen(admin): abo = _abo(admin) _hochladen(admin, abo["id"], _bild((300, 300), (10, 200, 10, 255))) erst = admin.get(f"/api/subscriptions/{abo['id']}/logo").content _hochladen(admin, abo["id"], _bild((300, 300), (10, 10, 200, 255))) zweit = admin.get(f"/api/subscriptions/{abo['id']}/logo").content assert zweit != erst, "das alte Logo blieb stehen" r = admin.delete(f"/api/subscriptions/{abo['id']}/logo") assert r.status_code == 200, r.text assert r.json()["has_logo"] is False assert admin.get(f"/api/subscriptions/{abo['id']}/logo").status_code == 404 def test_was_kein_bild_ist_wird_verstaendlich_abgewiesen(admin): abo = _abo(admin) r = _hochladen(admin, abo["id"], b"kein bild, nur text", "notiz.txt", "text/plain") assert r.status_code == 400, r.text assert "Bild" in r.json()["detail"] def test_logo_eines_fremden_vertrags_bleibt_verschlossen(admin): """Sonst laege es hinter einer geratenen Nummer offen.""" abo = _abo(admin) _hochladen(admin, abo["id"]) username, password = new_user(admin) anderer, _ = login_ready(username, password) assert anderer.get(f"/api/subscriptions/{abo['id']}/logo").status_code == 404 def test_geloeschter_vertrag_nimmt_sein_logo_mit(admin): from app.database import SessionLocal from app.models import SubscriptionLogo abo = _abo(admin) _hochladen(admin, abo["id"]) assert admin.delete(f"/api/subscriptions/{abo['id']}").status_code == 204 with SessionLocal() as s: assert s.get(SubscriptionLogo, abo["id"]) is None, "Logo blieb verwaist zurueck" def test_logo_uebersteht_sichern_und_einspielen(admin): """Ein Bild ist binaer – ohne Base64 scheitert schon das Schreiben der Sicherung, und zwar erst, wenn der Nutzer sie zu haben glaubt.""" abo = _abo(admin, name="Logo-Backup " + secrets.token_hex(3)) _hochladen(admin, abo["id"], _bild((200, 200), (7, 77, 177, 255))) vorher = admin.get(f"/api/subscriptions/{abo['id']}/logo").content sicherung = admin.get("/api/export/backup.json") assert sicherung.status_code == 200, sicherung.text roh = sicherung.content daten = json.loads(roh) haufen = daten.get("data", daten) logos = haufen.get("subscription_logos") assert logos, f"Logos fehlen in der Sicherung: {sorted(haufen)[:25]}" admin.delete(f"/api/subscriptions/{abo['id']}/logo") assert admin.get(f"/api/subscriptions/{abo['id']}/logo").status_code == 404 r = admin.post("/api/backup/restore", files={"file": ("backup.json", roh, "application/json")}, data={"mode": "merge"}) assert r.status_code == 200, r.text nachher = admin.get(f"/api/subscriptions/{abo['id']}/logo") assert nachher.status_code == 200, "Logo kam nicht zurueck" assert nachher.content == vorher, "Logo kam veraendert zurueck"