112 lines
3.5 KiB
Python
112 lines
3.5 KiB
Python
"""Migrations run from an empty database and from a legacy schema snapshot."""
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from photo_pipeline.db import run_migrations
|
|
|
|
|
|
def _alembic_head() -> str:
|
|
from alembic.config import Config as AlembicConfig
|
|
from alembic.script import ScriptDirectory
|
|
|
|
repo = Path(__file__).resolve().parents[2]
|
|
cfg = AlembicConfig(str(repo / "alembic.ini"))
|
|
cfg.set_main_option("script_location", str(repo / "migrations"))
|
|
return ScriptDirectory.from_config(cfg).get_current_head()
|
|
|
|
# Faithful snapshot of the donor photo_analyzer.py schema (photos + FTS + triggers).
|
|
LEGACY_SCHEMA = """
|
|
CREATE TABLE photos (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
path TEXT UNIQUE NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'pending',
|
|
phash TEXT, file_sha1 TEXT, dup_of TEXT,
|
|
description TEXT, tags TEXT, people_count INTEGER, setting TEXT,
|
|
time_of_day TEXT, season TEXT, mood TEXT, location_hint TEXT, approx_year INTEGER,
|
|
raw_response TEXT, error_message TEXT, analyzed_at TEXT, exif_written_at TEXT
|
|
);
|
|
CREATE INDEX idx_status ON photos(status);
|
|
CREATE INDEX idx_path ON photos(path);
|
|
CREATE VIRTUAL TABLE photos_fts USING fts5(
|
|
path, description, tags, mood, location_hint,
|
|
content=photos, content_rowid=id
|
|
);
|
|
"""
|
|
|
|
|
|
def _tables(db: Path) -> set[str]:
|
|
conn = sqlite3.connect(db)
|
|
try:
|
|
rows = conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
|
|
return {r[0] for r in rows}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _head_revision(db: Path) -> str:
|
|
conn = sqlite3.connect(db)
|
|
try:
|
|
return conn.execute("SELECT version_num FROM alembic_version").fetchone()[0]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_migrations_from_empty_database(tmp_path):
|
|
db = tmp_path / "empty.db"
|
|
run_migrations(f"sqlite:///{db}")
|
|
|
|
tables = _tables(db)
|
|
assert {"assets", "asset_paths"} <= tables
|
|
assert _head_revision(db) == _alembic_head()
|
|
|
|
|
|
def test_migrations_from_legacy_snapshot_preserve_existing_data(tmp_path):
|
|
db = tmp_path / "legacy.db"
|
|
conn = sqlite3.connect(db)
|
|
conn.executescript(LEGACY_SCHEMA)
|
|
conn.execute(
|
|
"INSERT INTO photos (path, status, description) VALUES (?, ?, ?)",
|
|
("album/one.jpg", "analyzed", "a red block"),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
run_migrations(f"sqlite:///{db}")
|
|
|
|
tables = _tables(db)
|
|
assert {"photos", "assets", "asset_paths"} <= tables
|
|
|
|
conn = sqlite3.connect(db)
|
|
try:
|
|
row = conn.execute("SELECT path, status, description FROM photos").fetchone()
|
|
finally:
|
|
conn.close()
|
|
assert row == ("album/one.jpg", "analyzed", "a red block")
|
|
assert _head_revision(db) == _alembic_head()
|
|
|
|
|
|
def test_migrations_are_idempotent(tmp_path):
|
|
db = tmp_path / "twice.db"
|
|
url = f"sqlite:///{db}"
|
|
run_migrations(url)
|
|
run_migrations(url) # second run is a no-op at head
|
|
assert _head_revision(db) == _alembic_head()
|
|
|
|
|
|
@pytest.mark.parametrize("expected", ["assets", "asset_paths"])
|
|
def test_identity_tables_have_versions_and_audit_columns(tmp_path, expected):
|
|
db = tmp_path / "cols.db"
|
|
run_migrations(f"sqlite:///{db}")
|
|
conn = sqlite3.connect(db)
|
|
try:
|
|
cols = {r[1] for r in conn.execute(f"PRAGMA table_info({expected})")}
|
|
finally:
|
|
conn.close()
|
|
if expected == "assets":
|
|
assert {"id", "state_version", "created_at", "updated_at", "discovered_at"} <= cols
|
|
else:
|
|
assert {"asset_id", "path", "valid_from", "valid_until"} <= cols
|