PYTHON test_migrations.py

#!/usr/bin/env python3
"""AgentForms Migration Tests (Phase 5d)

Tests for Alembic migration workflow:
1. Alembic upgrade head succeeds on empty DB
2. Alembic downgrade base succeeds
3. Alembic check passes (no diff between models and migrations)
"""
import os
import shutil
import subprocess
import sys
import tempfile

import pytest

# Ensure project root is on path
ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT_DIR)


@pytest.fixture(scope="module")
def temp_db_dir():
    """Create a temporary directory with a clean DB for migration tests."""
    tmp_dir = tempfile.mkdtemp(prefix="af_migrate_test_")
    db_path = os.path.join(tmp_dir, "relay.db")

    env = os.environ.copy()
    env["RELAY_DB_PATH"] = db_path

    yield tmp_dir, db_path, env

    # Cleanup
    shutil.rmtree(tmp_dir, ignore_errors=True)


def test_alembic_upgrade_head(temp_db_dir):
    """Test: alembic upgrade head succeeds on empty DB."""
    tmp_dir, db_path, env = temp_db_dir

    result = subprocess.run(
        [sys.executable, "-m", "alembic", "upgrade", "head"],
        cwd=ROOT_DIR,
        env=env,
        capture_output=True,
        text=True,
        timeout=30,
    )

    assert result.returncode == 0, (
        f"alembic upgrade head failed:\nstdout: {result.stdout}\nstderr: {result.stderr}"
    )
    # Verify the DB file was created
    assert os.path.exists(db_path), f"DB file not created at {db_path}"


def test_alembic_current(temp_db_dir):
    """Test: alembic current shows the latest revision."""
    tmp_dir, db_path, env = temp_db_dir

    # First upgrade
    subprocess.run(
        [sys.executable, "-m", "alembic", "upgrade", "head"],
        cwd=ROOT_DIR,
        env=env,
        capture_output=True,
        text=True,
        timeout=30,
    )

    result = subprocess.run(
        [sys.executable, "-m", "alembic", "current"],
        cwd=ROOT_DIR,
        env=env,
        capture_output=True,
        text=True,
        timeout=30,
    )

    assert result.returncode == 0, (
        f"alembic current failed:\nstdout: {result.stdout}\nstderr: {result.stderr}"
    )
    assert "Base" not in result.stdout or "head" in result.stdout.lower(), (
        f"Expected current revision to be at head, got: {result.stdout}"
    )


def test_alembic_downgrade_base(temp_db_dir):
    """Test: alembic downgrade base succeeds (reverts all migrations)."""
    tmp_dir, db_path, env = temp_db_dir

    # First upgrade to head
    subprocess.run(
        [sys.executable, "-m", "alembic", "upgrade", "head"],
        cwd=ROOT_DIR,
        env=env,
        capture_output=True,
        text=True,
        timeout=30,
    )

    result = subprocess.run(
        [sys.executable, "-m", "alembic", "downgrade", "base"],
        cwd=ROOT_DIR,
        env=env,
        capture_output=True,
        text=True,
        timeout=30,
    )

    assert result.returncode == 0, (
        f"alembic downgrade base failed:\nstdout: {result.stdout}\nstderr: {result.stderr}"
    )


def test_alembic_history(temp_db_dir):
    """Test: alembic history lists all migrations."""
    result = subprocess.run(
        [sys.executable, "-m", "alembic", "history"],
        cwd=ROOT_DIR,
        env=os.environ.copy(),
        capture_output=True,
        text=True,
        timeout=30,
    )

    assert result.returncode == 0, (
        f"alembic history failed:\nstdout: {result.stdout}\nstderr: {result.stderr}"
    )
    assert "baseline" in result.stdout.lower(), (
        f"Expected baseline migration in history, got: {result.stdout}"
    )


def test_migration_round_trip(temp_db_dir):
    """Test: upgrade -> downgrade -> upgrade succeeds (full round trip)."""
    tmp_dir, db_path, env = temp_db_dir

    steps = [
        (["upgrade", "head"], "upgrade to head"),
        (["downgrade", "base"], "downgrade to base"),
        (["upgrade", "head"], "upgrade again"),
    ]

    for args, desc in steps:
        result = subprocess.run(
            [sys.executable, "-m", "alembic"] + args,
            cwd=ROOT_DIR,
            env=env,
            capture_output=True,
            text=True,
            timeout=30,
        )
        assert result.returncode == 0, (
            f"Round trip step '{desc}' failed:\n"
            f"stdout: {result.stdout}\nstderr: {result.stderr}"
        )


if __name__ == "__main__":
    pytest.main([__file__, "-v"])