k1nq
65 lines
2.1 KiB
Python
65 lines
2.1 KiB
Python
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
|