tests/api/admin/test_audio_recordings.py - master - tsps/core

tests/api/admin/test_audio_recordings.py
720 строк26 KB
Василий Петров
Апи саммаризации
03 май 2026, 10:36
03 май 2026, 10:36836c998
import pytest
from datetime import datetime
from tests.utils import clean_table
from app.models import AudioRecording, CaseEntity, Judge, Court
from fastapi.testclient import TestClient
from app.main import app
from app.database import get_db, SessionLocal
from unittest.mock import patch, MagicMock

client = TestClient(app)


@pytest.fixture(autouse=True)
def clean_tables():
    """Очищает таблицы перед каждым тестом."""
    clean_table(AudioRecording)
    clean_table(CaseEntity)
    clean_table(Judge)
    clean_table(Court)


@pytest.fixture
def setup_basic_data():
    """Создаёт базовые данные напрямую в БД."""
    try:
        db = SessionLocal()

        court = Court(name="Тестовый суд")
        db.add(court)
        db.flush()

        judge = Judge(last_name="Иванов", first_name="Петр", middle_name="Сергеевич", current_court_id=court.id)
        db.add(judge)
        db.flush()

        case_entity = CaseEntity(number="1-234/2024")
        db.add(case_entity)
        db.flush()

        db.commit()

        return {
            "court_id": court.id,
            "judge_id": judge.id,
            "case_entity_id": case_entity.id
        }
    finally:
        db.close()


def test_read_audio_recordings_empty(auth_headers):
    """Список аудиозаписей пуст — возвращаем пустой список."""
    response = client.get("/api/admin/audio-recordings", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 0
    assert data["total"] == 0


def test_read_audio_recordings_with_data(auth_headers, setup_basic_data):
    """Список аудиозаписей с данными — проверка полей и связанных объектов."""
    db = next(get_db())
    try:
        recording = AudioRecording(
            file_name="recording_001.mp3",
            s3_key="audio/2024/01/recording_001.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="completed",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"],
            transcription="Текст расшифровки"
        )
        db.add(recording)
        db.commit()
    finally:
        db.close()

    response = client.get("/api/admin/audio-recordings", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 1
    assert data["items"][0]["file_name"] == "recording_001.mp3"
    assert data["items"][0]["duration_seconds"] == 3600
    assert data["items"][0]["file_size"] == 1024000
    assert data["items"][0]["transcription_status"] == "completed"
    assert data["items"][0]["judge"]["last_name"] == "Иванов"
    assert data["items"][0]["court"]["name"] == "Тестовый суд"
    assert data["items"][0]["case_entity"]["number"] == "1-234/2024"


def test_read_audio_recordings_pagination(auth_headers, setup_basic_data):
    """Пагинация списка аудиозаписей."""
    db = next(get_db())
    try:
        for i in range(5):
            db.add(AudioRecording(
                file_name=f"recording_{i+1}.mp3",
                s3_key=f"audio/2024/01/recording_{i+1}.mp3",
                mime_type="audio/mpeg",
                file_size=1024000,
                duration_seconds=3600,
                transcription_status="created",
                case_entity_id=setup_basic_data["case_entity_id"],
                judge_id=setup_basic_data["judge_id"],
                court_id=setup_basic_data["court_id"]
            ))
        db.commit()
    finally:
        db.close()

    response = client.get("/api/admin/audio-recordings?skip=0&limit=2", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 2
    assert data["total"] == 5

    response = client.get("/api/admin/audio-recordings?skip=2&limit=2", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 2

    response = client.get("/api/admin/audio-recordings?skip=10&limit=2", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 0


def test_read_audio_recordings_filter_by_judge_id(auth_headers, setup_basic_data):
    """Фильтр по судье."""
    db = next(get_db())
    try:
        judge2 = Judge(last_name="Петров", first_name="Андрей", current_court_id=setup_basic_data["court_id"])
        db.add(judge2)
        db.flush()

        db.add(AudioRecording(
            file_name="recording_1.mp3",
            s3_key="audio/2024/01/recording_1.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="created",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"]
        ))
        db.add(AudioRecording(
            file_name="recording_2.mp3",
            s3_key="audio/2024/01/recording_2.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="created",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=judge2.id,
            court_id=setup_basic_data["court_id"]
        ))
        db.commit()
    finally:
        db.close()

    response = client.get(f"/api/admin/audio-recordings?judge_id={setup_basic_data['judge_id']}", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 1
    assert data["items"][0]["judge"]["last_name"] == "Иванов"


def test_read_audio_recordings_filter_by_court_id(auth_headers, setup_basic_data):
    """Фильтр по суду."""
    db = next(get_db())
    try:
        court2 = Court(name="Другой суд")
        db.add(court2)
        db.flush()

        db.add(AudioRecording(
            file_name="recording_1.mp3",
            s3_key="audio/2024/01/recording_1.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="created",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"]
        ))
        db.add(AudioRecording(
            file_name="recording_2.mp3",
            s3_key="audio/2024/01/recording_2.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="created",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=court2.id
        ))
        db.commit()
    finally:
        db.close()

    response = client.get(f"/api/admin/audio-recordings?court_id={setup_basic_data['court_id']}", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 1
    assert data["items"][0]["court"]["name"] == "Тестовый суд"


def test_read_audio_recordings_filter_by_case_number(auth_headers, setup_basic_data):
    """Фильтр по номеру дела."""
    db = next(get_db())
    try:
        case2 = CaseEntity(number="2-456/2024")
        db.add(case2)
        db.flush()

        db.add(AudioRecording(
            file_name="recording_1.mp3",
            s3_key="audio/2024/01/recording_1.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="created",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"]
        ))
        db.add(AudioRecording(
            file_name="recording_2.mp3",
            s3_key="audio/2024/01/recording_2.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="created",
            case_entity_id=case2.id,
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"]
        ))
        db.commit()
    finally:
        db.close()

    response = client.get("/api/admin/audio-recordings?case_number=1-234", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 1
    assert data["items"][0]["case_entity"]["number"] == "1-234/2024"


def test_read_audio_recordings_filter_by_date_range(auth_headers, setup_basic_data):
    """Фильтр по диапазону дат создания."""
    db = next(get_db())
    try:
        db.add(AudioRecording(
            file_name="recording_1.mp3",
            s3_key="audio/2024/01/recording_1.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="created",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"],
            created_at=datetime(2024, 1, 10, 10, 0, 0)
        ))
        db.add(AudioRecording(
            file_name="recording_2.mp3",
            s3_key="audio/2024/01/recording_2.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="created",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"],
            created_at=datetime(2024, 1, 20, 10, 0, 0)
        ))
        db.add(AudioRecording(
            file_name="recording_3.mp3",
            s3_key="audio/2024/02/recording_3.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="created",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"],
            created_at=datetime(2024, 2, 1, 10, 0, 0)
        ))
        db.commit()
    finally:
        db.close()

    response = client.get(
        "/api/admin/audio-recordings?date_from=2024-01-15&date_to=2024-01-25",
        headers=auth_headers
    )
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 1
    assert data["items"][0]["file_name"] == "recording_2.mp3"


def test_read_audio_recordings_filter_by_transcription_status(auth_headers, setup_basic_data):
    """Фильтр по статусу расшифровки."""
    db = next(get_db())
    try:
        db.add(AudioRecording(
            file_name="recording_1.mp3",
            s3_key="audio/2024/01/recording_1.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="completed",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"]
        ))
        db.add(AudioRecording(
            file_name="recording_2.mp3",
            s3_key="audio/2024/01/recording_2.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="in_progress",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"]
        ))
        db.add(AudioRecording(
            file_name="recording_3.mp3",
            s3_key="audio/2024/01/recording_3.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="created",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"]
        ))
        db.commit()
    finally:
        db.close()

    response = client.get("/api/admin/audio-recordings?transcription_status=completed", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 1
    assert data["items"][0]["transcription_status"] == "completed"


def test_read_audio_recording_detail(auth_headers, setup_basic_data):
    """Просмотр полной информации об аудиозаписи."""
    db = next(get_db())
    try:
        recording = AudioRecording(
            file_name="recording_001.mp3",
            s3_key="audio/2024/01/recording_001.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="completed",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"],
            transcription="Текст расшифровки"
        )
        db.add(recording)
        db.commit()
        recording_id = recording.id
    finally:
        db.close()

    response = client.get(f"/api/admin/audio-recordings/{recording_id}", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()

    assert data["file_name"] == "recording_001.mp3"
    assert data["duration_seconds"] == 3600
    assert data["file_size"] == 1024000
    assert data["transcription_status"] == "completed"
    assert data["transcription"] == "Текст расшифровки"
    assert "presigned_url" in data
    assert data["presigned_url"].startswith("http")


def test_read_audio_recording_not_found(auth_headers):
    """Аудиозапись не найдена — 404."""
    response = client.get("/api/admin/audio-recordings/99999", headers=auth_headers)
    assert response.status_code == 404
    assert "не найдена" in response.json()["detail"]


def test_read_audio_recordings_without_auth():
    """Без авторизации — 401."""
    response = client.get("/api/admin/audio-recordings")
    assert response.status_code == 401


def test_read_audio_recording_without_auth():
    """Без авторизации на просмотр детали — 401."""
    response = client.get("/api/admin/audio-recordings/1")
    assert response.status_code == 401


def test_read_audio_recordings_sorting_by_created_at(auth_headers, setup_basic_data):
    """Сортировка по дате создания (DESC)."""
    db = SessionLocal()
    try:
        db.add(AudioRecording(
            file_name="recording_1.mp3",
            s3_key="audio/2024/01/recording_1.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="created",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"],
            created_at=datetime(2024, 1, 1, 10, 0, 0)
        ))
        db.add(AudioRecording(
            file_name="recording_2.mp3",
            s3_key="audio/2024/02/recording_2.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="created",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"],
            created_at=datetime(2024, 2, 1, 10, 0, 0)
        ))
        db.add(AudioRecording(
            file_name="recording_3.mp3",
            s3_key="audio/2024/03/recording_3.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="created",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"],
            created_at=datetime(2024, 3, 1, 10, 0, 0)
        ))
        db.commit()
    finally:
        db.close()

    response = client.get("/api/admin/audio-recordings", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 3
    # Сортировка DESC: сначала новые
    assert data["items"][0]["file_name"] == "recording_3.mp3"
    assert data["items"][1]["file_name"] == "recording_2.mp3"
    assert data["items"][2]["file_name"] == "recording_1.mp3"


def test_create_audio_recording_success(auth_headers, setup_basic_data, audio_file):
    """Успешное создание аудиозаписи."""
    file_content, filename, content_type = audio_file

    with patch("app.services.audio_recording.s3_service.upload_file") as mock_s3:
        mock_s3.return_value = "audio/test-key.mp3"
        
        response = client.post(
            "/api/admin/audio-recordings",
            headers=auth_headers,
            files={"file": (filename, file_content, content_type)},
            data={
                "case_entity_id": setup_basic_data["case_entity_id"],
                "judge_id": setup_basic_data["judge_id"],
                "court_id": setup_basic_data["court_id"]
            }
        )

    assert response.status_code == 200
    data = response.json()
    assert data["file_name"] == filename
    assert data["mime_type"] == content_type
    assert data["case_entity"]["id"] == setup_basic_data["case_entity_id"]
    assert data["judge"]["id"] == setup_basic_data["judge_id"]
    assert data["court"]["id"] == setup_basic_data["court_id"]
    assert data["transcription_status"] == "created"
    assert "presigned_url" in data


def test_create_audio_recording_invalid_mime_type(auth_headers, setup_basic_data):
    """Отказ при не-audio/* типе файла."""
    file_content = b"fake pdf content"
    
    with patch("app.services.audio_recording.s3_service"):
        response = client.post(
            "/api/admin/audio-recordings",
            headers=auth_headers,
            files={"file": ("document.pdf", file_content, "application/pdf")},
            data={
                "case_entity_id": setup_basic_data["case_entity_id"],
                "judge_id": setup_basic_data["judge_id"],
                "court_id": setup_basic_data["court_id"]
            }
        )
    
    assert response.status_code == 400
    assert "audio/*" in response.json()["detail"]


def test_create_audio_recording_too_large(auth_headers, setup_basic_data):
    """Отказ при файле > 500 МБ."""
    # Создаём mock-файл с завышенным размером
    large_content = b"x" * (501 * 1024 * 1024)  # 501 MB

    with patch("app.services.audio_recording.s3_service"):
        response = client.post(
            "/api/admin/audio-recordings",
            headers=auth_headers,
            files={"file": ("large.mp3", large_content, "audio/mpeg")},
            data={
                "case_entity_id": setup_basic_data["case_entity_id"],
                "judge_id": setup_basic_data["judge_id"],
                "court_id": setup_basic_data["court_id"]
            }
        )

    assert response.status_code == 413
    assert "500" in response.json()["detail"]


def test_create_audio_recording_without_auth(setup_basic_data, audio_file):
    """Создание без авторизации — 401."""
    file_content, filename, content_type = audio_file

    response = client.post(
        "/api/admin/audio-recordings",
        files={"file": (filename, file_content, content_type)},
        data={
            "case_entity_id": setup_basic_data["case_entity_id"],
            "judge_id": setup_basic_data["judge_id"],
            "court_id": setup_basic_data["court_id"]
        }
    )

    assert response.status_code == 401


def test_create_audio_recording_s3_error(auth_headers, setup_basic_data, audio_file):
    """Ошибка загрузки в S3 — 500."""
    file_content, filename, content_type = audio_file

    with patch("app.services.audio_recording.s3_service.upload_file") as mock_s3:
        mock_s3.side_effect = Exception("S3 connection failed")

        response = client.post(
            "/api/admin/audio-recordings",
            headers=auth_headers,
            files={"file": (filename, file_content, content_type)},
            data={
                "case_entity_id": setup_basic_data["case_entity_id"],
                "judge_id": setup_basic_data["judge_id"],
                "court_id": setup_basic_data["court_id"]
            }
        )

    assert response.status_code == 500
    assert "хранилище" in response.json()["detail"]


def test_create_audio_recording_missing_fields(auth_headers, audio_file):
    """Отсутствие обязательных полей — 422."""
    file_content, filename, content_type = audio_file
    
    with patch("app.services.audio_recording.s3_service"):
        response = client.post(
            "/api/admin/audio-recordings",
            headers=auth_headers,
            files={"file": (filename, file_content, content_type)},
            data={}  # нет обязательных полей
        )

    assert response.status_code == 422

def test_get_audio_recording_status_success(auth_headers, setup_basic_data):
    """Трекинг статуса — успешный ответ с минимальными данными."""
    db = SessionLocal()
    try:
        recording = AudioRecording(
            file_name="recording_status_test.mp3",
            s3_key="audio/2024/01/status_test.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            duration_seconds=3600,
            transcription_status="in_progress",
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"]
        )
        db.add(recording)
        db.commit()
        recording_id = recording.id
    finally:
        db.close()

    response = client.get(
        f"/api/admin/audio-recordings/{recording_id}/status",
        headers=auth_headers
    )
    assert response.status_code == 200
    data = response.json()

    # Проверяем только нужные поля
    assert data["id"] == recording_id
    assert data["transcription_status"] == "in_progress"
    assert "updated_at" in data

    # Убеждаемся, что нет лишних данных (лёгкий ответ)
    assert "presigned_url" not in data
    assert "transcription" not in data
    assert "case_entity" not in data
    assert "judge" not in data
    assert "court" not in data
    assert "file_name" not in data


def test_get_audio_recording_status_not_found(auth_headers):
    """Статус для несуществующей записи — 404."""
    response = client.get(
        "/api/admin/audio-recordings/99999/status",
        headers=auth_headers
    )
    assert response.status_code == 404
    assert "не найдена" in response.json()["detail"]


def test_get_audio_recording_status_without_auth():
    """Статус без авторизации — 401."""
    response = client.get("/api/admin/audio-recordings/1/status")
    assert response.status_code == 401


def test_get_audio_recording_status_all_values(auth_headers, setup_basic_data):
    """Проверка всех возможных значений transcription_status."""
    db = SessionLocal()
    try:
        statuses = ["created", "in_progress", "completed", "failed"]
        for i, status in enumerate(statuses):
            db.add(AudioRecording(
                file_name=f"recording_{status}.mp3",
                s3_key=f"audio/2024/01/recording_{status}.mp3",
                mime_type="audio/mpeg",
                file_size=1024000,
                transcription_status=status,
                case_entity_id=setup_basic_data["case_entity_id"],
                judge_id=setup_basic_data["judge_id"],
                court_id=setup_basic_data["court_id"]
            ))
        db.commit()
        recordings = db.query(AudioRecording).all()
        recording_ids = [r.id for r in recordings]
    finally:
        db.close()

    for recording_id, expected_status in zip(recording_ids, statuses):
        response = client.get(
            f"/api/admin/audio-recordings/{recording_id}/status",
            headers=auth_headers
        )
        assert response.status_code == 200
        data = response.json()
        assert data["transcription_status"] == expected_status


def test_summarize_audio_recording_success(auth_headers, setup_basic_data):
    """Запуск саммаризации — 204, задача в очереди."""
    db = SessionLocal()
    try:
        recording = AudioRecording(
            file_name="recording_summarize.mp3",
            s3_key="audio/2024/01/recording_summarize.mp3",
            mime_type="audio/mpeg",
            file_size=1024000,
            transcription_status="completed",  # можно запускать при любом статусе
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"]
        )
        db.add(recording)
        db.commit()
        recording_id = recording.id
    finally:
        db.close()

    with patch("app.services.audio_recording.enqueue_summarization") as mock_enqueue:
        response = client.post(
            f"/api/admin/audio-recordings/{recording_id}/summarize",
            headers=auth_headers
        )

    assert response.status_code == 204
    mock_enqueue.assert_called_once_with(recording_id)


def test_summarize_audio_recording_not_found(auth_headers):
    """Саммаризация несуществующей записи — 404."""
    response = client.post(
        "/api/admin/audio-recordings/99999/summarize",
        headers=auth_headers
    )
    assert response.status_code == 404
    assert "не найдена" in response.json()["detail"]


def test_summarize_audio_recording_without_auth():
    """Саммаризация без авторизации — 401."""
    response = client.post("/api/admin/audio-recordings/1/summarize")
    assert response.status_code == 401