tests/conftest.py - master - outthevoid/python_microblogging_service_project

tests/conftest.py
217 строк8 KB
VladimirChukavin
Revert "Вернул проект в исходное состояние."
07 апр 2026, 14:01
07 апр 2026, 14:017531b60
import asyncio
from pathlib import Path
from typing import AsyncGenerator, Dict, Generator

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy import insert
from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.config import get_settings
from app.core.session import get_db_session
from app.database.base import Base
from app.main import app
from app.models.like_model import Like
from app.models.media_model import Media
from app.models.tweet_model import Tweet
from app.models.user_model import User, followers_table

TEST_DB_URL = "sqlite+aiosqlite:///./test_microblog.db"


@pytest_asyncio.fixture(scope="session")
async def engine() -> AsyncGenerator:
    """
    Создаёт и инициализирует тестовый асинхронный движок БД.

    Фикстура поднимает временную SQLite-базу для тестов, выполняет создание
    схемы и после завершения тестового сеанса закрывает соединение и
    удаляет файл базы данных.

    :returns: Асинхронный движок тестовой базы данных.
    :rtype: sqlalchemy.ext.asyncio.AsyncEngine
    """

    test_engine = create_async_engine(TEST_DB_URL)

    async with test_engine.begin() as connection:
        await connection.run_sync(Base.metadata.drop_all)
        await connection.run_sync(Base.metadata.create_all)
    yield test_engine
    await test_engine.dispose()
    db_path = Path("test_microblog.db")
    if db_path.exists():
        db_path.unlink()


@pytest_asyncio.fixture
async def session(engine: AsyncEngine) -> AsyncGenerator[AsyncSession, None]:
    """
    Предоставляет асинхронную сессию БД для тестов.

    Для каждого теста создаётся новая сессия, по завершении которой
    выполняется откат транзакции и очистка всех таблиц.

    :param engine: Асинхронный движок тестовой базы данных.
    :type engine: sqlalchemy.ext.asyncio.AsyncEngine

    :returns: Асинхронная сессия базы данных.
    :rtype: sqlalchemy.ext.asyncio.AsyncSession
    """

    session_maker = async_sessionmaker(
        bind=engine,
        class_=AsyncSession,
        expire_on_commit=False,
    )

    async with session_maker() as test_session:
        yield test_session
        await test_session.rollback()

    async with engine.begin() as connection:
        tasks = [
            connection.execute(table.delete())
            for table in reversed(Base.metadata.sorted_tables)
        ]
        await asyncio.gather(*tasks, return_exceptions=True)


@pytest_asyncio.fixture
async def data_session(session: AsyncSession) -> AsyncGenerator[AsyncSession, None]:
    """
    Подготавливает тестовые данные в базе.

    Создаёт тестовых пользователей, связи подписчиков, твиты, медиафайлы
    и лайки, фиксирует изменения и возвращает сессию с подготовленным
    состоянием.

    :param session: Асинхронная сессия базы данных.
    :type session: sqlalchemy.ext.asyncio.AsyncSession

    :returns: Сессия с предварительно заполненными тестовыми данными.
    :rtype: sqlalchemy.ext.asyncio.AsyncSession
    """

    alice = User(name="Alice", api_key="alice-key")
    bob = User(name="Bob", api_key="bob-key")
    charlie = User(name="Charlie", api_key="charlie-key")

    session.add_all([alice, bob, charlie])
    await session.flush()

    stmt = insert(followers_table).values(
        [
            {"follower_id": alice.id, "following_id": bob.id},
            {"follower_id": alice.id, "following_id": charlie.id},
        ]
    )
    await session.execute(stmt)

    bob_tweet = Tweet(content="Bob tweeted!", user_id=bob.id)
    charlie_tweet = Tweet(content="Charlie tweeted!", user_id=charlie.id)

    session.add_all([bob_tweet, charlie_tweet])
    await session.flush()

    session.add_all(
        [
            Media(file_path="/static/images/img_test.jpg", tweet_id=bob_tweet.id),
            Like(tweet_id=bob_tweet.id, user_id=alice.id),
            Like(tweet_id=charlie_tweet.id, user_id=alice.id),
            Like(tweet_id=charlie_tweet.id, user_id=bob.id),
        ]
    )

    await session.commit()
    yield session


@pytest_asyncio.fixture
async def client(data_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
    """
    Предоставляет асинхронного HTTP-клиента для тестирования API.

    Переопределяет зависимость получения сессии БД в приложении так,
    чтобы использовать подготовленную тестовую сессию, и создаёт клиента,
    работающего через встроенный ASGI-транспорт.

    :param data_session: Сессия с предварительно подготовленными тестовыми данными.
    :type data_session: sqlalchemy.ext.asyncio.AsyncSession

    :returns: Асинхронный HTTP-клиент для выполнения запросов к приложению.
    :rtype: httpx.AsyncClient
    """

    async def override_get_session() -> AsyncGenerator[AsyncSession, None]:
        yield data_session

    app.dependency_overrides[get_db_session] = override_get_session

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as async_client:
        yield async_client

    app.dependency_overrides.clear()


@pytest.fixture
def auth_headers() -> Dict[str, str]:
    """
    Возвращает заголовки авторизации для тестов.

    Используется для эмуляции запросов от имени тестового пользователя
    с заданным API-ключом.

    :returns: Словарь HTTP-заголовков авторизации.
    :rtype: dict[str, str]
    """

    return {"api-key": "alice-key"}


@pytest.fixture
def media_tmp_dir(tmp_path: Path) -> Generator[Path, None, None]:
    """
    Создаёт временную директорию для загрузки медиафайлов в тестах.

    Временная директория создаётся в рамках каждого теста с использованием
    предоставленного pytest пути ``tmp_path`` и подкаталога ``media``.
    Путь к этой директории подменяет значение настроек
    :attr:`app.config.Settings.media_dir`, чтобы все загружаемые в ходе
    тестов медиафайлы сохранялись в изолированное файловое окружение.

    После завершения теста настройки ``media_dir`` и ``media_url`` будут
    восстановлены к исходным значениям, а сам временный каталог и все
    его содержимое будут автоматически удалены pytest.

    :param tmp_path: Базовый временный путь для текущего теста.
    :type tmp_path: pathlib.Path

    :returns: Путь к временной директории для медиафайлов.
    :rtype: pathlib.Path
    """

    settings = get_settings()

    media_dir = tmp_path / "media"
    media_dir.mkdir(parents=True, exist_ok=True)

    old_media_dir = settings.media_dir
    old_media_url = getattr(settings, "media_url", None)

    settings.media_dir = str(media_dir)
    settings.media_url = "/static/images"

    try:
        yield media_dir
    finally:
        settings.media_dir = old_media_dir
        if old_media_url is not None:
            settings.media_url = old_media_url