Fenris/tests/test_store_migration.py

"""Observation store migration unit tests (issue #48).

Tests the forward-only migration logic that underpins package upgrade
semantics: older stores are migrated, current stores pass through, and
newer stores are refused loudly.

Spec: §3.6, §9.5, §10.2
"""
import sqlite3
import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).parent.parent / "src"))

from fenris.store import (
    SCHEMA_VERSION,
    init_store,
    migrate_to_latest,
)
from fenris.status import NewerSchema, open_store_readonly


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _make_store(path: Path, version: int = 0) -> sqlite3.Connection:
    """Create a store at *path* with the given user_version."""
    conn = sqlite3.connect(str(path))
    conn.execute("PRAGMA journal_mode=WAL")
    if version == 0:
        # Fresh DB with no schema — user_version defaults to 0
        pass
    else:
        # Create a minimal schema so the DB is valid, then set version
        conn.execute("""
            CREATE TABLE IF NOT EXISTS samples (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                ts TEXT NOT NULL,
                device TEXT NOT NULL
            )
        """)
        conn.execute(f"PRAGMA user_version={version}")
    conn.commit()
    return conn


# ---------------------------------------------------------------------------
# migrate_to_latest
# ---------------------------------------------------------------------------

class TestMigrateToLatest:
    """Forward-only migration via migrate_to_latest()."""

    def test_migrates_from_zero(self, tmp_path):
        """Store at user_version=0 → SCHEMA_VERSION (fresh DB)."""
        db = tmp_path / "observations.db"
        _make_store(db, version=0)

        steps = migrate_to_latest(db)

        # SCHEMA_VERSION - 0 = SCHEMA_VERSION migration steps
        assert steps == SCHEMA_VERSION

        # Verify version was bumped
        conn = sqlite3.connect(str(db))
        v = conn.execute("PRAGMA user_version").fetchone()[0]
        conn.close()
        assert v == SCHEMA_VERSION

    def test_already_current_returns_zero(self, tmp_path):
        """Store already at SCHEMA_VERSION → 0 steps applied."""
        db = tmp_path / "observations.db"
        conn = _make_store(db, version=SCHEMA_VERSION)
        conn.close()

        steps = migrate_to_latest(db)
        assert steps == 0

    def test_refuses_newer_store(self, tmp_path):
        """Store with user_version > SCHEMA_VERSION → ValueError."""
        db = tmp_path / "observations.db"
        _make_store(db, version=SCHEMA_VERSION + 1)

        with pytest.raises(ValueError, match="newer Fenris"):
            migrate_to_latest(db)

    def test_refuses_much_newer_store(self, tmp_path):
        """Store several versions ahead → ValueError."""
        db = tmp_path / "observations.db"
        _make_store(db, version=SCHEMA_VERSION + 5)

        with pytest.raises(ValueError, match="newer Fenris"):
            migrate_to_latest(db)

    def test_store_not_corrupted_on_refusal(self, tmp_path):
        """After refusal, store is unchanged (no silent corruption)."""
        db = tmp_path / "observations.db"
        _make_store(db, version=SCHEMA_VERSION + 2)

        with pytest.raises(ValueError):
            migrate_to_latest(db)

        # Version should be unchanged
        conn = sqlite3.connect(str(db))
        v = conn.execute("PRAGMA user_version").fetchone()[0]
        conn.close()
        assert v == SCHEMA_VERSION + 2

    def test_idempotent_on_current(self, tmp_path):
        """Calling migrate_to_latest twice on a current store is safe."""
        db = tmp_path / "observations.db"
        _make_store(db, version=SCHEMA_VERSION)

        assert migrate_to_latest(db) == 0
        assert migrate_to_latest(db) == 0

    def test_migrates_intermediate_version(self, tmp_path):
        """Store at version 1 with SCHEMA_VERSION=1 → 0 steps (current)."""
        db = tmp_path / "observations.db"
        _make_store(db, version=1)
        # SCHEMA_VERSION is 1, so version 1 is current
        steps = migrate_to_latest(db)
        assert steps == 0


# ---------------------------------------------------------------------------
# init_store — downgrade refusal
# ---------------------------------------------------------------------------

class TestInitStoreDowngradeRefusal:
    """init_store() refuses newer-schema stores."""

    def test_refuses_newer_store(self, tmp_path):
        """init_store raises ValueError on newer-schema store."""
        db = tmp_path / "observations.db"
        _make_store(db, version=SCHEMA_VERSION + 1)

        with pytest.raises(ValueError, match="newer Fenris"):
            init_store(db)

    def test_store_not_corrupted_on_refusal(self, tmp_path):
        """After init_store refusal, store is unchanged."""
        db = tmp_path / "observations.db"
        _make_store(db, version=SCHEMA_VERSION + 1)

        with pytest.raises(ValueError):
            init_store(db)

        conn = sqlite3.connect(str(db))
        v = conn.execute("PRAGMA user_version").fetchone()[0]
        conn.close()
        assert v == SCHEMA_VERSION + 1


# ---------------------------------------------------------------------------
# open_store_readonly — downgrade refusal
# ---------------------------------------------------------------------------

class TestOpenStoreReadonlyDowngradeRefusal:
    """open_store_readonly() raises NewerSchema on newer-schema stores."""

    def test_raises_newer_schema(self, tmp_path):
        """Newer store → NewerSchema exception."""
        db = tmp_path / "observations.db"
        _make_store(db, version=SCHEMA_VERSION + 1)

        with pytest.raises(NewerSchema) as exc_info:
            open_store_readonly(db)

        assert exc_info.value.version == SCHEMA_VERSION + 1

    def test_store_not_corrupted(self, tmp_path):
        """After NewerSchema refusal, store is unchanged."""
        db = tmp_path / "observations.db"
        _make_store(db, version=SCHEMA_VERSION + 3)

        with pytest.raises(NewerSchema):
            open_store_readonly(db)

        conn = sqlite3.connect(str(db))
        v = conn.execute("PRAGMA user_version").fetchone()[0]
        conn.close()
        assert v == SCHEMA_VERSION + 3

    def test_current_store_opens(self, tmp_path):
        """Store at SCHEMA_VERSION opens without error."""
        db = tmp_path / "observations.db"
        _make_store(db, version=SCHEMA_VERSION)

        conn = open_store_readonly(db)
        assert conn is not None
        conn.close()