generated from john/python-template
109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
"""Tests for transcription.migrations — explicit Step 4 migration safety behavior."""
|
|
|
|
from sqlalchemy import inspect
|
|
from sqlalchemy import text
|
|
from sqlmodel import create_engine
|
|
from sqlmodel.pool import StaticPool
|
|
|
|
from transcription.migrations import apply_pending_migrations
|
|
from transcription.migrations import 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 == []
|