34 lines
1.6 KiB
Python
34 lines
1.6 KiB
Python
import asyncio
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import uuid
|
|
|
|
import asyncpg
|
|
import pytest
|
|
from sqlalchemy.engine import make_url
|
|
|
|
|
|
@pytest.mark.skipif(not os.getenv("TEST_DATABASE_URL", "").startswith("postgresql"), reason="requires disposable PostgreSQL")
|
|
@pytest.mark.parametrize("baseline", [None, "20260701_perf_indexes"])
|
|
async def test_fresh_migrations_and_schema_check(baseline):
|
|
url = make_url(os.environ["TEST_DATABASE_URL"])
|
|
temporary_database = "lottery_migration_test_" + uuid.uuid4().hex
|
|
admin_url = url.set(drivername="postgresql").render_as_string(hide_password=False)
|
|
connection = await asyncpg.connect(admin_url)
|
|
# Database name is generated here; the configured test database is never dropped.
|
|
await connection.execute(f'CREATE DATABASE "{temporary_database}"')
|
|
try:
|
|
environment = dict(os.environ, DATABASE_URL=url.set(database=temporary_database).render_as_string(hide_password=False))
|
|
def check(*command):
|
|
result = subprocess.run([sys.executable, *command], env=environment, capture_output=True, text=True, timeout=60)
|
|
assert result.returncode == 0, result.stderr
|
|
if baseline:
|
|
await asyncio.to_thread(check, "-m", "alembic", "upgrade", baseline)
|
|
await asyncio.to_thread(check, "-m", "alembic", "upgrade", "head")
|
|
await asyncio.to_thread(check, "-m", "alembic", "upgrade", "head")
|
|
await asyncio.to_thread(check, "scripts/check_schema.py")
|
|
finally:
|
|
await connection.execute(f'DROP DATABASE "{temporary_database}" WITH (FORCE)')
|
|
await connection.close()
|