hms-mediaengine/_entwicklung/tests/unit/test_persistence.py

"""Unit-Tests SQLite-Persistenz (PLAN.md §24).

Deckung von §29.1 (Schema- und Projektmigrationen) und §24.1/§24.4:
WAL, Foreign Keys, kurze Transaktionen, Backup vor Migration,
Integritätscheck, Projekt-Roundtrip, Plugin-Status-Zyklus.
"""

from __future__ import annotations

import sqlite3
import time
import uuid
from pathlib import Path

import pytest
from hms_persistence import SCHEMA_VERSION, Database


@pytest.fixture()
def db(tmp_path: Path) -> Database:
    database = Database(tmp_path / "userdata" / "database" / "hms.db")
    database.open()
    yield database
    database.close()


def _project(name: str = "Test Show", version: int = 1) -> dict:
    return {
        "id": str(uuid.uuid4()),
        "name": name,
        "schema_version": version,
        "created_at": "2026-09-11T00:00:00+00:00",
        "updated_at": "2026-09-11T00:00:00+00:00",
        "compositions": [],
    }


# ---------- Öffnen & PRAGMA (§24.1) ----------


def test_open_sets_wal_mode(db: Database) -> None:
    mode = db.connection.execute("PRAGMA journal_mode").fetchone()
    assert mode and mode[0].lower() == "wal"


def test_open_enables_foreign_keys(db: Database) -> None:
    fk = db.connection.execute("PRAGMA foreign_keys").fetchone()
    assert fk and fk[0] == 1


def test_open_creates_parent_directories(tmp_path: Path) -> None:
    deep = tmp_path / "userdata" / "database"
    database = Database(deep / "hms.db")
    database.open()
    database.close()
    assert (deep / "hms.db").is_file()


def test_integrity_check_failure_raises(tmp_path: Path) -> None:
    """Beschädigte Datei darf nicht still geöffnet werden (§24.1)."""
    bad = tmp_path / "corrupt.db"
    bad.write_bytes(b"this is definitely not a sqlite database" * 10)
    database = Database(bad)
    with pytest.raises(sqlite3.DatabaseError):
        database.open()


# ---------- Migrationen (§24.4) ----------


def test_migrate_from_empty_sets_schema_version(db: Database) -> None:
    result = db.migrate()
    assert result.from_version == 0
    assert result.to_version == SCHEMA_VERSION
    assert result.integrity_ok
    assert db.schema_version == SCHEMA_VERSION


def test_migrate_is_idempotent(db: Database) -> None:
    db.migrate()
    second = db.migrate()
    assert second.from_version == SCHEMA_VERSION
    assert second.backup_path is None  # kein erneutes Backup ohne Änderung


def test_migrate_creates_backup_of_existing_data(tmp_path: Path) -> None:
    db_path = tmp_path / "hms.db"
    database = Database(db_path)
    database.open()
    database.migrate()
    proj = _project("Altdaten")
    database.save_project(proj)
    database.close()

    # zweites Öffnen: Migration bereits aktuell → kein Backup nötig
    database2 = Database(db_path)
    database2.open(integrity_check=False)  # Backup-Szenario simulieren
    result = database2.migrate(backup_dir=tmp_path / "backups")
    assert result.backup_path is None  # Version identisch
    database2.close()

    # Backup-Pfad existsiert nur bei echter Migration; hier simulieren wir
    # eine Vorwärtsmigration über eine Versionserhöhung nicht (SCHEMA_VERSION
    # ist 1) – stattdessen prüfen wir, dass Altdaten nach Reopen lesbar bleiben.
    database3 = Database(db_path)
    database3.open()
    loaded = database3.load_project(proj["id"])
    assert loaded is not None and loaded["name"] == "Altdaten"  # §24.4: Test mit Altdaten
    database3.close()


def test_migration_backup_before_change(monkeypatch, tmp_path: Path) -> None:
    """Backup wird vor Schemaänderung angelegt, wenn Migration läuft."""
    import hms_persistence as persistence

    db_path = tmp_path / "hms.db"
    database = Database(db_path)
    database.open()
    # Schema v1 anlegen und befüllen
    database.migrate()
    proj = _project("Vor-Backup")
    database.save_project(proj)
    # Simulation: eine neue Version existiert → Migration greift
    monkeypatch.setattr(persistence, "SCHEMA_VERSION", 2)
    monkeypatch.setitem(
        persistence._MIGRATIONS,
        2,
        "CREATE TABLE IF NOT EXISTS future_table (id TEXT PRIMARY KEY);",
    )
    backups = tmp_path / "backups"
    result = database.migrate(backup_dir=backups)
    assert result.backup_path is not None and result.backup_path.is_file()
    assert result.to_version == 2
    # Altdaten weiterhin vorhanden (Vorwärtsmigration verliert nichts)
    loaded = database.load_project(proj["id"])
    assert loaded is not None and loaded["name"] == "Vor-Backup"
    database.close()


# ---------- Projekte (§24.2) ----------


def test_project_save_load_roundtrip(db: Database) -> None:
    db.migrate()
    proj = _project("Meine Show")
    proj["compositions"] = [{"id": str(uuid.uuid4()), "layers": []}]
    db.save_project(proj)
    loaded = db.load_project(proj["id"])
    assert loaded == proj


def test_project_list_sorted_by_update(db: Database) -> None:
    db.migrate()
    older = _project("Alt")
    newer = _project("Neu")
    newer["updated_at"] = "2026-09-11T12:00:00+00:00"
    db.save_project(older)
    db.save_project(newer)
    names = [p["name"] for p in db.list_projects()]
    assert names[0] == "Neu"  # jüngstes zuerst


def test_project_delete(db: Database) -> None:
    db.migrate()
    proj = _project()
    db.save_project(proj)
    db.delete_project(proj["id"])
    assert db.load_project(proj["id"]) is None


def test_project_save_validates_required_fields(db: Database) -> None:
    db.migrate()
    with pytest.raises(ValueError, match="missing field"):
        db.save_project({"id": "x"})


def test_project_save_overwrites_same_id(db: Database) -> None:
    db.migrate()
    proj = _project("Version A")
    db.save_project(proj)
    proj["name"] = "Version B"
    proj["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%S%z")
    db.save_project(proj)
    loaded = db.load_project(proj["id"])
    assert loaded is not None and loaded["name"] == "Version B"
    assert len(db.list_projects()) == 1


# ---------- Einstellungen ----------


def test_settings_roundtrip_and_default(db: Database) -> None:
    db.migrate()
    assert db.get_setting("artnet.bind") is None
    assert db.get_setting("artnet.bind", "0.0.0.0") == "0.0.0.0"
    db.set_setting("artnet.bind", "127.0.0.1")
    db.set_setting("artnet.bind", "10.0.0.5")  # Überschreiben
    assert db.get_setting("artnet.bind") == "10.0.0.5"


# ---------- Plugin-Status (§14.5) ----------


def test_plugin_status_lifecycle(db: Database) -> None:
    db.migrate()
    db.upsert_plugin_status(
        "com.hms.fx.example_passthrough", "1.0.0", enabled=False, state="discovered"
    )
    db.upsert_plugin_status(
        "com.hms.fx.example_passthrough", "1.0.0", enabled=True, state="active"
    )
    status = db.get_plugin_status()
    entry = status["com.hms.fx.example_passthrough"]
    assert entry["enabled"] is True
    assert entry["state"] == "active"


def test_plugin_status_rejects_invalid_state(db: Database) -> None:
    db.migrate()
    with pytest.raises(ValueError, match="invalid plugin state"):
        db.upsert_plugin_status("com.x.y", "1.0.0", enabled=True, state="exploded")


def test_plugin_status_persists_across_reopen(tmp_path: Path) -> None:
    db_path = tmp_path / "hms.db"
    database = Database(db_path)
    database.open()
    database.migrate()
    database.upsert_plugin_status(
        "com.hms.fx.gaussian_blur", "1.0.0", enabled=True, state="compiled",
        package_hash="deadbeef",
    )
    database.close()

    database2 = Database(db_path)
    database2.open()
    entry = database2.get_plugin_status()["com.hms.fx.gaussian_blur"]
    assert entry["package_hash"] == "deadbeef"
    assert entry["state"] == "compiled"
    database2.close()


# ---------- Verbindungsschutz ----------


def test_connection_property_requires_open(tmp_path: Path) -> None:
    database = Database(tmp_path / "hms.db")
    with pytest.raises(RuntimeError, match="not opened"):
        _ = database.connection