tests/e2e/conftest.py

"""
E2E Test Configuration

Fixtures for running real server integration tests with authentication.
"""
import os
import pytest
import httpx
import subprocess
import time
import socket
import sqlite3
import uuid
from contextlib import closing


# E2E test configuration
E2E_TEST_PORT = 8001
E2E_TEST_HOST = "127.0.0.1"
E2E_BASE_URL = f"http://{E2E_TEST_HOST}:{E2E_TEST_PORT}"
E2E_DB_FILE = "apigateway_production.db"  # Server uses this in development mode
# Use a 32+ char secret to pass library validation
JWT_SECRET = "e2e-test-jwt-secret-key-32-chars-minimum"


def find_free_port():
    """Find an available port."""
    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
        s.bind(('', 0))
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        return s.getsockname()[1]


def wait_for_server(url: str, timeout: int = 30) -> bool:
    """Wait for server to be ready."""
    start = time.time()
    while time.time() - start < timeout:
        try:
            response = httpx.get(f"{url}/health", timeout=2)
            if response.status_code == 200:
                return True
        except httpx.RequestError:
            pass
        time.sleep(0.5)
    return False


@pytest.fixture(scope="session")
def e2e_env():
    """Set up E2E test environment variables."""
    original_env = os.environ.copy()
    
    # Test environment configuration
    os.environ["CORS_ORIGINS"] = "http://localhost:3000"
    os.environ["JWT_SECRET"] = JWT_SECRET
    os.environ["AUTH_SIGN_IN_GOOGLE_CLIENT_ID"] = "test-client-id"
    os.environ["ENVIRONMENT"] = "development"
    os.environ["RESET_DB"] = "false"  # Handle manually to avoid race conditions
    os.environ["SKIP_SERVICE_REGISTRATION_CHECK"] = "true"
    
    # Explicitly set DB URL to absolute path to avoid any ambiguity
    db_path = "/home/jebin/git/apigateway/apigateway_production.db"
    os.environ["DATABASE_URL"] = f"sqlite+aiosqlite:///{db_path}"
    
    yield
    
    # Restore original environment
    os.environ.clear()
    os.environ.update(original_env)


@pytest.fixture(scope="session")
def live_server(e2e_env):
    """Start a real uvicorn server for E2E tests."""
    # Handle DB clean up manually BEFORE server starts
    db_path = "/home/jebin/git/apigateway/apigateway_production.db"
    
    # Force delete existing DB
    if os.path.exists(db_path):
        os.remove(db_path)
        
    # We need to initialize the DB structure since we just deleted it
    # We can use python -c to run the init_db code from app context
    # This ensures tables exists BEFORE we start the server and try to insert users
    # We'll use a simple script to create tables
    init_script = """
import asyncio
from core.database import init_db
async def main():
    await init_db()
if __name__ == "__main__":
    asyncio.run(main())
"""
    subprocess.run(
        ["python", "-c", init_script],
        cwd="/home/jebin/git/apigateway",
        env=os.environ.copy(),
        check=True
    )
    
    port = find_free_port()
    base_url = f"http://127.0.0.1:{port}"
    
    # Start uvicorn in subprocess
    process = subprocess.Popen(
        [
            "python", "-m", "uvicorn",
            "app:app",
            "--host", "127.0.0.1",
            "--port", str(port),
            "--log-level", "warning"
        ],
        cwd="/home/jebin/git/apigateway",
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        env=os.environ.copy()
    )
    
    # Wait for server to be ready
    if not wait_for_server(base_url):
        process.terminate()
        stdout, stderr = process.communicate(timeout=5)
        raise RuntimeError(f"Server failed to start. stderr: {stderr.decode()}")
    
    yield base_url
    
    # Cleanup
    process.terminate()
    try:
        process.wait(timeout=5)
    except subprocess.TimeoutExpired:
        process.kill()


@pytest.fixture
def api_client(live_server):
    """HTTP client for making real API requests."""
    with httpx.Client(base_url=live_server, timeout=30.0) as client:
        yield client


@pytest.fixture
def test_user_data():
    """Generate unique test user data."""
    unique_id = str(uuid.uuid4())[:8]
    return {
        "user_id": f"e2e_user_{unique_id}",
        "email": f"e2e_test_{unique_id}@example.com",
        "google_id": f"google_{unique_id}",
        "name": f"E2E Test User {unique_id}",
        "credits": 100,
        "token_version": 1
    }


@pytest.fixture
def create_test_user(live_server, test_user_data):
    """
    Create a test user directly in the database.
    Returns user data and access token.
    """
    
    # Connect to the database the server is using
    db_path = "/home/jebin/git/apigateway/apigateway_production.db"
    
    # Wait for DB and tables to be ready (server creates them on startup)
    max_retries = 20
    for attempt in range(max_retries):
        try:
            conn = sqlite3.connect(db_path)
            cursor = conn.cursor()
            
            # Check if users table exists
            cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='users'")
            if cursor.fetchone() is None:
                conn.close()
                time.sleep(0.5)
                continue
            
            # Insert test user using the app's own codebase to ensure compatibility
            insert_script = f"""
import asyncio
import os
from sqlalchemy import select
from core.database import async_session_maker
from core.models import User
import datetime

async def create_user():
    async with async_session_maker() as db:
        # Check if user exists
        stmt = select(User).where(User.user_id == '{test_user_data["user_id"]}')
        result = await db.execute(stmt)
        if result.scalar_one_or_none():
            return

        user = User(
            user_id='{test_user_data["user_id"]}',
            email='{test_user_data["email"]}',
            google_id='{test_user_data["google_id"]}',
            name='{test_user_data["name"]}',
            credits={test_user_data["credits"]},
            token_version={test_user_data["token_version"]},
            is_active=True,
            created_at=datetime.datetime.utcnow(),
            updated_at=datetime.datetime.utcnow()
        )
        db.add(user)
        await db.commit()

if __name__ == "__main__":
    asyncio.run(create_user())
"""
            # Run instructions in subprocess
            subprocess.run(
                ["python", "-c", insert_script],
                cwd="/home/jebin/git/apigateway",
                env=os.environ.copy(),
                check=True,
                capture_output=True
            )
            
            # Verify via sqlite3 just in case
            conn = sqlite3.connect(db_path)
            cursor = conn.cursor()
            cursor.execute("SELECT user_id FROM users WHERE user_id=?", (test_user_data["user_id"],))
            if cursor.fetchone():
                conn.close()
                break
            conn.close()
            
        except (sqlite3.OperationalError, subprocess.CalledProcessError) as e:
            if attempt < max_retries - 1:
                time.sleep(0.5)
            else:
                raise RuntimeError(f"Failed to create test user after {max_retries} attempts: {e}")
    
    # Generate a valid JWT token using the library with SAME secret as server
    from google_auth_service import JWTService
    jwt_service = JWTService(secret_key=JWT_SECRET)
    access_token = jwt_service.create_access_token(
        user_id=test_user_data["user_id"],
        email=test_user_data["email"],
        token_version=test_user_data["token_version"]
    )
    
    return {
        **test_user_data,
        "access_token": access_token
    }


@pytest.fixture
def authenticated_client(api_client, create_test_user):
    """
    HTTP client with valid authentication token.
    Uses a real user created in the database.
    """
    token = create_test_user["access_token"]
    api_client.headers["Authorization"] = f"Bearer {token}"
    
    yield api_client, create_test_user
    
    # Cleanup - remove auth header
    if "Authorization" in api_client.headers:
        del api_client.headers["Authorization"]