rdmsandClaude Opus 5.5
2aed685d7e
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
112 lines
5.1 KiB
Python
112 lines
5.1 KiB
Python
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
from app.engine.scenario import (PlainModifier, PlainRecurring, build_cashflows,
|
|
effective_end, modified_amount)
|
|
from app.models.tables import (Category, Loan, PlannedItem, RecurringItem,
|
|
Scenario, ScenarioLoan, ScenarioModifier,
|
|
ScenarioPlannedItem, ScenarioRecurringItem)
|
|
from app.services.projection_service import load_scenario_inputs
|
|
|
|
|
|
def test_modifikator_helfer_oeffentlich():
|
|
mods = [PlainModifier("recurring", 1, "percent", Decimal("10")),
|
|
PlainModifier("recurring", 1, "ende", Decimal("0"), date(2030, 6, 30))]
|
|
assert modified_amount(Decimal("-100.00"), None, 1, mods, on=date(2030, 1, 1)) == Decimal("-90.00")
|
|
assert modified_amount(Decimal("-100.00"), None, 2, mods, on=date(2030, 1, 1)) == Decimal("-100.00")
|
|
r = PlainRecurring(1, "R", Decimal("-100.00"), "monthly", 1, None,
|
|
date(2030, 12, 31), None)
|
|
assert effective_end(r, mods) == date(2030, 6, 30)
|
|
|
|
|
|
def _daten(db):
|
|
rec = RecurringItem(name="R", amount=Decimal("-10.00"), rhythm="monthly", due_day=1)
|
|
pln = PlannedItem(name="P", amount=Decimal("-20.00"), due=date(2030, 1, 1))
|
|
k1 = Loan(name="K1", principal=Decimal("100.00"), annual_rate_pct=Decimal("0"),
|
|
term_months=2, payout_date=date(2030, 1, 1), repayment_type="annuity")
|
|
k2 = Loan(name="K2", principal=Decimal("200.00"), annual_rate_pct=Decimal("0"),
|
|
term_months=2, payout_date=date(2030, 1, 1), repayment_type="annuity")
|
|
sc = Scenario(name="S", description="", include_recurring=True,
|
|
include_planned=False)
|
|
db.add_all([rec, pln, k1, k2, sc])
|
|
db.flush()
|
|
db.add_all([
|
|
ScenarioLoan(scenario_id=sc.id, loan_id=k1.id),
|
|
ScenarioModifier(scenario_id=sc.id, target_type="recurring",
|
|
target_id=rec.id, kind="percent", value=Decimal("50")),
|
|
ScenarioPlannedItem(scenario_id=sc.id, name="SP", amount=Decimal("-5.00"),
|
|
due=date(2030, 2, 1)),
|
|
])
|
|
db.commit()
|
|
return sc
|
|
|
|
|
|
def test_load_scenario_inputs_szenario(db):
|
|
sc = _daten(db)
|
|
inp = load_scenario_inputs(db, sc)
|
|
assert [r.name for r in inp.recurring] == ["R"]
|
|
# include_planned=False: globale Einmalposten fehlen, Szenario-eigene zählen immer.
|
|
assert [p.name for p in inp.planned] == ["SP"]
|
|
assert inp.planned[0].category_id is None
|
|
assert [l.name for l in inp.loans] == ["K1"]
|
|
assert [(m.kind, m.value) for m in inp.modifiers] == [("percent", Decimal("50"))]
|
|
|
|
|
|
def test_load_scenario_inputs_ohne_fixposten(db):
|
|
sc = _daten(db)
|
|
sc.include_recurring = False
|
|
db.commit()
|
|
assert load_scenario_inputs(db, sc).recurring == []
|
|
|
|
|
|
def test_load_scenario_inputs_system_plan(db):
|
|
_daten(db)
|
|
inp = load_scenario_inputs(db, None)
|
|
assert [r.name for r in inp.recurring] == ["R"]
|
|
assert [p.name for p in inp.planned] == ["P"]
|
|
assert [l.name for l in inp.loans] == ["K1", "K2"]
|
|
assert inp.modifiers == []
|
|
|
|
|
|
def test_szenario_zahlung_zaehlt_immer_und_id_kollision(db):
|
|
rec = RecurringItem(name="R", amount=Decimal("-10.00"), rhythm="monthly", due_day=1)
|
|
sc = Scenario(name="S-A14", description="", include_recurring=False,
|
|
include_planned=False)
|
|
db.add_all([rec, sc])
|
|
db.flush()
|
|
sri = ScenarioRecurringItem(scenario_id=sc.id, name="SR", amount=Decimal("-100.00"),
|
|
rhythm="monthly", due_day=1)
|
|
db.add(sri)
|
|
db.flush()
|
|
# Posten-Eintrag, dessen target_id zufällig der ID der Szenario-Zahlung entspricht:
|
|
db.add(ScenarioModifier(scenario_id=sc.id, target_type="recurring",
|
|
target_id=sri.id, kind="remove", value=Decimal("0")))
|
|
db.commit()
|
|
inp = load_scenario_inputs(db, sc)
|
|
# include_recurring=False: globaler Posten fehlt, Szenario-Zahlung zählt, id=None.
|
|
assert [(r.id, r.name) for r in inp.recurring] == [(None, "SR")]
|
|
flows = build_cashflows(inp.recurring, inp.planned, [], [], inp.modifiers,
|
|
date(2030, 1, 1), date(2030, 1, 31))
|
|
assert flows == [(date(2030, 1, 1), Decimal("-100.00"))]
|
|
# System-Plan kennt keine Szenario-Zahlungen.
|
|
assert [r.name for r in load_scenario_inputs(db, None).recurring] == ["R"]
|
|
|
|
|
|
def test_zeitraum_und_kategorie_werden_geladen(db):
|
|
cat = Category(name="K-A14")
|
|
sc = Scenario(name="S-A14b", description="")
|
|
db.add_all([cat, sc])
|
|
db.flush()
|
|
db.add_all([
|
|
ScenarioModifier(scenario_id=sc.id, target_type="category", target_id=cat.id,
|
|
kind="percent", value=Decimal("50"),
|
|
start_date=date(2030, 2, 1), end_date=date(2030, 3, 31)),
|
|
ScenarioPlannedItem(scenario_id=sc.id, name="E", amount=Decimal("-5.00"),
|
|
due=date(2030, 2, 1), category_id=cat.id),
|
|
])
|
|
db.commit()
|
|
inp = load_scenario_inputs(db, sc)
|
|
m = inp.modifiers[0]
|
|
assert (m.start_date, m.end_date) == (date(2030, 2, 1), date(2030, 3, 31))
|
|
assert [(p.name, p.category_id) for p in inp.planned] == [("E", cat.id)]
|