tests/conftest.py - master - tsps/core

tests/conftest.py
180 строк6 KB
Василий Петров
Загрузка аудио
26 мар 2026, 12:11
26 мар 2026, 12:11ea6ac40
import os
import io
from pathlib import Path
# Устанавливаем переменную окружения, чтобы загрузить тестовые настройки
os.environ["ENV_FILE"] = str(Path(__file__).parent / ".env.test")

# Сбрасываем кэш модулей app
import sys
for mod in list(sys.modules):
    if mod.startswith("app."):
        del sys.modules[mod]

# Импортируем модули для тестов после сброса кэша и переопределения env
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import text
from app.main import app
from app.models import Judge, Admin, JudgeAccount
from app.core.security import create_token, _SESSIONS
from app.database import Base, engine
from app.database import SessionLocal

@pytest.fixture(scope="session", autouse=True)
def setup_test_db():
    """Создаём все таблицы перед тестами."""
    Base.metadata.drop_all(bind=engine)  # очищаем (на случай повторного запуска)
    Base.metadata.create_all(bind=engine)
    yield

@pytest.fixture
def client():
    """Тестовый клиент FastAPI."""
    return TestClient(app)

@pytest.fixture
def db_session():
    """Сессия БД для подготовки данных."""
    session = SessionLocal()
    yield session
    session.rollback()
    session.close()


@pytest.fixture
def auth_headers():
    db = SessionLocal()
    email = "test_admin@example.com"
    # Удаляем старого админа, если есть
    db.execute(text("DELETE FROM admin WHERE email = :email"), {"email": email})
    db.commit()

    # Создаём админа
    admin = Admin(email=email, first_name="Тест", last_name="Админ")
    admin.set_password("any")  # пароль не важен — токен создаём напрямую
    db.add(admin)
    db.commit()
    db.refresh(admin)
    admin_id = admin.id
    db.close()

    # Создаём токен напрямую
    token = create_token(admin_id, owner_type="admin")
    headers = {"Authorization": f"Bearer {token}"}

    yield headers

    # Очистка
    db = SessionLocal()
    db.execute(text("DELETE FROM admin WHERE email = :email"), {"email": email})
    db.commit()
    db.close()
    if token in _SESSIONS:
        del _SESSIONS[token]

@pytest.fixture
def auditor_auth_headers():
    """
    Создаёт админа с ролью 'auditor' и возвращает заголовки с валидным токеном.
    Используется для проверки запрета деструктивных операций.
    """
    db = SessionLocal()
    email = "test_auditor@example.com"

    # Удаляем старого, если есть
    db.execute(text("DELETE FROM admin WHERE email = :email"), {"email": email})
    db.commit()

    # Создаём админа с ролью auditor
    admin = Admin(email=email, first_name="Тест", last_name="Аудитор", role="auditor")
    admin.set_password("any")
    db.add(admin)
    db.commit()
    db.refresh(admin)
    admin_id = admin.id
    db.close()

    # Создаём токен напрямую
    token = create_token(admin_id, owner_type="admin")
    headers = {"Authorization": f"Bearer {token}"}

    yield headers

    # Очистка
    db = SessionLocal()
    db.execute(text("DELETE FROM admin WHERE email = :email"), {"email": email})
    db.commit()
    db.close()
    if token in _SESSIONS:
        del _SESSIONS[token]

@pytest.fixture
def lk_auth_headers():
    """
    Создаёт тестового судью и аккаунт, возвращает заголовки с токеном для /api/lk.
    """
    db = SessionLocal()
    try:
        email = "test_judge@example.com"

        # Очищаем данные (сначала аккаунты, потом судей, чтобы избежать FK ошибок, 
        # хотя TRUNCATE CASCADE в utils это решает, здесь делаем явно для чистоты)
        db.execute(text("DELETE FROM document WHERE judge_id IN (SELECT id FROM judge WHERE last_name = 'Тест' AND first_name = 'Судья')"))
        db.execute(text("DELETE FROM judge_account WHERE email = :email"), {"email": email})
        db.execute(text("DELETE FROM judge WHERE last_name = 'Тест' AND first_name = 'Судья'"))
        db.commit()

        judge = Judge(
            last_name="Тест",
            first_name="Судья",
            middle_name="Тестович",
            auto_load_enabled=False
        )
        db.add(judge)
        db.commit()
        db.refresh(judge)

        account = JudgeAccount(
            judge_id=judge.id,
            email=email,
            is_active=True
        )
        account.set_password("any")  # Пароль не важен, токен создаём напрямую
        db.add(account)
        db.commit()
        db.refresh(account)
        
        account_id = account.id
    finally:
        db.close()

    # Создаём токен напрямую (привязываем к account.id)
    token = create_token(account_id, owner_type="judge_account")
    headers = {"Authorization": f"Bearer {token}"}

    yield headers

    # Очистка
    db = SessionLocal()
    try:
        # Снова находим судью для очистки
        judge = db.query(Judge).filter(Judge.last_name == "Тест", Judge.first_name == "Судья").first()
        
        if judge:
            db.execute(text("DELETE FROM audio_recording WHERE judge_id = :judge_id"), {"judge_id": judge.id})
            db.execute(text("DELETE FROM document WHERE judge_id = :judge_id"), {"judge_id": judge.id})
            db.execute(text("DELETE FROM judge_account WHERE judge_id = :judge_id"), {"judge_id": judge.id})
            db.delete(judge)
            db.commit()
    finally:
        db.close()

    # Инвалидируем токен в памяти
    if token in _SESSIONS:
        del _SESSIONS[token]

@pytest.fixture
def audio_file():
    """Создаёт тестовый аудиофайл в памяти."""
    content = b"fake audio content" * 100  # ~1.6 KB
    return io.BytesIO(content), "test_recording.mp3", "audio/mpeg"