- 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>
188 lines
9.2 KiB
Python
188 lines
9.2 KiB
Python
"""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()
|