US01-02: Establish Application and Database Foundation (#47)

tests/integration/test_migrations.py

@@ -0,0 +1,101 @@
+"""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
+
+# 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) == "0001_initial_identity"
+
+
+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) == "0001_initial_identity"
+
+
+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) == "0001_initial_identity"
+
+
+@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