89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
"""Guard against model/migration drift.
|
|
|
|
The functional suite builds its schema with ``Base.metadata.create_all`` for speed, so a
|
|
column or index added to a model but never written into an Alembic migration would only
|
|
surface on the first real deployment. This test runs the migration chain from an empty
|
|
database and asserts that Alembic's autogenerate sees nothing left to do.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from alembic.autogenerate import compare_metadata
|
|
from alembic.config import Config
|
|
from alembic.runtime.migration import MigrationContext
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.engine import make_url
|
|
|
|
from alembic import command
|
|
from app.core.config import get_settings
|
|
from app.models import Base
|
|
|
|
BACKEND_DIR = Path(__file__).resolve().parents[1]
|
|
SCRATCH_DB = "mobilityops_migration_check"
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def migrated_database_url() -> str:
|
|
settings = get_settings()
|
|
base_url = make_url(settings.database_url)
|
|
admin_engine = create_engine(
|
|
base_url.set(database="postgres"), isolation_level="AUTOCOMMIT", poolclass=None
|
|
)
|
|
with admin_engine.connect() as conn:
|
|
conn.execute(text(f'DROP DATABASE IF EXISTS "{SCRATCH_DB}"'))
|
|
conn.execute(text(f'CREATE DATABASE "{SCRATCH_DB}"'))
|
|
scratch_url = base_url.set(database=SCRATCH_DB).render_as_string(hide_password=False)
|
|
try:
|
|
yield scratch_url
|
|
finally:
|
|
admin_engine.dispose()
|
|
admin_engine = create_engine(
|
|
base_url.set(database="postgres"), isolation_level="AUTOCOMMIT", poolclass=None
|
|
)
|
|
with admin_engine.connect() as conn:
|
|
conn.execute(text(f'DROP DATABASE IF EXISTS "{SCRATCH_DB}" WITH (FORCE)'))
|
|
admin_engine.dispose()
|
|
|
|
|
|
def _alembic_config(database_url: str) -> Config:
|
|
config = Config(str(BACKEND_DIR / "alembic.ini"))
|
|
config.set_main_option("script_location", str(BACKEND_DIR / "alembic"))
|
|
config.set_main_option("sqlalchemy.url", database_url)
|
|
return config
|
|
|
|
|
|
def test_migrations_upgrade_from_empty_and_match_models(migrated_database_url: str) -> None:
|
|
config = _alembic_config(migrated_database_url)
|
|
# env.py reads DATABASE_URL from settings; override it for the scratch database.
|
|
import os
|
|
|
|
previous = os.environ.get("DATABASE_URL")
|
|
os.environ["DATABASE_URL"] = migrated_database_url
|
|
get_settings.cache_clear()
|
|
try:
|
|
command.upgrade(config, "head")
|
|
finally:
|
|
if previous is None:
|
|
os.environ.pop("DATABASE_URL", None)
|
|
else:
|
|
os.environ["DATABASE_URL"] = previous
|
|
get_settings.cache_clear()
|
|
|
|
engine = create_engine(migrated_database_url)
|
|
try:
|
|
with engine.connect() as conn:
|
|
context = MigrationContext.configure(
|
|
conn, opts={"compare_type": True, "compare_server_default": False}
|
|
)
|
|
diff = compare_metadata(context, Base.metadata)
|
|
finally:
|
|
engine.dispose()
|
|
|
|
assert diff == [], (
|
|
"Models and Alembic migrations have drifted; write a migration for:\n"
|
|
+ "\n".join(repr(entry) for entry in diff)
|
|
)
|