ver1 - Step 4 implemented - Revision functionality added

tests/test_migrations.py

@@ -0,0 +1,106 @@
+"""Tests for transcription.migrations — explicit Step 4 migration safety behavior."""
+
+from sqlalchemy import inspect, text
+from sqlmodel import create_engine
+from sqlmodel.pool import StaticPool
+
+from transcription.migrations import apply_pending_migrations, list_pending_migrations
+
+
+def _in_memory_engine():
+    """Create isolated in-memory SQLite engine."""
+    return create_engine(
+        "sqlite://",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+
+
+class TestMigrations:
+    """Verify migration listing and application behavior."""
+
+    def test_list_pending_returns_all_before_apply(self):
+        """All known migrations are pending on a fresh legacy-shaped database."""
+        engine = _in_memory_engine()
+
+        with engine.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    CREATE TABLE job (
+                        id TEXT PRIMARY KEY,
+                        document_id TEXT NOT NULL,
+                        status TEXT NOT NULL,
+                        created_at TEXT NOT NULL,
+                        updated_at TEXT NOT NULL
+                    )
+                    """
+                )
+            )
+
+        pending = list_pending_migrations(engine=engine)
+        assert [migration.revision_id for migration in pending] == [
+            "0001_add_retry_count_to_job",
+            "0002_create_transcriptrevision_table",
+        ]
+
+    def test_apply_pending_migrations_records_history_and_schema(self):
+        """Applying pending migrations mutates schema and records revision history."""
+        engine = _in_memory_engine()
+
+        with engine.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    CREATE TABLE job (
+                        id TEXT PRIMARY KEY,
+                        document_id TEXT NOT NULL,
+                        status TEXT NOT NULL,
+                        created_at TEXT NOT NULL,
+                        updated_at TEXT NOT NULL
+                    )
+                    """
+                )
+            )
+
+        applied = apply_pending_migrations(engine=engine)
+        assert applied == [
+            "0001_add_retry_count_to_job",
+            "0002_create_transcriptrevision_table",
+        ]
+
+        inspector = inspect(engine)
+        job_columns = {column["name"] for column in inspector.get_columns("job")}
+        assert "retry_count" in job_columns
+        assert "transcriptrevision" in set(inspector.get_table_names())
+
+        with engine.begin() as connection:
+            rows = connection.execute(
+                text("SELECT revision_id FROM schema_migration_history ORDER BY revision_id")
+            ).fetchall()
+        assert [row[0] for row in rows] == applied
+
+    def test_apply_pending_migrations_is_idempotent(self):
+        """Re-running apply_pending_migrations with no pending revisions is a no-op."""
+        engine = _in_memory_engine()
+
+        with engine.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    CREATE TABLE job (
+                        id TEXT PRIMARY KEY,
+                        document_id TEXT NOT NULL,
+                        status TEXT NOT NULL,
+                        created_at TEXT NOT NULL,
+                        updated_at TEXT NOT NULL
+                    )
+                    """
+                )
+            )
+
+        first_apply = apply_pending_migrations(engine=engine)
+        second_apply = apply_pending_migrations(engine=engine)
+
+        assert len(first_apply) == 2
+        assert second_apply == []