rdmsandClaude Opus 5.5
979e1c0edc
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
83 lines
3.3 KiB
Python
83 lines
3.3 KiB
Python
"""Postgres-only-Prüfung der Grafana-Views (Ausbaustufe 11).
|
|
|
|
Läuft nur mit FB_TEST_PG_URL (Wegwerf-Postgres, DB-Name muss auf '_test'
|
|
enden - Schutz gegen versehentliche Ausführung gegen die Produktiv-DB).
|
|
Beispiel:
|
|
podman run -d --rm --name a11-pg -e POSTGRES_PASSWORD=wegwerf \
|
|
-e POSTGRES_DB=fb_test -p 127.0.0.1:55432:5432 docker.io/library/postgres:17.10
|
|
FB_TEST_PG_URL=postgresql+psycopg://postgres:wegwerf@127.0.0.1:55432/fb_test \
|
|
.venv/bin/python -m pytest -q tests/test_views_pg.py
|
|
"""
|
|
import os
|
|
from datetime import timedelta
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.engine import make_url
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db import Base
|
|
import app.models # noqa: F401
|
|
from app.models.tables import (Account, ProjectionPoint, ProjectionResult,
|
|
Scenario, Transaction)
|
|
from app.models.views import RESERVED_SCENARIO_NAME, VIEWS, create_views
|
|
|
|
PG_URL = os.environ.get("FB_TEST_PG_URL")
|
|
pytestmark = pytest.mark.skipif(not PG_URL, reason="FB_TEST_PG_URL nicht gesetzt")
|
|
|
|
|
|
def _reset(engine):
|
|
with engine.begin() as conn:
|
|
for name in VIEWS:
|
|
conn.execute(text(f"DROP VIEW IF EXISTS {name} CASCADE"))
|
|
Base.metadata.drop_all(engine)
|
|
|
|
|
|
@pytest.fixture
|
|
def pg():
|
|
assert make_url(PG_URL).database.endswith("_test"), "nur Wegwerf-DB *_test!"
|
|
engine = create_engine(PG_URL)
|
|
_reset(engine)
|
|
Base.metadata.create_all(engine)
|
|
create_views(engine)
|
|
with Session(engine) as session:
|
|
yield session
|
|
_reset(engine)
|
|
engine.dispose()
|
|
|
|
|
|
def test_v_projection_ist_serie_und_szenario_ab_heute(pg):
|
|
today = pg.execute(text("SELECT CURRENT_DATE")).scalar()
|
|
acc = Account(bank="t", iban="DE-PG-1", name="Testkonto", type="giro",
|
|
anchor_date=today - timedelta(days=10),
|
|
anchor_balance=Decimal("1000.00"))
|
|
pg.add(acc)
|
|
pg.flush()
|
|
pg.add(Transaction(account_id=acc.id, booking_date=today - timedelta(days=2),
|
|
amount=Decimal("-100.00"), purpose="t", counterparty="t",
|
|
status="confirmed", dedup_hash="pg-1"))
|
|
stale = Scenario(name="Veraltet", description="")
|
|
empty = Scenario(name="Ohne Punkte", description="")
|
|
pg.add_all([stale, empty])
|
|
pg.flush()
|
|
# Veraltete Momentaufnahme: Punkte von heute-5 bis heute+5.
|
|
for off in range(-5, 6):
|
|
pg.add(ProjectionPoint(scenario_id=stale.id, day=today + timedelta(days=off),
|
|
balance=Decimal("5000.00")))
|
|
pg.add(ProjectionResult(scenario_id=stale.id, computed_at=today,
|
|
low_point_date=today, low_point_balance=Decimal("0")))
|
|
pg.commit()
|
|
|
|
rows = pg.execute(text(
|
|
"SELECT scenario, day, balance FROM v_projection ORDER BY scenario, day")).all()
|
|
ist = [r for r in rows if r.scenario == RESERVED_SCENARIO_NAME]
|
|
sc = [r for r in rows if r.scenario == "Veraltet"]
|
|
assert max(r.day for r in ist) == today
|
|
assert [r.balance for r in ist if r.day == today] == [Decimal("900.00")]
|
|
# Szenario startet heute am Ist-Saldo, keine alten Punkte davor.
|
|
assert sc[0].day == today and sc[0].balance == Decimal("900.00")
|
|
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)
|