import asyncio import os import sys from pathlib import Path from sqlalchemy import inspect, text from sqlalchemy.ext.asyncio import create_async_engine from app.db import Base PROJECT_ROOT = Path(__file__).parents[2] async def run_alembic(command: str, database_url: str) -> None: process = await asyncio.create_subprocess_exec( sys.executable, "-m", "alembic", command, "head" if command == "upgrade" else "base", cwd=PROJECT_ROOT, env={**os.environ, "DATABASE_URL": database_url}, ) assert await process.wait() == 0 async def test_migration_up_and_down(postgres_url: str) -> None: engine = create_async_engine(postgres_url) async with engine.begin() as connection: await connection.run_sync(Base.metadata.drop_all) await connection.execute(text("DROP TABLE IF EXISTS alembic_version")) await engine.dispose() await run_alembic("upgrade", postgres_url) engine = create_async_engine(postgres_url) async with engine.connect() as connection: tables, indexes, unique_constraints = await connection.run_sync( lambda sync_connection: ( set(inspect(sync_connection).get_table_names()), {index["name"] for index in inspect(sync_connection).get_indexes("outbox")}, { constraint["name"] for constraint in inspect(sync_connection).get_unique_constraints("payments") }, ) ) await engine.dispose() assert {"payments", "outbox", "alembic_version"} <= tables assert {"ix_outbox_aggregate_id", "ix_outbox_published_at"} <= indexes assert unique_constraints await run_alembic("downgrade", postgres_url) engine = create_async_engine(postgres_url) async with engine.connect() as connection: remaining_tables = await connection.run_sync( lambda sync_connection: set(inspect(sync_connection).get_table_names()) ) await engine.dispose() assert "payments" not in remaining_tables assert "outbox" not in remaining_tables