leocrm/scripts/restore_test.sh

#!/bin/bash
# =============================================================================
# Restore Test — Verifies that backups can be restored successfully
# =============================================================================
# This script performs a restore test in a separate test environment.
# It should be run periodically (e.g. weekly) to verify backup integrity.
#
# Prerequisites:
#   - SSH access to Coolify server
#   - PostgreSQL backup available
#   - Storage backup available
#   - TEST_DATABASE_URL set to a test database (NOT production!)
#
# Usage:
#   bash scripts/restore_test.sh
#
# Exit codes:
#   0 = restore successful
#   1 = restore 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}[RESTORE] Running: ${name}${NC}"
    if eval "$cmd" 2>&1 | tail -10; then
        echo -e "${GREEN}[RESTORE] PASS: ${name}${NC}"
        PASS=$((PASS + 1))
    else
        echo -e "${RED}[RESTORE] FAIL: ${name}${NC}"
        FAIL=$((FAIL + 1))
    fi
}

SSH_KEY="${SSH_KEY:-/a0/usr/workdir/.ssh/coolify-01-root}"
SERVER_IP="${SERVER_IP:-46.225.91.159}"
TEST_DB_URL="${TEST_DATABASE_URL:-}"

if [ -z "$TEST_DB_URL" ]; then
    echo -e "${YELLOW}[RESTORE] SKIP: No TEST_DATABASE_URL set. Restore test requires a separate test database.${NC}"
    echo -e "${YELLOW}[RESTORE] To run: Set TEST_DATABASE_URL to a test database and execute this script.${NC}"
    exit 0
fi

echo "============================================================"
echo "  Restore Test — Backup Verification"
echo "============================================================"

# ── 1. Create test database from production backup ────────────────────────────
check "1. Restore PostgreSQL Backup" "echo 'Restore pg backup to test DB (manual step required)' && \
    ssh -o StrictHostKeyChecking=no -i $SSH_KEY root@$SERVER_IP \
    'docker exec crm-postgres pg_dump -U crm_user crm_db --no-owner --no-acl' | \
    python3 -c \"
import asyncio, os, sys
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import text

async def main():
    url = os.environ['TEST_DATABASE_URL']
    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('Test DB schema reset')
asyncio.run(main())
" && \
    ssh -o StrictHostKeyChecking=no -i $SSH_KEY root@$SERVER_IP \
    'docker exec crm-postgres pg_dump -U crm_user crm_db --no-owner --no-acl' | \
    python3 -c \"
import asyncio, os, sys
from sqlalchemy.ext.asyncio import create_async_engine

async def main():
    url = os.environ['TEST_DATABASE_URL']
    engine = create_async_engine(url)
    async with engine.begin() as conn:
        # Read SQL from stdin and execute
        sql = sys.stdin.read()
        # Split on semicolons (simplified — works for pg_dump output)
        statements = sql.split(';')
        for stmt in statements:
            stmt = stmt.strip()
            if stmt and not stmt.startswith('--'):
                try:
                    await conn.execute(text(stmt))
                except Exception:
                    pass  # Skip statements that fail (e.g. SET commands)
    await engine.dispose()
    print('PostgreSQL backup restored to test DB')
asyncio.run(main())
""

# ── 2. Run migrations on restored DB ──────────────────────────────────────────
check "2. Run Migrations on Restored DB" "DATABASE_URL=$TEST_DB_URL alembic upgrade head 2>&1 | tail -5"

# ── 3. Verify data integrity ──────────────────────────────────────────────────
check "3. Verify 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['TEST_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 table count
        result = await conn.execute(text(\"SELECT count(*) FROM pg_tables WHERE schemaname='public'\"))
        tables = result.scalar()
        print(f'Tables: {tables}')
        # Check RLS
        result = await conn.execute(text(\"SELECT count(*) FROM pg_tables WHERE schemaname='public' AND rowsecurity=true\"))
        rls = result.scalar()
        print(f'RLS tables: {rls}')
        assert rls >= 100, f'RLS count too low: {rls}'
        # Check contacts exist
        result = await conn.execute(text(\"SET app.current_tenant_id = 'bfe4d09e-e84d-4e01-ba00-8bc2aa49e5aa'; SELECT count(*) FROM contacts\"))
        contacts = result.scalar()
        print(f'Contacts (real tenant): {contacts}')
        assert contacts > 0, 'No contacts found in restored DB'
    await engine.dispose()
    print('Data integrity verified')
asyncio.run(main())
""

# ── 4. RLS re-test on restored DB ─────────────────────────────────────────────
check "4. RLS Re-test" "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['TEST_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, 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('RLS re-test passed: 0 rows without/fake tenant')
asyncio.run(main())
""

# ── Summary ──────────────────────────────────────────────────────────────────
echo ""
echo "============================================================"
echo "  Restore Test: ${PASS} passed, ${FAIL} failed"
echo "============================================================"

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