Imago/tests/conftest.py

"""
Fixtures pytest — base de données de test + clients API.
"""
import secrets
import asyncio
from typing import AsyncGenerator

import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

from app.database import Base, get_db
from app.dependencies.auth import hash_api_key
from app.models.client import APIClient, ClientPlan
from app.main import app


# ─────────────────────────────────────────────────────────────
# Engine de test (SQLite in-memory)
# ─────────────────────────────────────────────────────────────

TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

test_engine = create_async_engine(TEST_DATABASE_URL, echo=False)
TestSessionLocal = async_sessionmaker(
    bind=test_engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autoflush=False,
    autocommit=False,
)


# ─────────────────────────────────────────────────────────────
# Override de la dépendance get_db
# ─────────────────────────────────────────────────────────────

async def override_get_db() -> AsyncGenerator[AsyncSession, None]:
    async with TestSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()


app.dependency_overrides[get_db] = override_get_db


# ─────────────────────────────────────────────────────────────
# Mock ARQ pool et Redis pour les tests (pas de Redis requis)
# ─────────────────────────────────────────────────────────────

from unittest.mock import AsyncMock

_mock_arq_pool = AsyncMock()
_mock_arq_pool.enqueue_job = AsyncMock(return_value=None)
_mock_arq_pool.close = AsyncMock()

_mock_redis = AsyncMock()
_mock_redis.ping = AsyncMock(return_value=True)
_mock_redis.publish = AsyncMock(return_value=0)

app.state.arq_pool = _mock_arq_pool
app.state.redis = _mock_redis


# ─────────────────────────────────────────────────────────────
# Fixtures
# ─────────────────────────────────────────────────────────────

@pytest_asyncio.fixture(autouse=True)
async def setup_database():
    """Crée et détruit toutes les tables avant/après chaque test."""
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)


@pytest_asyncio.fixture
async def db_session() -> AsyncGenerator[AsyncSession, None]:
    """Session DB de test."""
    async with TestSessionLocal() as session:
        yield session


@pytest_asyncio.fixture
async def client_a_key() -> str:
    """Clé API en clair pour le client A."""
    return "test-key-client-a-secret-123456"


@pytest_asyncio.fixture
async def client_b_key() -> str:
    """Clé API en clair pour le client B."""
    return "test-key-client-b-secret-789012"


@pytest_asyncio.fixture
async def admin_key() -> str:
    """Clé API en clair pour le client admin."""
    return "test-key-admin-secret-000000"


@pytest_asyncio.fixture
async def client_a(db_session: AsyncSession, client_a_key: str) -> APIClient:
    """Client A — droits standard (images + AI)."""
    client = APIClient(
        name="Test Client A",
        api_key_hash=hash_api_key(client_a_key),
        scopes=["images:read", "images:write", "images:delete", "ai:use"],
        plan=ClientPlan.STANDARD,
    )
    db_session.add(client)
    await db_session.commit()
    await db_session.refresh(client)
    return client


@pytest_asyncio.fixture
async def client_b(db_session: AsyncSession, client_b_key: str) -> APIClient:
    """Client B — droits standard (images + AI)."""
    client = APIClient(
        name="Test Client B",
        api_key_hash=hash_api_key(client_b_key),
        scopes=["images:read", "images:write", "images:delete", "ai:use"],
        plan=ClientPlan.FREE,
    )
    db_session.add(client)
    await db_session.commit()
    await db_session.refresh(client)
    return client


@pytest_asyncio.fixture
async def admin_client(db_session: AsyncSession, admin_key: str) -> APIClient:
    """Client admin — tous les droits."""
    client = APIClient(
        name="Admin Client",
        api_key_hash=hash_api_key(admin_key),
        scopes=["images:read", "images:write", "images:delete", "ai:use", "admin"],
        plan=ClientPlan.PREMIUM,
    )
    db_session.add(client)
    await db_session.commit()
    await db_session.refresh(client)
    return client


@pytest_asyncio.fixture
async def auth_headers_a(client_a_key: str) -> dict[str, str]:
    """Headers d'authentification pour le client A."""
    return {"Authorization": f"Bearer {client_a_key}"}


@pytest_asyncio.fixture
async def auth_headers_b(client_b_key: str) -> dict[str, str]:
    """Headers d'authentification pour le client B."""
    return {"Authorization": f"Bearer {client_b_key}"}


@pytest_asyncio.fixture
async def admin_headers(admin_key: str) -> dict[str, str]:
    """Headers d'authentification pour le client admin."""
    return {"Authorization": f"Bearer {admin_key}"}


@pytest_asyncio.fixture
async def async_client() -> AsyncGenerator[AsyncClient, None]:
    """Client HTTP async pour tester l'API FastAPI."""
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac