Files
p42.f-w/backend/tests/test_migrationen.py
guyg1474andClaude Opus 5.5 1d7497bead Reisen: Wohnorte mit Datum, Anreise je Besuch, Umzüge in der Kilometer-Rechnung
- Wohnorte als Liste mit "seit"-Datum (Tabelle travel_homes, ersetzt den
  einzelnen Wohnort; Migration Schritt 5 übernimmt ihn): Reisen starten
  beim damals gültigen Wohnort, Wechsel zählen als Umzug, Tage am
  Wohnort nicht als Reisetage
- "In dieser Zeit hier gewohnt" im Besuch: legt Wohnort ab Beginn und den
  vorherigen wieder ab Ende + 1 Tag an
- Anreise je Besuch (automatisch / von zuhause / vom vorherigen Ort):
  Rundreisen auch mit mehreren Tagen zwischen den Orten
- Fix: Besuche bleiben bearbeitbar, auch wenn eine Fotodatei fehlt;
  Speicherfehler erscheinen im Dialog statt verdeckt dahinter

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 20:40:40 +02:00

188 lines
9.2 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Alte Datenbank-Stände nachbauen und prüfen, dass der Start sie verlustfrei auf den aktuellen Stand bringt."""
import sqlite3
import pytest
from app import db
T = "2026-01-01T00:00:00.000Z"
KANBAN_PROJECTS = """CREATE TABLE kanban_projects (
id INTEGER PRIMARY KEY, name TEXT NOT NULL, description TEXT NOT NULL DEFAULT '', position REAL NOT NULL,
completed_at TEXT, created_at TEXT NOT NULL, created_by INTEGER, updated_at TEXT NOT NULL, updated_by INTEGER)"""
# Kanban-Aufgaben vor "erledigt seit"
KANBAN_TASKS_OHNE_DONE = """CREATE TABLE kanban_tasks (
id INTEGER PRIMARY KEY, project_id INTEGER NOT NULL REFERENCES kanban_projects(id) ON DELETE CASCADE,
title TEXT NOT NULL, description TEXT NOT NULL DEFAULT '', status TEXT NOT NULL, assignee INTEGER,
priority INTEGER NOT NULL DEFAULT 2, position REAL NOT NULL, created_at TEXT NOT NULL, created_by INTEGER,
updated_at TEXT NOT NULL, updated_by INTEGER)"""
# Reise-Städte vor Höhe, Zeitzone und Hauptstadt
TRAVEL_ALT = """CREATE TABLE travel_cities (
id INTEGER PRIMARY KEY, user_id INTEGER NOT NULL, geoname_id INTEGER NOT NULL, name TEXT NOT NULL,
admin1 TEXT NOT NULL DEFAULT '', country TEXT NOT NULL, lat REAL NOT NULL, lon REAL NOT NULL,
population INTEGER, created_at TEXT NOT NULL, UNIQUE (user_id, geoname_id))"""
# Vorrat/Einkauf ganz alt: ohne Einheit
HAUSHALT_OHNE_EINHEIT = """CREATE TABLE household_items (
id INTEGER PRIMARY KEY, key TEXT NOT NULL, name TEXT NOT NULL, list TEXT NOT NULL, quantity INTEGER NOT NULL,
updated_at TEXT NOT NULL, updated_by INTEGER, UNIQUE (key, list))"""
# Vorrat/Einkauf mit Einheit, aber ohne Listen (Stand auf dem Server vor den Listen)
HAUSHALT_OHNE_LISTEN = """CREATE TABLE household_items (
id INTEGER PRIMARY KEY, key TEXT NOT NULL, name TEXT NOT NULL, list TEXT NOT NULL, quantity REAL NOT NULL,
unit TEXT NOT NULL DEFAULT 'stk', updated_at TEXT NOT NULL, updated_by INTEGER, UNIQUE (key, list, unit))"""
# Zwischenstand: mehrere Einkaufslisten, Vorrat ohne Liste (list_id NULL)
ZWISCHENSTAND = [
"""CREATE TABLE shopping_lists (id INTEGER PRIMARY KEY, name TEXT NOT NULL, everyone INTEGER NOT NULL DEFAULT 1,
position REAL NOT NULL, created_at TEXT NOT NULL, created_by INTEGER)""",
"""CREATE TABLE household_items (
id INTEGER PRIMARY KEY, key TEXT NOT NULL, name TEXT NOT NULL, list TEXT NOT NULL,
list_id INTEGER REFERENCES shopping_lists(id) ON DELETE CASCADE, quantity REAL NOT NULL,
unit TEXT NOT NULL DEFAULT 'stk', updated_at TEXT NOT NULL, updated_by INTEGER,
CHECK ((list = 'einkauf') = (list_id IS NOT NULL)))""",
"CREATE UNIQUE INDEX household_vorrat ON household_items (key, unit) WHERE list_id IS NULL",
"CREATE UNIQUE INDEX household_einkauf ON household_items (list_id, key, unit) WHERE list_id IS NOT NULL",
f"INSERT INTO shopping_lists VALUES (1, 'Einkaufsliste', 1, 1, '{T}', NULL)",
]
@pytest.fixture
def old_db(tmp_path, monkeypatch):
"""Leere Datei mit den angegebenen alten Tabellen; danach db.init() wie beim Serverstart."""
path = tmp_path / "alt.db"
monkeypatch.setattr(db, "DB_FILE", path)
def build(*statements):
c = sqlite3.connect(path)
for sql in statements:
c.execute(sql)
c.commit()
c.close()
db.init()
conn = db.connect()
return conn
return build
def household(conn):
return sorted(tuple(r) for r in conn.execute("SELECT name, list, list_id, quantity, unit, checked FROM household_items"))
def test_haushalt_ohne_einheit(old_db):
conn = old_db(
HAUSHALT_OHNE_EINHEIT,
f"INSERT INTO household_items VALUES (1, 'milch', 'Milch', 'vorrat', 2, '{T}', NULL)",
f"INSERT INTO household_items VALUES (2, 'brot', 'Brot', 'einkauf', 1, '{T}', NULL)",
)
assert [tuple(r) for r in conn.execute("SELECT id, name FROM shopping_lists")] == [(1, "Zuhause")]
assert household(conn) == [("Brot", "einkauf", 1, 1, "stk", 0), ("Milch", "vorrat", 1, 2, "stk", 0)]
def test_haushalt_ohne_listen(old_db):
conn = old_db(
HAUSHALT_OHNE_LISTEN,
f"INSERT INTO household_items VALUES (1, 'mehl', 'Mehl', 'vorrat', 1.5, 'kg', '{T}', NULL)",
f"INSERT INTO household_items VALUES (2, 'mehl', 'Mehl', 'einkauf', 2, 'stk', '{T}', NULL)",
)
assert [r["name"] for r in conn.execute("SELECT name FROM shopping_lists")] == ["Zuhause"]
assert household(conn) == [("Mehl", "einkauf", 1, 2, "stk", 0), ("Mehl", "vorrat", 1, 1.5, "kg", 0)]
def test_haushalt_zwischenstand(old_db):
conn = old_db(
*ZWISCHENSTAND,
f"INSERT INTO household_items VALUES (1, 'reis', 'Reis', 'vorrat', NULL, 1, 'stk', '{T}', NULL)",
f"INSERT INTO household_items VALUES (2, 'reis', 'Reis', 'einkauf', 1, 3, 'stk', '{T}', NULL)",
)
# "Einkaufsliste" heißt jetzt "Zuhause" und enthält auch den Vorrat
assert [tuple(r) for r in conn.execute("SELECT id, name FROM shopping_lists")] == [(1, "Zuhause")]
assert household(conn) == [("Reis", "einkauf", 1, 3, "stk", 0), ("Reis", "vorrat", 1, 1, "stk", 0)]
indexes = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type = 'index'")}
assert not {"household_vorrat", "household_einkauf"} & indexes
def test_kanban_ohne_erledigt_zeitpunkt(old_db):
conn = old_db(
KANBAN_PROJECTS,
KANBAN_TASKS_OHNE_DONE,
f"INSERT INTO kanban_projects VALUES (1, 'P', '', 1, NULL, '{T}', NULL, '{T}', NULL)",
f"INSERT INTO kanban_tasks VALUES (1, 1, 'fertig', '', 'erledigt', NULL, 2, 1, '{T}', NULL, '{T}', NULL)",
f"INSERT INTO kanban_tasks VALUES (2, 1, 'offen', '', 'offen', NULL, 2, 2, '{T}', NULL, '{T}', NULL)",
)
rows = dict(conn.execute("SELECT title, done_at FROM kanban_tasks").fetchall())
# Erledigte zählen ab dem Update – nicht ab dem alten Änderungsdatum (sonst sofort gelöscht)
assert rows["fertig"] > T and rows["offen"] is None
def test_reisen_ohne_neue_spalten(old_db):
conn = old_db(
TRAVEL_ALT,
f"INSERT INTO travel_cities VALUES (1, 1, 2950159, 'Berlin', '', 'DE', 52.5, 13.4, NULL, '{T}')",
)
row = conn.execute("SELECT name, elevation, timezone, capital FROM travel_cities").fetchone()
assert tuple(row) == ("Berlin", None, None, None)
def test_zweiter_start_aendert_nichts(old_db):
conn = old_db(
HAUSHALT_OHNE_LISTEN,
f"INSERT INTO household_items VALUES (1, 'mehl', 'Mehl', 'vorrat', 1, 'stk', '{T}', NULL)",
)
before = household(conn), [tuple(r) for r in conn.execute("SELECT * FROM shopping_lists")]
db.init()
assert (household(conn), [tuple(r) for r in conn.execute("SELECT * FROM shopping_lists")]) == before
def test_geloeschte_listen_kommen_nicht_zurueck(conn):
with conn:
conn.execute("DELETE FROM shopping_lists")
db.init()
assert conn.execute("SELECT COUNT(*) FROM shopping_lists").fetchone()[0] == 0
def test_neue_datenbank_hat_neueste_version(conn):
from app import migrations
assert conn.execute("PRAGMA user_version").fetchone()[0] == len(migrations.STEPS)
def test_alte_datenbank_bekommt_neueste_version(old_db):
from app import migrations
conn = old_db(HAUSHALT_OHNE_LISTEN)
assert conn.execute("PRAGMA user_version").fetchone()[0] == len(migrations.STEPS)
def test_schritt_laeuft_nur_einmal(old_db, monkeypatch):
from app import migrations
calls = []
conn = old_db(HAUSHALT_OHNE_LISTEN)
monkeypatch.setattr(migrations, "STEPS", [*migrations.STEPS, lambda c: calls.append(1)])
db.init()
db.init()
assert calls == [1]
assert conn.execute("PRAGMA user_version").fetchone()[0] == len(migrations.STEPS)
def test_uralte_aufgaben_umbau_laesst_neue_tabellen_heil(old_db):
"""Aufgaben ohne Projekt (ganz alter Stand): Umbau darf die Verweise von Checkliste/Abhängigkeiten nicht verbiegen."""
conn = old_db(
"""CREATE TABLE kanban_tasks (id INTEGER PRIMARY KEY, title TEXT NOT NULL, description TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL, assignee INTEGER, position REAL NOT NULL, created_at TEXT NOT NULL, created_by INTEGER,
updated_at TEXT NOT NULL, updated_by INTEGER)""",
f"INSERT INTO kanban_tasks VALUES (1, 'alt', '', 'offen', NULL, 1, '{T}', NULL, '{T}', NULL)",
)
with conn:
conn.execute("INSERT INTO kanban_checklist (task_id, text, done, position) VALUES (1, 'x', 0, 0)")
sql = conn.execute("SELECT sql FROM sqlite_master WHERE name = 'kanban_checklist'").fetchone()[0]
assert "kanban_tasks_alt" not in sql
def test_einzelner_wohnort_wird_liste(conn, login):
"""Zwischenstand mit travel_home (ein Wohnort ohne Datum) → travel_homes, Besuche bekommen "continued"."""
anna = login("anna")
with conn:
conn.execute("""CREATE TABLE travel_home (user_id INTEGER PRIMARY KEY, geoname_id INTEGER NOT NULL, name TEXT NOT NULL,
country TEXT NOT NULL, lat REAL NOT NULL, lon REAL NOT NULL)""")
conn.execute("INSERT INTO travel_home VALUES (?, 2950159, 'Berlin', 'DE', 52.5, 13.4)", (anna.user["id"],))
conn.execute("PRAGMA user_version = 4")
db.init()
assert [(h["name"], h["since"]) for h in anna.get("/api/reisen").json()["homes"]] == [("Berlin", None)]
assert not conn.execute("SELECT 1 FROM sqlite_master WHERE name = 'travel_home'").fetchone()