leocrm/scripts/migration_release_gate.sh

#!/bin/bash
# =============================================================================
# Migrations Release Gate — Pre-Release Verification
# =============================================================================
# Runs before any release that includes migration changes.
# Verifies both installation paths produce the same schema.
#
# Prerequisites:
#   - Docker available
#   - PostgreSQL accessible
#   - DATABASE_URL set to a test database (NOT production!)
#
# Usage:
#   bash scripts/migration_release_gate.sh
#
# Exit codes:
#   0 = all checks passed
#   1 = one or more checks failed
# =============================================================================

set -e

RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m'

PASS=0
FAIL=0

check() {
    local name="$1"
    local cmd="$2"
    echo -e "${YELLOW}[GATE] Running: ${name}${NC}"
    if eval "$cmd" 2>&1 | tail -10; then
        echo -e "${GREEN}[GATE] PASS: ${name}${NC}"
        PASS=$((PASS + 1))
    else
        echo -e "${RED}[GATE] FAIL: ${name}${NC}"
        FAIL=$((FAIL + 1))
    fi
}

if [ -z "${DATABASE_URL:-}" ]; then
    echo -e "${RED}[GATE] ERROR: DATABASE_URL must be set to a TEST database (not production!)${NC}"
    exit 1
fi

# ── 1. Fresh Install: Empty DB → Alembic Head → Plugin Migrations ─────────────
check "Fresh Install (empty DB)" "python3 -c \"
import asyncio, os
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import text

async def main():
    url = os.environ['DATABASE_URL']
    # Drop all tables for fresh install
    engine = create_async_engine(url)
    async with engine.begin() as conn:
        await conn.execute(text('DROP SCHEMA IF EXISTS public CASCADE'))
        await conn.execute(text('CREATE SCHEMA public'))
    await engine.dispose()
    print('Fresh DB created (schema dropped and recreated)')

asyncio.run(main())
" && alembic upgrade head && python3 -c \"
import asyncio, os
from app.core.db import async_session_maker
from app.core.bootstrap import bootstrap_roles
async def main():
    async with async_session_maker() as db:
        await bootstrap_roles(db)
    print('Roles bootstrapped')
asyncio.run(main())
""

# ── 2. Schema Snapshot (fresh install) ────────────────────────────────────────
check "Schema Snapshot (fresh)" "python3 -c \"
import asyncio, os, json
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import text

async def main():
    engine = create_async_engine(os.environ['DATABASE_URL'])
    async with engine.connect() as conn:
        # Tables
        result = await conn.execute(text(\"SELECT tablename FROM pg_tables WHERE schemaname='public' ORDER BY tablename\"))
        tables = sorted([r[0] for r in result])
        # RLS
        result = await conn.execute(text(\"SELECT tablename FROM pg_tables WHERE schemaname='public' AND rowsecurity=true ORDER BY tablename\"))
        rls = sorted([r[0] for r in result])
        # Indexes
        result = await conn.execute(text(\"SELECT indexname FROM pg_indexes WHERE schemaname='public' ORDER BY indexname\"))
        indexes = sorted([r[0] for r in result])
    await engine.dispose()
    snapshot = {'tables': tables, 'rls': rls, 'indexes': indexes}
    with open('/tmp/schema_fresh.json', 'w') as f:
        json.dump(snapshot, f, indent=2)
    print(f'Fresh: {len(tables)} tables, {len(rls)} RLS, {len(indexes)} indexes')
asyncio.run(main())
""

# ── 3. RLS and Grants Check ───────────────────────────────────────────────────
check "RLS and Grants" "python3 -c \"
import asyncio, os
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import text

async def main():
    engine = create_async_engine(os.environ['DATABASE_URL'])
    async with engine.connect() as conn:
        # Check RLS enabled on critical tables
        result = await conn.execute(text(\"SELECT count(*) FROM pg_tables WHERE schemaname='public' AND rowsecurity=true\"))
        rls_count = result.scalar()
        assert rls_count >= 100, f'RLS count too low: {rls_count}'
        # Check roles exist
        result = await conn.execute(text(\"SELECT count(*) FROM pg_roles WHERE rolname IN ('crm_api','crm_auth','crm_worker','crm_migration')\"))
        roles_count = result.scalar()
        assert roles_count == 4, f'Expected 4 roles, got {roles_count}'
        # Check crm_api has no BYPASSRLS
        result = await conn.execute(text(\"SELECT rolbypassrls FROM pg_roles WHERE rolname='crm_api'\"))
        bypass = result.scalar()
        assert not bypass, 'crm_api must not have BYPASSRLS'
    await engine.dispose()
    print(f'RLS: {rls_count} tables, 4 roles, no BYPASSRLS on crm_api')
asyncio.run(main())
""

# ── 4. Cross-Tenant Read/Write Test ───────────────────────────────────────────
check "Cross-Tenant Read/Write" "python3 -c \"
import asyncio, os
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import text

async def main():
    engine = create_async_engine(os.environ['DATABASE_URL'])
    async with engine.connect() as conn:
        # No tenant context → 0 rows
        await conn.execute(text('RESET app.current_tenant_id'))
        result = await conn.execute(text('SELECT count(*) FROM contacts'))
        count = result.scalar()
        assert count == 0, f'Expected 0 rows without tenant context, got {count}'
        # Fake tenant → 0 rows
        await conn.execute(text(\"SET app.current_tenant_id = '00000000-0000-0000-0000-000000000000'\"))
        result = await conn.execute(text('SELECT count(*) FROM contacts'))
        count = result.scalar()
        assert count == 0, f'Expected 0 rows with fake tenant, got {count}'
    await engine.dispose()
    print('Cross-Tenant: 0 rows without/fake tenant context')
asyncio.run(main())
""

# ── 5. Data Integrity Check ───────────────────────────────────────────────────
check "Data Integrity" "python3 -c \"
import asyncio, os
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import text

async def main():
    engine = create_async_engine(os.environ['DATABASE_URL'])
    async with engine.connect() as conn:
        # Check alembic version
        result = await conn.execute(text('SELECT version_num FROM alembic_version'))
        version = result.scalar()
        print(f'Alembic version: {version}')
        # Check no orphaned FKs
        result = await conn.execute(text(\"SELECT count(*) FROM pg_constraint WHERE contype='f' AND connamespace='public'::regnamespace\"))
        fk_count = result.scalar()
        print(f'Foreign keys: {fk_count}')
    await engine.dispose()
asyncio.run(main())
""

# ── Summary ──────────────────────────────────────────────────────────────────
echo ""
echo "============================================================"
echo "  Migration Release Gate: ${PASS} passed, ${FAIL} failed"
echo "============================================================"

if [ $FAIL -gt 0 ]; then
    echo -e "${RED}[GATE] FAILED — ${FAIL} checks failed${NC}"
    exit 1
else
    echo -e "${GREEN}[GATE] PASSED — all ${PASS} checks passed${NC}"
    exit 0
fi