ledgrab/server/tests/storage/test_data_migrations.py

"""Tests for the versioned data-migration runner.

These verify the safety properties that replaced the string-replace migration:

* migrations parse JSON properly (no substring corruption);
* the runner records each migration name in ``data_migrations`` so subsequent
  runs are no-ops;
* a migration that touches no rows still gets recorded as applied;
* a migration that raises does not get recorded (so a fixed migration can be
  re-attempted on the next boot);
* the legacy ``static`` rename only rewrites rows whose ``source_type`` field
  actually equals ``static`` — values that merely contain the substring (e.g.
  an animation type ``static_wave``) are left untouched.
"""

from __future__ import annotations

import json
from datetime import datetime, timezone

import pytest

from ledgrab.storage.data_migrations import (
    ALL_MIGRATIONS,
    DataMigration,
    MigrationRunner,
    StaticToSingleColorMigration,
)
from ledgrab.storage.database import Database


@pytest.fixture
def tmp_db(tmp_path):
    db = Database(tmp_path / "test.db")
    yield db
    db.close()


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _insert_css(db: Database, item_id: str, blob: dict) -> None:
    db.upsert("color_strip_sources", item_id, blob.get("name", ""), blob)


# ---------------------------------------------------------------------------
# Runner mechanics
# ---------------------------------------------------------------------------


def test_runner_creates_audit_table(tmp_db):
    MigrationRunner(tmp_db)
    cursor = tmp_db.execute(
        "SELECT name FROM sqlite_master WHERE type='table' AND name='data_migrations'"
    )
    assert cursor.fetchone() is not None


def test_runner_records_applied_migration(tmp_db):
    class _Noop(DataMigration):
        name = "test_noop"

        def apply(self, conn):
            return 0

    runner = MigrationRunner(tmp_db)
    records = runner.run([_Noop()])

    assert [r.name for r in records] == ["test_noop"]
    assert "test_noop" in runner.applied_names()


def test_runner_skips_already_applied_migration(tmp_db):
    calls = 0

    class _CountedMigration(DataMigration):
        name = "test_counted"

        def apply(self, conn):
            nonlocal calls
            calls += 1
            return 0

    runner = MigrationRunner(tmp_db)
    runner.run([_CountedMigration()])
    runner.run([_CountedMigration()])
    runner.run([_CountedMigration()])

    assert calls == 1


def test_runner_records_zero_row_migration_so_it_does_not_repeat(tmp_db):
    """Even when a migration changes 0 rows, it is recorded as applied."""

    class _NoMatch(DataMigration):
        name = "test_no_match"

        def apply(self, conn):
            return 0

    runner = MigrationRunner(tmp_db)
    runner.run([_NoMatch()])

    assert "test_no_match" in runner.applied_names()


def test_runner_does_not_record_failed_migration(tmp_db):
    class _Boom(DataMigration):
        name = "test_boom"

        def apply(self, conn):
            raise RuntimeError("kaboom")

    runner = MigrationRunner(tmp_db)
    with pytest.raises(RuntimeError, match="kaboom"):
        runner.run([_Boom()])

    assert "test_boom" not in runner.applied_names()


def test_runner_rejects_unnamed_migration(tmp_db):
    class _Unnamed(DataMigration):
        def apply(self, conn):
            return 0

    runner = MigrationRunner(tmp_db)
    with pytest.raises(ValueError, match="non-empty name"):
        runner.run([_Unnamed()])


def test_runner_rolls_back_data_change_when_migration_raises(tmp_db):
    """A migration that raises mid-way leaves no partial UPDATEs behind."""
    blob = {
        "id": "css_partial",
        "name": "Partial",
        "source_type": "static",
        "color": [10, 20, 30],
        "tags": [],
        "led_count": 30,
        "created_at": _now(),
        "updated_at": _now(),
    }
    _insert_css(tmp_db, "css_partial", blob)

    class _PartialBoom(DataMigration):
        name = "test_partial_boom"

        def apply(self, conn):
            conn.execute(
                "UPDATE color_strip_sources SET data = ? WHERE id = ?",
                (json.dumps({**blob, "source_type": "single_color"}), "css_partial"),
            )
            raise RuntimeError("explode after update")

    runner = MigrationRunner(tmp_db)
    with pytest.raises(RuntimeError, match="explode after update"):
        runner.run([_PartialBoom()])

    rows = tmp_db.load_all("color_strip_sources")
    assert rows[0]["source_type"] == "static", "UPDATE was not rolled back"
    assert "test_partial_boom" not in runner.applied_names()


def test_runner_skips_when_writes_are_frozen(tmp_db, monkeypatch):
    """Frozen-write databases must not be mutated by the runner."""
    monkeypatch.setattr("ledgrab.storage.data_migrations.is_writes_frozen", lambda: True)

    calls = 0

    class _Counted(DataMigration):
        name = "test_frozen"

        def apply(self, conn):
            nonlocal calls
            calls += 1
            return 0

    runner = MigrationRunner(tmp_db)
    records = runner.run([_Counted()])

    assert records == []
    assert calls == 0
    assert "test_frozen" not in runner.applied_names()


# ---------------------------------------------------------------------------
# Legacy static-rename specifics
# ---------------------------------------------------------------------------


def test_static_rename_only_touches_source_type_field(tmp_db):
    """A row whose ``animation.type`` happens to contain ``static`` survives intact."""
    blob = {
        "id": "css_safe01",
        "name": "Has static_wave anim",
        "source_type": "gradient",  # not "static" — should be untouched
        "animation": {"type": "static_wave", "speed": 1.0},  # substring trap
        "stops": [{"position": 0.0, "color": [10, 20, 30]}],
        "tags": [],
        "led_count": 30,
        "created_at": _now(),
        "updated_at": _now(),
    }
    _insert_css(tmp_db, "css_safe01", blob)

    # Run the migration via the runner (clears the existing audit row first
    # because the Database fixture already triggered it through any incidental
    # store creation — here we drive it directly).
    runner = MigrationRunner(tmp_db)
    runner.run([StaticToSingleColorMigration()])

    rows = tmp_db.load_all("color_strip_sources")
    assert len(rows) == 1
    assert rows[0]["source_type"] == "gradient"
    assert rows[0]["animation"]["type"] == "static_wave"


def test_static_rename_rewrites_only_legacy_rows(tmp_db):
    legacy = {
        "id": "css_legacy01",
        "name": "Legacy",
        "source_type": "static",
        "color": [255, 0, 0],
        "tags": [],
        "led_count": 30,
        "created_at": _now(),
        "updated_at": _now(),
    }
    current = {
        "id": "css_modern01",
        "name": "Modern",
        "source_type": "single_color",
        "color": [0, 128, 255],
        "tags": [],
        "led_count": 30,
        "created_at": _now(),
        "updated_at": _now(),
    }
    _insert_css(tmp_db, "css_legacy01", legacy)
    _insert_css(tmp_db, "css_modern01", current)

    runner = MigrationRunner(tmp_db)
    runner.run([StaticToSingleColorMigration()])

    rows = {r["id"]: r for r in tmp_db.load_all("color_strip_sources")}
    assert rows["css_legacy01"]["source_type"] == "single_color"
    assert rows["css_modern01"]["source_type"] == "single_color"


def test_static_rename_survives_corrupt_blob(tmp_db):
    """A row whose JSON blob is corrupt does not crash the migration."""
    # Insert a row with deliberately broken JSON via direct SQL (bypass upsert
    # which would re-serialize). We use a hand-crafted blob.
    tmp_db.execute(
        "INSERT INTO color_strip_sources (id, name, data) VALUES (?, ?, ?)",
        ("css_bad01", "Bad", "{not valid json"),
    )
    legacy = {
        "id": "css_legacy02",
        "name": "Legacy after corrupt",
        "source_type": "static",
        "color": [1, 2, 3],
        "tags": [],
        "led_count": 30,
        "created_at": _now(),
        "updated_at": _now(),
    }
    _insert_css(tmp_db, "css_legacy02", legacy)

    runner = MigrationRunner(tmp_db)
    runner.run([StaticToSingleColorMigration()])

    # Legacy row still migrated, corrupt row untouched, no exception
    cursor = tmp_db.execute("SELECT id, data FROM color_strip_sources")
    out = {row["id"]: row["data"] for row in cursor.fetchall()}
    assert out["css_bad01"] == "{not valid json"
    assert json.loads(out["css_legacy02"])["source_type"] == "single_color"


def test_all_migrations_have_unique_names():
    names = [m.name for m in ALL_MIGRATIONS]
    assert len(names) == len(set(names)), f"duplicate names: {names}"


def test_color_strip_store_construction_records_static_migration(tmp_db):
    """Integration: constructing the store applies AND records the migration."""
    from ledgrab.storage.color_strip_source import SingleColorStripSource
    from ledgrab.storage.color_strip_store import ColorStripStore

    legacy = {
        "id": "css_int01",
        "name": "Integration legacy",
        "source_type": "static",
        "color": [1, 2, 3],
        "tags": [],
        "led_count": 30,
        "created_at": _now(),
        "updated_at": _now(),
    }
    _insert_css(tmp_db, "css_int01", legacy)

    store = ColorStripStore(tmp_db)

    # In-memory shape is correct
    assert isinstance(store.get("css_int01"), SingleColorStripSource)
    # Audit row exists
    runner = MigrationRunner(tmp_db)
    assert "001_color_strip_static_to_single_color" in runner.applied_names()