280 lines
14 KiB
Python
280 lines
14 KiB
Python
from datetime import date, timedelta
|
|
from decimal import Decimal
|
|
|
|
from app.models.tables import Account, Transaction
|
|
from app.services.balances import account_balance
|
|
|
|
H = {"Authorization": "Bearer test-key"}
|
|
|
|
|
|
def test_account_crud_and_balance(client, db):
|
|
# Portiert von Statement-closing-Logik auf Konto-Saldo-Anker (Ausbaustufe 3
|
|
# Task 1): frueher lieferte die closing_balance des juengsten bestaetigten
|
|
# Statements die Basis, jetzt uebernimmt der Anker (Datum+Betrag am Konto)
|
|
# dieselbe Rolle. `at` wird explizit gesetzt statt auf date.today() zu
|
|
# vertrauen, damit der Test unabhaengig vom Ausfuehrungsdatum ist.
|
|
r = client.post("/api/accounts", headers=H, json={
|
|
"bank": "DKB", "iban": "DE02120300000000202051",
|
|
"name": "Giro", "type": "giro"})
|
|
assert r.status_code == 201
|
|
acc_id = r.json()["id"]
|
|
acc = db.get(Account, acc_id)
|
|
acc.anchor_date = date(2026, 6, 30)
|
|
acc.anchor_balance = Decimal("100.00")
|
|
db.add(Transaction(account_id=acc_id, booking_date=date(2026, 7, 2),
|
|
amount=Decimal("-30.00"), purpose="Bar", status="confirmed",
|
|
dedup_hash="x1"))
|
|
db.commit()
|
|
acc = db.get(Account, acc_id)
|
|
assert account_balance(db, acc, at=date(2026, 7, 15)) == Decimal("70.00")
|
|
|
|
|
|
def test_account_balance_anchor_forward_and_backward(client, db):
|
|
# Spec-Interface (Plan Ausbaustufe 3 Task 1): anchor_balance +
|
|
# Sum(tx: anchor_date < booking_date <= at) - Sum(tx: at < booking_date
|
|
# <= anchor_date). Buchungen VOR und NACH dem Anker duerfen sich nicht
|
|
# gegenseitig beeinflussen.
|
|
acc = Account(bank="dkb", iban="DE-ANCHOR-1", name="Anker-Konto", type="giro")
|
|
db.add(acc)
|
|
db.flush()
|
|
anchor_date = date(2026, 6, 30)
|
|
acc.anchor_date = anchor_date
|
|
acc.anchor_balance = Decimal("500.00")
|
|
db.add(Transaction(account_id=acc.id, booking_date=date(2026, 6, 29),
|
|
amount=Decimal("-20.00"), purpose="vor Anker",
|
|
status="confirmed", dedup_hash="x2"))
|
|
db.add(Transaction(account_id=acc.id, booking_date=date(2026, 7, 1),
|
|
amount=Decimal("50.00"), purpose="nach Anker",
|
|
status="confirmed", dedup_hash="x3"))
|
|
db.commit()
|
|
acc = db.get(Account, acc.id)
|
|
|
|
assert account_balance(db, acc, at=anchor_date) == Decimal("500.00")
|
|
assert account_balance(db, acc, at=anchor_date + timedelta(days=2)) == Decimal("550.00")
|
|
assert account_balance(db, acc, at=anchor_date - timedelta(days=2)) == Decimal("520.00")
|
|
|
|
|
|
def test_account_balance_anchor_day_transaction_convention(client, db):
|
|
# Konvention (Plan Step 5): der Anker gilt PER TAGESENDE des Ankerdatums.
|
|
# Eine Buchung AM Ankertag ist im Anker bereits enthalten -> zaehlt NICHT
|
|
# zur Vorwaertssumme, wohl aber zur Rueckwaertssumme.
|
|
acc = Account(bank="dkb", iban="DE-ANCHOR-2", name="Anker-Konto 2", type="giro")
|
|
db.add(acc)
|
|
db.flush()
|
|
anchor_date = date(2026, 6, 30)
|
|
acc.anchor_date = anchor_date
|
|
acc.anchor_balance = Decimal("200.00")
|
|
db.add(Transaction(account_id=acc.id, booking_date=anchor_date,
|
|
amount=Decimal("15.00"), purpose="am Ankertag",
|
|
status="confirmed", dedup_hash="x4"))
|
|
db.commit()
|
|
acc = db.get(Account, acc.id)
|
|
|
|
assert account_balance(db, acc, at=anchor_date + timedelta(days=1)) == Decimal("200.00")
|
|
assert account_balance(db, acc, at=anchor_date - timedelta(days=1)) == Decimal("185.00")
|
|
|
|
|
|
def test_account_without_anchor_uses_zero_base(client, db):
|
|
# Ohne Anker: Summe aller bestaetigten Buchungen bis `at` (Basis 0) -
|
|
# bisheriges Verhalten bleibt fuer Konten ohne Anker erhalten.
|
|
acc = Account(bank="dkb", iban="DE-NOANCHOR", name="Kein Anker", type="giro")
|
|
db.add(acc)
|
|
db.flush()
|
|
db.add(Transaction(account_id=acc.id, booking_date=date(2026, 7, 1),
|
|
amount=Decimal("42.00"), purpose="ohne Anker",
|
|
status="confirmed", dedup_hash="x5"))
|
|
db.commit()
|
|
acc = db.get(Account, acc.id)
|
|
assert account_balance(db, acc, at=date(2026, 7, 15)) == Decimal("42.00")
|
|
|
|
|
|
def test_manual_transaction_dedup(client):
|
|
acc = client.post("/api/accounts", headers=H, json={
|
|
"bank": "DKB", "iban": "DE99", "name": "G", "type": "giro"}).json()
|
|
payload = {"account_id": acc["id"], "booking_date": "2026-07-01",
|
|
"amount": "-10.00", "purpose": "Kaffee", "counterparty": ""}
|
|
assert client.post("/api/transactions", headers=H, json=payload).status_code == 201
|
|
assert client.post("/api/transactions", headers=H, json=payload).status_code == 409
|
|
payload["force"] = True
|
|
assert client.post("/api/transactions", headers=H, json=payload).status_code == 201
|
|
|
|
|
|
def test_rules_categorize(client):
|
|
cat = client.post("/api/categories", headers=H, json={"name": "Energie"}).json()
|
|
client.post("/api/category-rules", headers=H,
|
|
json={"pattern": "stadtwerke", "category_id": cat["id"]})
|
|
acc = client.post("/api/accounts", headers=H, json={
|
|
"bank": "DKB", "iban": "DE98", "name": "G2", "type": "giro"}).json()
|
|
client.post("/api/transactions", headers=H, json={
|
|
"account_id": acc["id"], "booking_date": "2026-07-03",
|
|
"amount": "-80.00", "purpose": "STADTWERKE Abschlag", "counterparty": ""})
|
|
txs = client.get("/api/transactions", headers=H,
|
|
params={"account_id": acc["id"]}).json()
|
|
assert txs[0]["category_id"] == cat["id"]
|
|
|
|
|
|
def test_patch_account_name(client):
|
|
acc = client.post("/api/accounts", headers=H, json={
|
|
"bank": "DKB", "iban": "DE71", "name": "DE71", "type": "giro"}).json()
|
|
r = client.patch(f"/api/accounts/{acc['id']}", headers=H,
|
|
json={"name": "DKB Giro"})
|
|
assert r.status_code == 200 and r.json()["name"] == "DKB Giro"
|
|
assert client.patch("/api/accounts/9999", headers=H,
|
|
json={"name": "x"}).status_code == 404
|
|
assert client.patch(f"/api/accounts/{acc['id']}", headers=H,
|
|
json={"name": " "}).status_code == 422
|
|
|
|
|
|
def test_patch_account_anchor_set_and_clear(client):
|
|
acc = client.post("/api/accounts", headers=H, json={
|
|
"bank": "DKB", "iban": "DE72", "name": "Anker-Test", "type": "giro"}).json()
|
|
assert acc["anchor_date"] is None
|
|
assert acc["anchor_balance"] is None
|
|
|
|
r = client.patch(f"/api/accounts/{acc['id']}", headers=H,
|
|
json={"anchor_date": "2026-06-30", "anchor_balance": "500.00"})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["anchor_date"] == "2026-06-30"
|
|
assert Decimal(body["anchor_balance"]) == Decimal("500.00")
|
|
|
|
r = client.patch(f"/api/accounts/{acc['id']}", headers=H,
|
|
json={"anchor_date": None, "anchor_balance": None})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["anchor_date"] is None
|
|
assert body["anchor_balance"] is None
|
|
|
|
|
|
def test_patch_account_anchor_requires_both_fields(client):
|
|
acc = client.post("/api/accounts", headers=H, json={
|
|
"bank": "DKB", "iban": "DE73", "name": "Anker-422", "type": "giro"}).json()
|
|
|
|
assert client.patch(f"/api/accounts/{acc['id']}", headers=H,
|
|
json={"anchor_date": "2026-06-30"}).status_code == 422
|
|
assert client.patch(f"/api/accounts/{acc['id']}", headers=H,
|
|
json={"anchor_balance": "500.00"}).status_code == 422
|
|
assert client.patch(f"/api/accounts/{acc['id']}", headers=H,
|
|
json={"anchor_date": "2026-06-30",
|
|
"anchor_balance": None}).status_code == 422
|
|
|
|
|
|
def test_patch_transaction_category(client):
|
|
cat = client.post("/api/categories", headers=H, json={"name": "Freizeit"}).json()
|
|
acc = client.post("/api/accounts", headers=H, json={
|
|
"bank": "DKB", "iban": "DE77", "name": "G3", "type": "giro"}).json()
|
|
tx = client.post("/api/transactions", headers=H, json={
|
|
"account_id": acc["id"], "booking_date": "2026-07-04",
|
|
"amount": "-15.00", "purpose": "Kino", "counterparty": ""}).json()
|
|
tx_id = tx["id"]
|
|
assert tx["category_id"] is None
|
|
|
|
r = client.patch(f"/api/transactions/{tx_id}", headers=H,
|
|
json={"category_id": cat["id"]})
|
|
assert r.status_code == 200
|
|
assert r.json()["category_id"] == cat["id"]
|
|
|
|
r = client.patch(f"/api/transactions/{tx_id}", headers=H, json={})
|
|
assert r.status_code == 200
|
|
assert r.json()["category_id"] == cat["id"]
|
|
|
|
r = client.patch(f"/api/transactions/{tx_id}", headers=H,
|
|
json={"category_id": None})
|
|
assert r.status_code == 200
|
|
assert r.json()["category_id"] is None
|
|
|
|
r = client.patch(f"/api/transactions/{tx_id}", headers=H,
|
|
json={"category_id": 9999})
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_scenario_patch_beschreibung_leeren(client):
|
|
sc = client.post("/api/scenarios", headers=H, json={"name": "DescNull", "description": "x"}).json()
|
|
r = client.patch(f"/api/scenarios/{sc['id']}", headers=H, json={"description": None})
|
|
assert r.status_code == 200 and r.json()["description"] == ""
|
|
|
|
|
|
def test_modifier_ende_validierung(client, db):
|
|
sc = client.post("/api/scenarios", headers=H, json={"name": "EndeVal"}).json()
|
|
# ende ohne Datum -> 422
|
|
r = client.post(f"/api/scenarios/{sc['id']}/modifiers", headers=H, json={
|
|
"target_type": "recurring", "target_id": 1, "kind": "ende"})
|
|
assert r.status_code == 422
|
|
# ende mit Datum -> 201, end_date in der Antwort
|
|
r = client.post(f"/api/scenarios/{sc['id']}/modifiers", headers=H, json={
|
|
"target_type": "recurring", "target_id": 1, "kind": "ende",
|
|
"end_date": "2026-08-31"})
|
|
assert r.status_code == 201
|
|
assert r.json()["end_date"] == "2026-08-31"
|
|
# andere Arten nullen end_date
|
|
r = client.post(f"/api/scenarios/{sc['id']}/modifiers", headers=H, json={
|
|
"target_type": "recurring", "target_id": 1, "kind": "remove",
|
|
"end_date": "2026-08-31"})
|
|
assert r.status_code == 201 and r.json()["end_date"] is None
|
|
|
|
|
|
def test_scenario_planned_crud_und_cleanup(client, db):
|
|
sc = client.post("/api/scenarios", headers=H, json={"name": "SPI-CRUD"}).json()
|
|
r = client.post(f"/api/scenarios/{sc['id']}/planned", headers=H, json={
|
|
"name": "Sonderzahlung", "amount": "5000.00", "due": "2026-07-30"})
|
|
assert r.status_code == 201
|
|
item = r.json()
|
|
assert item["scenario_id"] == sc["id"] and item["amount"] == "5000.00"
|
|
assert client.get(f"/api/scenarios/{sc['id']}/planned", headers=H).json()[0]["name"] == "Sonderzahlung"
|
|
# DELETE mit fremder scenario_id -> 404
|
|
other = client.post("/api/scenarios", headers=H, json={"name": "SPI-Other"}).json()
|
|
assert client.delete(f"/api/scenarios/{other['id']}/planned/{item['id']}",
|
|
headers=H).status_code == 404
|
|
# Szenario loeschen raeumt Einmalzahlungen mit ab (kein FK-Fehler):
|
|
assert client.delete(f"/api/scenarios/{sc['id']}", headers=H).status_code == 204
|
|
from app.models.tables import ScenarioPlannedItem
|
|
assert db.query(ScenarioPlannedItem).count() == 0
|
|
|
|
|
|
def test_modifier_patch(client, db):
|
|
sc = client.post("/api/scenarios", headers=H, json={"name": "ModPatch"}).json()
|
|
m = client.post(f"/api/scenarios/{sc['id']}/modifiers", headers=H, json={
|
|
"target_type": "recurring", "target_id": 1, "kind": "percent",
|
|
"value": "10.00"}).json()
|
|
# Wert aendern
|
|
r = client.patch(f"/api/scenarios/{sc['id']}/modifiers/{m['id']}", headers=H, json={
|
|
"target_type": "recurring", "target_id": 1, "kind": "percent", "value": "25.00"})
|
|
assert r.status_code == 200 and r.json()["value"] == "25.00"
|
|
# Art auf ende wechseln (mit Datum); value wird vom Validator genullt
|
|
r = client.patch(f"/api/scenarios/{sc['id']}/modifiers/{m['id']}", headers=H, json={
|
|
"target_type": "recurring", "target_id": 1, "kind": "ende",
|
|
"end_date": "2026-08-31"})
|
|
assert r.status_code == 200
|
|
assert r.json()["kind"] == "ende" and r.json()["end_date"] == "2026-08-31"
|
|
# ende ohne Datum -> 422
|
|
r = client.patch(f"/api/scenarios/{sc['id']}/modifiers/{m['id']}", headers=H, json={
|
|
"target_type": "recurring", "target_id": 1, "kind": "ende"})
|
|
assert r.status_code == 422
|
|
# zurueck auf remove: end_date wird genullt
|
|
r = client.patch(f"/api/scenarios/{sc['id']}/modifiers/{m['id']}", headers=H, json={
|
|
"target_type": "recurring", "target_id": 1, "kind": "remove",
|
|
"end_date": "2026-08-31"})
|
|
assert r.status_code == 200 and r.json()["end_date"] is None
|
|
# fremdes Szenario -> 404
|
|
other = client.post("/api/scenarios", headers=H, json={"name": "ModPatch2"}).json()
|
|
r = client.patch(f"/api/scenarios/{other['id']}/modifiers/{m['id']}", headers=H, json={
|
|
"target_type": "recurring", "target_id": 1, "kind": "remove"})
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_scenario_planned_patch(client, db):
|
|
sc = client.post("/api/scenarios", headers=H, json={"name": "SpiPatch"}).json()
|
|
p = client.post(f"/api/scenarios/{sc['id']}/planned", headers=H, json={
|
|
"name": "Zufluss", "amount": "5000.00", "due": "2026-07-30"}).json()
|
|
r = client.patch(f"/api/scenarios/{sc['id']}/planned/{p['id']}", headers=H, json={
|
|
"name": "Zufluss geaendert", "amount": "6000.00", "due": "2026-08-15"})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["name"] == "Zufluss geaendert"
|
|
assert body["amount"] == "6000.00" and body["due"] == "2026-08-15"
|
|
# fremdes Szenario -> 404
|
|
other = client.post("/api/scenarios", headers=H, json={"name": "SpiPatch2"}).json()
|
|
r = client.patch(f"/api/scenarios/{other['id']}/planned/{p['id']}", headers=H, json={
|
|
"name": "x", "amount": "1.00", "due": "2026-08-15"})
|
|
assert r.status_code == 404
|