rdmsandClaude Opus 5.5
979e1c0edc
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
from datetime import date, timedelta
|
|
|
|
from sqlalchemy import func, select
|
|
|
|
from app.config import get_settings
|
|
from app.models.tables import ProjectionPoint, ProjectionResult, Scenario
|
|
from app.services.projection_service import refresh_stale_projections, run_projection
|
|
|
|
TODAY = date(2026, 9, 28)
|
|
|
|
|
|
def _scenario(db, name):
|
|
sc = Scenario(name=name, description="", include_recurring=False,
|
|
include_planned=False)
|
|
db.add(sc)
|
|
db.commit()
|
|
return sc
|
|
|
|
|
|
def _first_day(db, sid):
|
|
return db.execute(select(func.min(ProjectionPoint.day))
|
|
.where(ProjectionPoint.scenario_id == sid)).scalar()
|
|
|
|
|
|
def _count(db, sid):
|
|
return db.execute(select(func.count()).select_from(ProjectionPoint)
|
|
.where(ProjectionPoint.scenario_id == sid)).scalar()
|
|
|
|
|
|
def test_veraltetes_szenario_wird_ab_heute_neu_gerechnet(db):
|
|
sc = _scenario(db, "Alt")
|
|
run_projection(db, sc, 10, TODAY - timedelta(days=5))
|
|
assert _first_day(db, sc.id) == TODAY - timedelta(days=4)
|
|
assert refresh_stale_projections(db, TODAY) == 1
|
|
assert _first_day(db, sc.id) == TODAY + timedelta(days=1)
|
|
assert _count(db, sc.id) == get_settings().horizon_days
|
|
|
|
|
|
def test_gestern_gerechnet_gilt_als_veraltet(db):
|
|
sc = _scenario(db, "Gestern")
|
|
run_projection(db, sc, 10, TODAY - timedelta(days=1))
|
|
assert refresh_stale_projections(db, TODAY) == 1
|
|
|
|
|
|
def test_heute_gerechnet_bleibt_unveraendert(db):
|
|
sc = _scenario(db, "Heute")
|
|
run_projection(db, sc, 10, TODAY)
|
|
before = db.get(ProjectionResult, sc.id).computed_at
|
|
assert refresh_stale_projections(db, TODAY) == 0
|
|
assert _count(db, sc.id) == 10
|
|
assert db.get(ProjectionResult, sc.id).computed_at == before
|
|
|
|
|
|
def test_nie_gerechnetes_szenario_bleibt_ohne_ergebnis(db):
|
|
sc = _scenario(db, "Nie")
|
|
assert refresh_stale_projections(db, TODAY) == 0
|
|
assert db.get(ProjectionResult, sc.id) is None
|
|
assert _count(db, sc.id) == 0
|
|
|
|
|
|
def test_refresh_ist_idempotent(db):
|
|
sc = _scenario(db, "Zweimal")
|
|
run_projection(db, sc, 10, TODAY - timedelta(days=3))
|
|
assert refresh_stale_projections(db, TODAY) == 1
|
|
assert refresh_stale_projections(db, TODAY) == 0
|
|
assert _count(db, sc.id) == get_settings().horizon_days
|