blog.pyaqa.ru/tests/api/conftest.py

"""API test configuration.

This module provides fixtures for testing API endpoints with a real database
and mock authentication client in development mode.
"""

import asyncio
import os
import uuid
from collections.abc import Generator
from typing import Any, cast

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine

from alembic import command
from alembic.config import Config
from app.infrastructure.config import settings
from app.infrastructure.database.models import Base
from app.main import app_factory

API_PREFIX = "/api/v1/posts"

USER_TOKEN = "dev-token-user"
USER2_TOKEN = "dev-token-user2"
ADMIN_TOKEN = "dev-token-admin"
GUEST_TOKEN = "dev-token-guest"

USER_ID = "dev-user"
USER2_ID = "dev-user2"
ADMIN_ID = "dev-admin"


def _sync_url(db_url: str) -> str:
    """Strip async driver suffix for sync engine.

    Args:
        db_url: Database URL with async driver.

    Returns:
        Database URL without async driver suffix.
    """
    return db_url.replace("+aiosqlite", "").replace("+asyncpg", "")


def _build_alembic_config(db_url: str) -> Config:
    """Build alembic config with given database URL.

    Args:
        db_url: Database URL to use.

    Returns:
        Alembic Config instance.
    """
    alembic_cfg = Config("alembic.ini")
    alembic_cfg.set_main_option("sqlalchemy.url", db_url)
    return alembic_cfg


@pytest.fixture(scope="session")
def event_loop() -> Generator[asyncio.AbstractEventLoop]:
    """Create event loop for the test session.

    Yields:
        Event loop instance.
    """
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(scope="session", autouse=True)
def setup_database() -> Generator[None]:
    """Set up database tables once per test session.

    Drops and recreates all tables, then stamps alembic head
    so that alembic-aware operations (like E2E tests) can proceed.
    Uses sync SQLAlchemy engine to avoid async complications.
    """
    db_url = os.environ.get("DB_URL", settings.database_url)
    sync_engine = create_engine(_sync_url(db_url))
    Base.metadata.drop_all(sync_engine)
    Base.metadata.create_all(sync_engine)
    sync_engine.dispose()

    alembic_cfg = _build_alembic_config(db_url)
    command.stamp(alembic_cfg, "head")
    yield


@pytest.fixture(scope="session")
def app() -> TestClient:
    """Create FastAPI test application.

    The app uses MockKeycloakClient in dev mode, so auth tokens
    like 'dev-token-user', 'dev-token-admin', etc. are recognized.
    Lifespan is not entered (no init_db/close_db) — the
    setup_database fixture handles table creation instead.

    Returns:
        Configured TestClient instance.
    """
    return TestClient(app_factory())


@pytest.fixture
def client(app: TestClient) -> TestClient:
    """Provide TestClient for API requests.

    The TestClient is created without entering the app lifespan
    to avoid disposing the module-level database engine. Database
    lifecycle is managed by setup_database fixture.

    Args:
        app: Test application fixture.

    Returns:
        TestClient instance.
    """
    return app


@pytest.fixture
def user_headers() -> dict[str, str]:
    """Headers for standard user authentication.

    User has role 'user', user_id 'dev-user'.

    Returns:
        Headers dict with Authorization bearer token.
    """
    return {"Authorization": f"Bearer {USER_TOKEN}"}


@pytest.fixture
def user2_headers() -> dict[str, str]:
    """Headers for second user authentication.

    User2 has role 'user', user_id 'dev-user2'. Used for ownership
    policy tests (editing/deleting another user's post).

    Returns:
        Headers dict with Authorization bearer token.
    """
    return {"Authorization": f"Bearer {USER2_TOKEN}"}


@pytest.fixture
def admin_headers() -> dict[str, str]:
    """Headers for admin authentication.

    Admin has role 'admin', user_id 'dev-admin'. Can edit/delete
    any post regardless of ownership.

    Returns:
        Headers dict with Authorization bearer token.
    """
    return {"Authorization": f"Bearer {ADMIN_TOKEN}"}


@pytest.fixture
def guest_headers() -> dict[str, str]:
    """Headers for guest (inactive token).

    Guest token returns inactive TokenInfo, which the API treats
    as unauthenticated.

    Returns:
        Headers dict with Authorization bearer token.
    """
    return {"Authorization": f"Bearer {GUEST_TOKEN}"}


@pytest.fixture
def post_payload() -> dict[str, Any]:
    """Generate a unique post payload for testing.

    Uses uuid4 for title to ensure unique slug generation
    across tests. Prevents slug uniqueness constraint errors.

    Returns:
        Dict with title, content, and tags fields.
    """
    unique_id = uuid.uuid4().hex[:8]
    return {
        "title": f"Test Post {unique_id}",
        "content": f"This is the content of test post {unique_id}. It has enough length to pass validation.",
        "tags": ["test", f"tag-{unique_id}"],
    }


@pytest.fixture
def created_post(
    client: TestClient,
    user_headers: dict[str, str],
    post_payload: dict[str, Any],
) -> dict[str, Any]:
    """Create a post via API and return its response data.

    Used as test data for read, update, delete, and publish tests.
    Created by the standard user ('dev-user').

    Args:
        client: TestClient fixture.
        user_headers: User auth headers.
        post_payload: Post creation payload.

    Returns:
        Post response data dict.
    """
    response = client.post(
        f"{API_PREFIX}",
        json=post_payload,
        headers=user_headers,
    )
    assert response.status_code == 201, f"Failed to create test post: {response.text}"
    return cast(dict[str, Any], response.json())