test(upgrade): add migration unit tests and enhance packaging upgrade tests for #48
Add comprehensive test coverage for upgrade semantics: - 12 Python unit tests for store migration (forward-only, downgrade refusal, idempotent behavior) across migrate_to_latest, init_store, and open_store_readonly - Enhanced packaging upgrade test to verify all five acceptance criteria: snapshot before migration, store not rebuilt, config survival, timer/removal no-ops during upgrade - RPM-specific test for config file preservation (noreplace conffile) Co-authored-by: CommandCodeBot <noreply@commandcode.ai>
This commit is contained in:
xavierk
co-authored by
CommandCodeBot
CommandCodeBot
parent
e9d6881e38
commit
c91ca10df7
@@ -0,0 +1,195 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user