tests/api/lk/test_documents.py - master - tsps/core

tests/api/lk/test_documents.py
504 строки18 KB
Василий Петров
Документы в ЛК
21 мар 2026, 16:39
21 мар 2026, 16:39cc0cd1f
import pytest
from datetime import date
from tests.utils import clean_table
from app.models import Document, DocumentType, CaseEntity, Judge, Court, DocumentFile, JudgeAccount
from fastapi.testclient import TestClient
from app.main import app
from sqlalchemy.orm import Session

client = TestClient(app)


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


@pytest.fixture
def setup_lk_data(db_session: Session, lk_auth_headers):
    """
    Создаёт базовые данные для LK тестов.
    Возвращает словарь с ID созданных объектов.
    """
    # Находим текущего судью из фикстуры
    current_judge = db_session.query(Judge).filter(
        Judge.last_name == "Тест",
        Judge.first_name == "Судья"
    ).first()
    assert current_judge is not None, "Судья из фикстуры не найден"

    # Создаём суд
    court = Court(name="Тестовый суд LK")
    db_session.add(court)
    db_session.flush()

    # Создаём тип документа
    doc_type = DocumentType(name="Решение")
    db_session.add(doc_type)
    db_session.flush()

    # Создаём дело
    case_entity = CaseEntity(number="1-234/2024")
    db_session.add(case_entity)
    db_session.flush()

    db_session.commit()

    return {
        "court_id": court.id,
        "judge_id": current_judge.id,
        "doc_type_id": doc_type.id,
        "case_entity_id": case_entity.id
    }


@pytest.fixture
def setup_other_judge(db_session: Session):
    """
    Создаёт другого судью для проверки изоляции данных.
    """
    other_judge = Judge(
        last_name="Другой",
        first_name="Судья",
        middle_name="Тестович"
    )
    db_session.add(other_judge)
    db_session.flush()
    db_session.commit()

    return {"judge_id": other_judge.id}


def test_lk_read_documents_empty(lk_auth_headers):
    """ЛК: нет документов - пустой список."""
    response = client.get("/api/lk/documents", headers=lk_auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 0
    assert data["total"] == 0


def test_lk_read_documents_only_own(lk_auth_headers, db_session: Session, setup_lk_data, setup_other_judge):
    """
    ЛК: видит только свои документы, документы другого судьи не отображаются.
    """
    # Создаём документ для текущего судьи
    doc_own = Document(
        title="Мой документ",
        date=date(2024, 1, 15),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    )
    db_session.add(doc_own)

    # Создаём документ для другого судьи
    doc_other = Document(
        title="Чужой документ",
        date=date(2024, 1, 15),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_other_judge["judge_id"],
        court_id=setup_lk_data["court_id"]
    )
    db_session.add(doc_other)
    db_session.commit()

    response = client.get("/api/lk/documents", headers=lk_auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 1
    assert data["items"][0]["title"] == "Мой документ"
    assert data["total"] == 1


def test_lk_read_documents_with_data(lk_auth_headers, db_session: Session, setup_lk_data):
    """ЛК: получение списка документов с полными данными."""
    doc = Document(
        title="Тестовый документ LK",
        date=date(2024, 1, 15),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    )
    db_session.add(doc)
    db_session.commit()

    response = client.get("/api/lk/documents", headers=lk_auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 1
    assert data["items"][0]["title"] == "Тестовый документ LK"
    assert data["items"][0]["type"]["name"] == "Решение"
    assert data["items"][0]["judge"]["last_name"] == "Тест"
    assert data["items"][0]["court"]["name"] == "Тестовый суд LK"
    assert data["items"][0]["case_entity"]["number"] == "1-234/2024"


def test_lk_read_documents_pagination(lk_auth_headers, db_session: Session, setup_lk_data):
    """ЛК: пагинация работает корректно."""
    for i in range(5):
        doc = Document(
            title=f"Документ {i+1}",
            date=date(2024, 1, 15),
            type_id=setup_lk_data["doc_type_id"],
            case_entity_id=setup_lk_data["case_entity_id"],
            judge_id=setup_lk_data["judge_id"],
            court_id=setup_lk_data["court_id"]
        )
        db_session.add(doc)
    db_session.commit()

    response = client.get("/api/lk/documents?skip=0&limit=2", headers=lk_auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 2
    assert data["total"] == 5

    response = client.get("/api/lk/documents?skip=2&limit=2", headers=lk_auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 2

    response = client.get("/api/lk/documents?skip=10&limit=2", headers=lk_auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 0


def test_lk_read_documents_filter_by_title(lk_auth_headers, db_session: Session, setup_lk_data):
    """ЛК: фильтрация по заголовку."""
    db_session.add(Document(
        title="Решение по делу А",
        date=date(2024, 1, 15),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    ))
    db_session.add(Document(
        title="Определение по делу Б",
        date=date(2024, 1, 15),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    ))
    db_session.commit()

    response = client.get("/api/lk/documents?title=решение", headers=lk_auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 1
    assert data["items"][0]["title"] == "Решение по делу А"


def test_lk_read_documents_filter_by_date_range(lk_auth_headers, db_session: Session, setup_lk_data):
    """ЛК: фильтрация по диапазону дат."""
    db_session.add(Document(
        title="Документ 1",
        date=date(2024, 1, 10),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    ))
    db_session.add(Document(
        title="Документ 2",
        date=date(2024, 1, 20),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    ))
    db_session.add(Document(
        title="Документ 3",
        date=date(2024, 2, 1),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    ))
    db_session.commit()

    response = client.get(
        "/api/lk/documents?date_from=2024-01-15&date_to=2024-01-25",
        headers=lk_auth_headers
    )
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 1
    assert data["items"][0]["title"] == "Документ 2"


def test_lk_read_documents_filter_by_type_id(lk_auth_headers, db_session: Session, setup_lk_data):
    """ЛК: фильтрация по типу документа."""
    type2 = DocumentType(name="Определение")
    db_session.add(type2)
    db_session.flush()

    db_session.add(Document(
        title="Решение",
        date=date(2024, 1, 15),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    ))
    db_session.add(Document(
        title="Определение",
        date=date(2024, 1, 15),
        type_id=type2.id,
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    ))
    db_session.commit()

    response = client.get(
        f"/api/lk/documents?type_id={setup_lk_data['doc_type_id']}",
        headers=lk_auth_headers
    )
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 1
    assert data["items"][0]["title"] == "Решение"


def test_lk_read_documents_filter_by_case_number(lk_auth_headers, db_session: Session, setup_lk_data):
    """ЛК: фильтрация по номеру дела."""
    case2 = CaseEntity(number="2-456/2024")
    db_session.add(case2)
    db_session.flush()

    db_session.add(Document(
        title="Документ 1",
        date=date(2024, 1, 15),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    ))
    db_session.add(Document(
        title="Документ 2",
        date=date(2024, 1, 15),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=case2.id,
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    ))
    db_session.commit()

    response = client.get("/api/lk/documents?case_number=1-234", headers=lk_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_lk_read_documents_sorting_by_date(lk_auth_headers, db_session: Session, setup_lk_data):
    """ЛК: сортировка по дате (NULLS LAST)."""
    db_session.add(Document(
        title="Поздний",
        date=date(2024, 2, 1),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    ))
    db_session.add(Document(
        title="Ранний",
        date=date(2024, 1, 1),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    ))
    db_session.add(Document(
        title="Без даты",
        date=None,
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    ))
    db_session.commit()

    response = client.get("/api/lk/documents", headers=lk_auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 3
    assert data["items"][0]["title"] == "Ранний"
    assert data["items"][1]["title"] == "Поздний"
    assert data["items"][2]["title"] == "Без даты"


def test_lk_read_document_by_id(lk_auth_headers, db_session: Session, setup_lk_data):
    """ЛК: получение конкретного документа по ID."""
    doc = Document(
        title="Документ для просмотра",
        date=date(2024, 1, 15),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"],
        text="Полный текст документа"
    )
    db_session.add(doc)
    db_session.commit()
    doc_id = doc.id

    response = client.get(f"/api/lk/documents/{doc_id}", headers=lk_auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert data["id"] == doc_id
    assert data["title"] == "Документ для просмотра"
    assert data["text"] == "Полный текст документа"


def test_lk_read_document_other_judge_forbidden(lk_auth_headers, db_session: Session, setup_lk_data, setup_other_judge):
    """ЛК: попытка получить документ другого судьи возвращает 403."""
    # Создаём документ для другого судьи
    doc = Document(
        title="Чужой документ",
        date=date(2024, 1, 15),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_other_judge["judge_id"],
        court_id=setup_lk_data["court_id"]
    )
    db_session.add(doc)
    db_session.commit()
    doc_id = doc.id

    response = client.get(f"/api/lk/documents/{doc_id}", headers=lk_auth_headers)
    assert response.status_code == 403
    assert response.json()["detail"] == "Доступ запрещён: документ принадлежит другому судье"


def test_lk_read_document_not_found(lk_auth_headers):
    """ЛК: запрос несуществующего документа возвращает 404."""
    response = client.get("/api/lk/documents/99999", headers=lk_auth_headers)
    assert response.status_code == 404


def test_lk_download_document_file(lk_auth_headers, db_session: Session, setup_lk_data):
    """ЛК: скачивание файла документа."""
    doc = Document(
        title="Решение по делу LK",
        date=date(2024, 2, 20),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"],
        text="Текст решения"
    )
    db_session.add(doc)
    db_session.flush()

    doc_file = DocumentFile(
        document_id=doc.id,
        file_data=b"%PDF-1.4 test content",
        file_size=1024,
        mime_type="application/pdf"
    )
    db_session.add(doc_file)
    db_session.commit()
    doc_id = doc.id

    response = client.get(
        f"/api/lk/documents/{doc_id}/document_file",
        headers=lk_auth_headers
    )
    assert response.status_code == 200
    assert response.content == b"%PDF-1.4 test content"
    assert response.headers["content-type"] == "application/pdf"
    assert "filename*=UTF-8''" in response.headers["content-disposition"]


def test_lk_download_document_file_without_file(lk_auth_headers, db_session: Session, setup_lk_data):
    """ЛК: документ есть, но файл не прикреплён — 204 No Content."""
    doc = Document(
        title="Документ без файла",
        date=date(2024, 3, 10),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    )
    db_session.add(doc)
    db_session.commit()
    doc_id = doc.id

    response = client.get(
        f"/api/lk/documents/{doc_id}/document_file",
        headers=lk_auth_headers
    )
    assert response.status_code == 204
    assert response.content == b""


def test_lk_download_document_file_other_judge_forbidden(lk_auth_headers, db_session: Session, setup_lk_data, setup_other_judge):
    """ЛК: попытка скачать файл чужого документа возвращает 403."""
    doc = Document(
        title="Чужой документ с файлом",
        date=date(2024, 3, 10),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_other_judge["judge_id"],
        court_id=setup_lk_data["court_id"]
    )
    db_session.add(doc)
    db_session.flush()

    doc_file = DocumentFile(
        document_id=doc.id,
        file_data=b"test",
        file_size=4,
        mime_type="application/pdf"
    )
    db_session.add(doc_file)
    db_session.commit()
    doc_id = doc.id

    response = client.get(
        f"/api/lk/documents/{doc_id}/document_file",
        headers=lk_auth_headers
    )
    assert response.status_code == 403
    assert response.json()["detail"] == "Доступ запрещён: документ принадлежит другому судье"


def test_lk_download_document_file_without_auth(db_session: Session, setup_lk_data):
    """ЛК: без авторизации — 401."""
    doc = Document(
        title="Документ",
        date=date(2024, 3, 10),
        type_id=setup_lk_data["doc_type_id"],
        case_entity_id=setup_lk_data["case_entity_id"],
        judge_id=setup_lk_data["judge_id"],
        court_id=setup_lk_data["court_id"]
    )
    db_session.add(doc)
    db_session.flush()
    db_session.commit()
    doc_id = doc.id

    response = client.get(f"/api/lk/documents/{doc_id}/document_file")
    assert response.status_code == 401


def test_lk_read_documents_without_auth():
    """ЛК: без авторизации — 401."""
    response = client.get("/api/lk/documents")
    assert response.status_code == 401