leocrm/scripts/test_migrations.sh

#!/usr/bin/env bash
# Migration Test: Verify alembic upgrade head works on an empty database.
# Also verifies alembic downgrade base works.
#
# Usage: bash scripts/test_migrations.sh [DATABASE_URL]
# If DATABASE_URL not provided, uses DATABASE_URL env var.
#
# F21 (Astra P1) hardening:
#   - Sets BOTH DATABASE_URL and MIGRATION_DATABASE_URL to the temporary
#     test database. alembic/env.py prefers MIGRATION_DATABASE_URL — setting
#     only DATABASE_URL made the test run upgrade/downgrade against whatever
#     MIGRATION_DATABASE_URL pointed at (potentially a REAL database).
#   - Verifies the actual connection target BEFORE running any DDL:
#     the current database must be the freshly created test database.
#   - psql URLs: strips the SQLAlchemy driver (+asyncpg/+psycopg2) so psql
#     receives a plain postgresql:// connection string.
#   - Cleanup via trap — the test database is dropped even on failure/interrupt.
#
# Exit codes:
#   0 = all migrations pass
#   1 = upgrade failed / target verification failed
#   2 = downgrade failed
#   3 = data integrity check failed

set -euo pipefail

DB_URL="${1:-${DATABASE_URL:-}}"
if [ -z "$DB_URL" ]; then
    echo "ERROR: DATABASE_URL not set"
    echo "Usage: bash scripts/test_migrations.sh [DATABASE_URL]"
    exit 1
fi

# Convert asyncpg URL to psycopg2 for alembic (DDL operations)
DB_URL_PSYNC="${DB_URL/postgresql+asyncpg/postgresql+psycopg2}"

# Plain postgres:// URL for psql (strip SQLAlchemy driver suffix)
psql_url() {
    printf '%s' "${1/postgresql+psycopg2/postgresql}"
}

# Create a test database name
TEST_DB="leocrm_migration_test_$(date +%s)"
DB_BASE="${DB_URL_PSYNC%/*}"
TEST_URL="$DB_BASE/$TEST_DB"

# Cleanup on ANY exit — the temporary test database must never survive
TEST_DB_NAME="$TEST_DB"
PSQL_ADMIN_URL="$(psql_url "$DB_BASE/postgres")"
cleanup() {
    psql "$PSQL_ADMIN_URL" -c "DROP DATABASE IF EXISTS $TEST_DB_NAME;" 2>/dev/null || true
}
trap cleanup EXIT

echo "=== Migration Test ==="
echo "Creating test database: $TEST_DB"

# Create test database
psql "$PSQL_ADMIN_URL" -c "DROP DATABASE IF EXISTS $TEST_DB;" 2>/dev/null || true
psql "$PSQL_ADMIN_URL" -c "CREATE DATABASE $TEST_DB;"

# F21: verify the connection target BEFORE any DDL — upgrade/downgrade must
# run against the freshly created test database, never against a real one.
ACTUAL_DB=$(psql "$(psql_url "$TEST_URL")" -t -A -c "SELECT current_database();" 2>/dev/null | xargs)
if [ "$ACTUAL_DB" != "$TEST_DB" ]; then
    echo "❌ F21 target verification failed: connected to '$ACTUAL_DB', expected '$TEST_DB'"
    echo "   Refusing to run migrations — this would modify a non-test database."
    exit 1
fi
echo "✅ Target verified: migrations will run against '$ACTUAL_DB' (test database)"

echo "=== Step 1: Upgrade head on empty DB ==="
# F21: set BOTH variables — alembic/env.py prefers MIGRATION_DATABASE_URL
DATABASE_URL="$TEST_URL" MIGRATION_DATABASE_URL="$TEST_URL" alembic upgrade head 2>&1
if [ $? -ne 0 ]; then
    echo "❌ Upgrade head failed!"
    exit 1
fi

echo "✅ Upgrade head succeeded"

echo "=== Step 2: Verify tables exist ==="
TABLE_COUNT=$(psql "$(psql_url "$TEST_URL")" -t -c "SELECT count(*) FROM information_schema.tables WHERE table_schema = 'public';" 2>/dev/null | xargs)
echo "Tables created: $TABLE_COUNT"
if [ "$TABLE_COUNT" -lt 50 ]; then
    echo "❌ Too few tables ($TABLE_COUNT < 50) — migration may be incomplete"
    exit 3
fi
echo "✅ Table count OK ($TABLE_COUNT tables)"

echo "=== Step 3: Verify alembic version ==="
VERSION=$(psql "$(psql_url "$TEST_URL")" -t -c "SELECT version_num FROM alembic_version;" 2>/dev/null | xargs)
echo "Alembic version: $VERSION"
if [ -z "$VERSION" ]; then
    echo "❌ No alembic version found"
    exit 3
fi
echo "✅ Alembic version OK"

echo "=== Step 4: Downgrade base ==="
DATABASE_URL="$TEST_URL" MIGRATION_DATABASE_URL="$TEST_URL" alembic downgrade base 2>&1
if [ $? -ne 0 ]; then
    echo "⚠️  Downgrade base failed (non-critical)"
    # Don't fail the test — downgrade is not always lossless
else
    echo "✅ Downgrade base succeeded"
fi

echo "=== Step 5: Re-upgrade head (idempotency) ==="
DATABASE_URL="$TEST_URL" MIGRATION_DATABASE_URL="$TEST_URL" alembic upgrade head 2>&1
if [ $? -ne 0 ]; then
    echo "❌ Re-upgrade head failed!"
    exit 1
fi
echo "✅ Re-upgrade head succeeded"

echo "=== Cleanup ==="
cleanup
echo "✅ Test database dropped"

echo ""
echo "=== All migration tests passed ==="
exit 0