feat: Grafana-Panel Planung nach Kategorie inkl. Kreditraten (A11-T3)
Neue Tabelle planning_monthly (Monatssummen je Kategorie aus wiederkehrenden Posten, Einmalposten und allen Kreditraten), Service planning_chart mit refresh_planning_chart (Advisory-Lock, komplette Neuschreibung), GUI-Hooks in index/planung_page ueber _safe_refresh, neues Barchart-Panel im Grafana-Dashboard (Panel 5). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -49,3 +49,23 @@ def test_time_window_covers_projection_horizon():
|
||||
f"Projektionshorizont von {horizon} Tagen nicht ab -> Projektionskurve "
|
||||
f"wird abgeschnitten"
|
||||
)
|
||||
|
||||
|
||||
def test_panel_planung_nach_kategorie():
|
||||
panels = {p["title"]: p for p in _load()["panels"]}
|
||||
p = panels["Planung nach Kategorie"]
|
||||
assert p["type"] == "barchart"
|
||||
assert "planning_monthly" in p["targets"][0]["rawSql"]
|
||||
assert p["options"]["stacking"] == "normal"
|
||||
|
||||
|
||||
def test_panels_ueberlappen_nicht_und_ids_eindeutig():
|
||||
panels = _load()["panels"]
|
||||
assert len({p["id"] for p in panels}) == len(panels)
|
||||
cells = set()
|
||||
for p in panels:
|
||||
g = p["gridPos"]
|
||||
for x in range(g["x"], g["x"] + g["w"]):
|
||||
for y in range(g["y"], g["y"] + g["h"]):
|
||||
assert (x, y) not in cells, p["title"]
|
||||
cells.add((x, y))
|
||||
|
||||
@@ -565,3 +565,22 @@ def test_v_projection_zeigt_ist_serie_und_keine_alten_punkte():
|
||||
assert "p.day > CURRENT_DATE" in body
|
||||
names = list(VIEWS)
|
||||
assert names.index("v_balance_total") < names.index("v_projection")
|
||||
|
||||
|
||||
def test_planung_seite_befuellt_planungs_diagramm(client, db):
|
||||
from app.models.tables import PlanningMonthly
|
||||
client.post("/login", data={"username": "admin", "password": "geheim"})
|
||||
db.add(PlannedItem(name="Kauf", amount=Decimal("-10.00"),
|
||||
due=date.today() + timedelta(days=3)))
|
||||
db.commit()
|
||||
assert client.get("/planung").status_code == 200
|
||||
assert db.query(PlanningMonthly).count() == 1
|
||||
|
||||
|
||||
def test_planungs_refresh_fehler_bricht_seiten_nicht(client, monkeypatch):
|
||||
def kaputt(session, today):
|
||||
raise RuntimeError("absichtlich kaputt")
|
||||
monkeypatch.setattr("app.routers.gui.refresh_planning_chart", kaputt)
|
||||
client.post("/login", data={"username": "admin", "password": "geheim"})
|
||||
for path in ("/", "/planung"):
|
||||
assert client.get(path).status_code == 200, path
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from app.models.tables import (Category, Loan, PlannedItem, PlanningMonthly,
|
||||
RecurringItem)
|
||||
from app.services.planning_chart import (KREDITRATEN, UNKATEGORISIERT,
|
||||
planning_monthly_totals,
|
||||
refresh_planning_chart)
|
||||
|
||||
TODAY = date(2026, 9, 28)
|
||||
M = lambda y, m: date(y, m, 1) # noqa: E731
|
||||
|
||||
|
||||
def _setup(db):
|
||||
wohnen = Category(name="Wohnen")
|
||||
vers = Category(name="Versicherung")
|
||||
eink = Category(name="Einkommen")
|
||||
db.add_all([wohnen, vers, eink])
|
||||
db.flush()
|
||||
db.add_all([
|
||||
# monatlich, ohne Start -> 15.09., 15.10., 15.11., 15.12.
|
||||
RecurringItem(name="Miete", amount=Decimal("-50.00"), rhythm="monthly",
|
||||
due_day=15, category_id=wohnen.id),
|
||||
# quartalsweise, Phase ab 01.08. -> nur 01.11. im Fenster
|
||||
RecurringItem(name="Police", amount=Decimal("-300.00"), rhythm="quarterly",
|
||||
due_day=1, start_date=date(2026, 8, 1), category_id=vers.id),
|
||||
# jährlich ohne Kategorie, Phase Dezember -> 10.12.2026
|
||||
RecurringItem(name="Beitrag", amount=Decimal("-120.00"), rhythm="yearly",
|
||||
due_day=10, start_date=date(2025, 12, 10)),
|
||||
# endet 31.10. -> nur 01.09. und 01.10.
|
||||
RecurringItem(name="Gehalt", amount=Decimal("1000.00"), rhythm="monthly",
|
||||
due_day=1, end_date=date(2026, 10, 31), category_id=eink.id),
|
||||
# Einmalposten: innerhalb (Okt), vor dem Fenster, nach dem Fenster
|
||||
PlannedItem(name="Reparatur", amount=Decimal("-200.00"),
|
||||
due=date(2026, 10, 5), category_id=wohnen.id),
|
||||
PlannedItem(name="Vorher", amount=Decimal("-999.00"), due=date(2026, 8, 31)),
|
||||
PlannedItem(name="Nachher", amount=Decimal("-999.00"), due=date(2027, 2, 1)),
|
||||
# Kredit 1.200 zu 0 % auf 12 Monate, Auszahlung 10.09. -> Raten je -100
|
||||
# am 10.10., 10.11., 10.12. (10.01.2027 liegt hinter dem Fenster).
|
||||
Loan(name="Auto", principal=Decimal("1200.00"), annual_rate_pct=Decimal("0"),
|
||||
term_months=12, payout_date=date(2026, 9, 10), repayment_type="annuity"),
|
||||
])
|
||||
db.commit()
|
||||
|
||||
|
||||
def test_planung_nach_kategorie_handrechnung(db):
|
||||
_setup(db)
|
||||
assert planning_monthly_totals(db, TODAY, 100) == {
|
||||
(M(2026, 9), "Wohnen"): Decimal("-50.00"),
|
||||
(M(2026, 10), "Wohnen"): Decimal("-250.00"),
|
||||
(M(2026, 11), "Wohnen"): Decimal("-50.00"),
|
||||
(M(2026, 12), "Wohnen"): Decimal("-50.00"),
|
||||
(M(2026, 11), "Versicherung"): Decimal("-300.00"),
|
||||
(M(2026, 12), UNKATEGORISIERT): Decimal("-120.00"),
|
||||
(M(2026, 9), "Einkommen"): Decimal("1000.00"),
|
||||
(M(2026, 10), "Einkommen"): Decimal("1000.00"),
|
||||
(M(2026, 10), KREDITRATEN): Decimal("-100.00"),
|
||||
(M(2026, 11), KREDITRATEN): Decimal("-100.00"),
|
||||
(M(2026, 12), KREDITRATEN): Decimal("-100.00"),
|
||||
}
|
||||
|
||||
|
||||
def test_kredit_auszahlung_zaehlt_nicht(db):
|
||||
_setup(db)
|
||||
totals = planning_monthly_totals(db, TODAY, 100)
|
||||
assert all(v < 0 for (m, c), v in totals.items() if c == KREDITRATEN)
|
||||
assert (M(2026, 9), KREDITRATEN) not in totals
|
||||
|
||||
|
||||
def test_nullsummen_entfallen(db):
|
||||
c = Category(name="Durchlauf")
|
||||
db.add(c)
|
||||
db.flush()
|
||||
db.add_all([
|
||||
PlannedItem(name="rein", amount=Decimal("10.00"), due=date(2026, 10, 1), category_id=c.id),
|
||||
PlannedItem(name="raus", amount=Decimal("-10.00"), due=date(2026, 10, 2), category_id=c.id),
|
||||
])
|
||||
db.commit()
|
||||
assert planning_monthly_totals(db, TODAY, 100) == {}
|
||||
|
||||
|
||||
def test_refresh_schreibt_tabelle_idempotent(db):
|
||||
_setup(db)
|
||||
n1 = refresh_planning_chart(db, TODAY)
|
||||
n2 = refresh_planning_chart(db, TODAY)
|
||||
assert n1 == n2 > 0
|
||||
assert db.execute(select(func.count()).select_from(PlanningMonthly)).scalar() == n2
|
||||
row = db.get(PlanningMonthly, (M(2026, 11), "Versicherung"))
|
||||
assert row.total == Decimal("-300.00")
|
||||
@@ -80,3 +80,14 @@ def test_v_projection_ist_serie_und_szenario_ab_heute(pg):
|
||||
assert all(r.day >= today for r in sc)
|
||||
assert [r.balance for r in sc[1:]] == [Decimal("5000.00")] * 5
|
||||
assert not any(r.scenario == "Ohne Punkte" for r in rows)
|
||||
|
||||
|
||||
def test_planning_chart_refresh_auf_postgres(pg):
|
||||
from app.models.tables import PlannedItem, PlanningMonthly
|
||||
from app.services.planning_chart import refresh_planning_chart
|
||||
today = pg.execute(text("SELECT CURRENT_DATE")).scalar()
|
||||
pg.add(PlannedItem(name="t", amount=Decimal("-5.00"), due=today + timedelta(days=2)))
|
||||
pg.commit()
|
||||
assert refresh_planning_chart(pg, today) == 1
|
||||
assert refresh_planning_chart(pg, today) == 1
|
||||
assert pg.query(PlanningMonthly).count() == 1
|
||||
|
||||
Reference in New Issue
Block a user