p42.f-w/backend/tests/test_migrationen.py

"""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)