rdmsandClaude Opus 5.5
b0a1ba80a1
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
80 lines
3.8 KiB
Python
80 lines
3.8 KiB
Python
"""Postgres-only: Alembic-Revision e7f1a9c3b214 (Ausbaustufe 14) hoch/runter
|
|
auf einer Wegwerf-DB (Name muss auf '_test' enden). Beispiel:
|
|
podman run -d --rm --name a14-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_migration_pg.py
|
|
"""
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from sqlalchemy import create_engine, inspect, text
|
|
from sqlalchemy.engine import make_url
|
|
|
|
PG_URL = os.environ.get("FB_TEST_PG_URL")
|
|
pytestmark = pytest.mark.skipif(not PG_URL, reason="FB_TEST_PG_URL nicht gesetzt")
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
@pytest.fixture
|
|
def alembic_cfg(monkeypatch):
|
|
assert make_url(PG_URL).database.endswith("_test"), "nur Wegwerf-DB *_test!"
|
|
monkeypatch.setenv("FB_DATABASE_URL", PG_URL)
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
engine = create_engine(PG_URL)
|
|
with engine.begin() as conn:
|
|
conn.execute(text("DROP SCHEMA public CASCADE; CREATE SCHEMA public"))
|
|
cfg = Config(str(ROOT / "alembic.ini"))
|
|
cfg.set_main_option("script_location", str(ROOT / "alembic"))
|
|
yield cfg, engine
|
|
# Schema leeren, damit test_views_pg.py (create_all) nicht auf dem
|
|
# herabgestuften Alembic-Stand aufsetzt.
|
|
with engine.begin() as conn:
|
|
conn.execute(text("DROP SCHEMA public CASCADE; CREATE SCHEMA public"))
|
|
engine.dispose()
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_migration_a14_up_down(alembic_cfg):
|
|
cfg, engine = alembic_cfg
|
|
command.upgrade(cfg, "d5e8b3c7a410")
|
|
with engine.begin() as conn:
|
|
conn.execute(text("INSERT INTO scenarios (id, name, description, include_recurring, "
|
|
"include_planned) VALUES (1, 'S', '', true, true)"))
|
|
conn.execute(text(
|
|
"INSERT INTO scenario_modifiers (id, scenario_id, target_type, target_id, kind, "
|
|
"value, end_date) VALUES "
|
|
"(1, 1, 'recurring', 7, 'ende', 0, '2030-10-15'), "
|
|
"(2, 1, 'category', 3, 'percent', 10, NULL)"))
|
|
command.upgrade(cfg, "head")
|
|
with engine.begin() as conn:
|
|
rows = conn.execute(text("SELECT id, kind, value, start_date, end_date "
|
|
"FROM scenario_modifiers ORDER BY id")).all()
|
|
assert [(r.id, r.kind, str(r.value), r.start_date, str(r.end_date))
|
|
for r in rows] == [(1, "ende", "0.00", None, "2030-10-15"),
|
|
(2, "percent", "10.00", None, "None")]
|
|
conn.execute(text(
|
|
"INSERT INTO scenario_modifiers (id, scenario_id, target_type, target_id, kind, "
|
|
"value, start_date, end_date) VALUES "
|
|
"(3, 1, 'recurring', 7, 'set', -80, '2030-11-01', NULL), "
|
|
"(4, 1, 'recurring', 7, 'percent', 50, '2030-11-01', '2030-12-31')"))
|
|
conn.execute(text(
|
|
"INSERT INTO scenario_recurring_items (scenario_id, name, amount, rhythm, due_day) "
|
|
"VALUES (1, 'W', -10, 'monthly', 14)"))
|
|
insp = inspect(engine)
|
|
assert "scenario_recurring_items" in insp.get_table_names()
|
|
assert "category_id" in {c["name"] for c in insp.get_columns("scenario_planned_items")}
|
|
command.downgrade(cfg, "d5e8b3c7a410")
|
|
with engine.begin() as conn:
|
|
rows = conn.execute(text("SELECT id, kind, end_date FROM scenario_modifiers "
|
|
"ORDER BY id")).all()
|
|
assert [(r.id, r.kind, str(r.end_date)) for r in rows] == [
|
|
(1, "ende", "2030-10-15"), (2, "percent", "None"), (4, "percent", "None")]
|
|
insp = inspect(engine)
|
|
assert "scenario_recurring_items" not in insp.get_table_names()
|
|
assert "start_date" not in {c["name"] for c in insp.get_columns("scenario_modifiers")}
|