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

tests/api/admin/test_documents.py
464 строки15 KB
Василий Петров
Просмотр документа
23 фев 2026, 21:23
23 фев 2026, 21:2388c3ec6
import pytest
from datetime import date
from tests.utils import clean_table
from app.models import Document, DocumentType, CaseEntity, Judge, Court, DocumentFile
from fastapi.testclient import TestClient
from app.main import app
from app.database import get_db
from app.database import SessionLocal

client = TestClient(app)


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


@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()

        doc_type = DocumentType(name="Решение")
        db.add(doc_type)
        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,
            "doc_type_id": doc_type.id,
            "case_entity_id": case_entity.id
        }
    finally:
        db.close()


def test_read_documents_empty(auth_headers):
    response = client.get("/api/admin/documents", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 0
    assert data["total"] == 0


def test_read_documents_with_data(auth_headers, setup_basic_data):
    db = next(get_db())

    doc = Document(
        title="Тестовый документ",
        date=date(2024, 1, 15),
        type_id=setup_basic_data["doc_type_id"],
        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(doc)
    db.commit()

    response = client.get("/api/admin/documents", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 1
    assert data["items"][0]["title"] == "Тестовый документ"
    assert data["items"][0]["type"]["name"] == "Решение"
    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_documents_pagination(auth_headers, setup_basic_data):
    db = next(get_db())

    for i in range(5):
        doc = Document(
            title=f"Документ {i+1}",
            date=date(2024, 1, 15),
            type_id=setup_basic_data["doc_type_id"],
            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(doc)
    db.commit()

    response = client.get("/api/admin/documents?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/documents?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/documents?skip=10&limit=2", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 0


def test_read_documents_filter_by_title(auth_headers, setup_basic_data):
    db = next(get_db())

    db.add(Document(
        title="Решение по делу А",
        date=date(2024, 1, 15),
        type_id=setup_basic_data["doc_type_id"],
        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(Document(
        title="Определение по делу Б",
        date=date(2024, 1, 15),
        type_id=setup_basic_data["doc_type_id"],
        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()

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


def test_read_documents_filter_by_date_range(auth_headers, setup_basic_data):
    db = next(get_db())

    db.add(Document(
        title="Документ 1",
        date=date(2024, 1, 10),
        type_id=setup_basic_data["doc_type_id"],
        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(Document(
        title="Документ 2",
        date=date(2024, 1, 20),
        type_id=setup_basic_data["doc_type_id"],
        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(Document(
        title="Документ 3",
        date=date(2024, 2, 1),
        type_id=setup_basic_data["doc_type_id"],
        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()

    response = client.get("/api/admin/documents?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]["title"] == "Документ 2"


def test_read_documents_filter_by_type_id(auth_headers, setup_basic_data):
    db = next(get_db())

    type2 = DocumentType(name="Определение")
    db.add(type2)
    db.flush()

    db.add(Document(
        title="Решение",
        date=date(2024, 1, 15),
        type_id=setup_basic_data["doc_type_id"],
        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(Document(
        title="Определение",
        date=date(2024, 1, 15),
        type_id=type2.id,
        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()

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


def test_read_documents_filter_by_court_id(auth_headers, setup_basic_data):
    db = next(get_db())

    court2 = Court(name="Другой суд")
    db.add(court2)
    db.flush()
    
    db.add(Document(
        title="Документ 1",
        date=date(2024, 1, 15),
        type_id=setup_basic_data["doc_type_id"],
        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(Document(
        title="Документ 2",
        date=date(2024, 1, 15),
        type_id=setup_basic_data["doc_type_id"],
        case_entity_id=setup_basic_data["case_entity_id"],
        judge_id=setup_basic_data["judge_id"],
        court_id=court2.id
    ))
    db.commit()

    response = client.get(f"/api/admin/documents?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_documents_filter_by_judge_id(auth_headers, setup_basic_data):
    db = next(get_db())

    judge2 = Judge(last_name="Петров", first_name="Андрей", current_court_id=setup_basic_data["court_id"])
    db.add(judge2)
    db.flush()

    db.add(Document(
        title="Документ 1",
        date=date(2024, 1, 15),
        type_id=setup_basic_data["doc_type_id"],
        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(Document(
        title="Документ 2",
        date=date(2024, 1, 15),
        type_id=setup_basic_data["doc_type_id"],
        case_entity_id=setup_basic_data["case_entity_id"],
        judge_id=judge2.id,
        court_id=setup_basic_data["court_id"]
    ))
    db.commit()

    response = client.get(f"/api/admin/documents?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_documents_filter_by_case_number(auth_headers, setup_basic_data):
    db = next(get_db())

    case2 = CaseEntity(number="2-456/2024")
    db.add(case2)
    db.flush()

    db.add(Document(
        title="Документ 1",
        date=date(2024, 1, 15),
        type_id=setup_basic_data["doc_type_id"],
        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(Document(
        title="Документ 2",
        date=date(2024, 1, 15),
        type_id=setup_basic_data["doc_type_id"],
        case_entity_id=case2.id,
        judge_id=setup_basic_data["judge_id"],
        court_id=setup_basic_data["court_id"]
    ))
    db.commit()

    response = client.get("/api/admin/documents?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_documents_sorting_by_date(auth_headers, setup_basic_data):
    db = next(get_db())

    db.add(Document(
        title="Поздний",
        date=date(2024, 2, 1),
        type_id=setup_basic_data["doc_type_id"],
        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(Document(
        title="Ранний",
        date=date(2024, 1, 1),
        type_id=setup_basic_data["doc_type_id"],
        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(Document(
        title="Без даты",
        date=None,
        type_id=setup_basic_data["doc_type_id"],
        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()

    response = client.get("/api/admin/documents", headers=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_read_documents_without_auth():
    response = client.get("/api/admin/documents")
    assert response.status_code == 401


def test_read_document_with_file(auth_headers, setup_basic_data):
  
    # Создаём документ с файлом
    db = SessionLocal()
    try:
        document = Document(
            title="Документ с файлом",
            date=date(2024, 2, 20),
            type_id=setup_basic_data["doc_type_id"],
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"],
            text="Текст документа"
        )
        db.add(document)
        db.commit()

        doc_file = DocumentFile(
            document_id=document.id,
            file_data=b"%PDF-1.4...",
            file_size=1024,
            mime_type="application/pdf"
        )
        db.add(doc_file)

        db.commit()
        document_id = document.id
    finally:
        db.close()

    response = client.get(f"/api/admin/documents/{document_id}", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()

    assert data["document_file"] is not None
    assert data["document_file"]["file_size"] == 1024
    assert "uploaded_at" in data["document_file"]


def test_download_document_file(auth_headers, setup_basic_data):
    """Скачивание файла документа — успешный случай."""
    db = SessionLocal()
    try:
        document = Document(
            title="Решение по делу 1-234",
            date=date(2024, 2, 20),
            type_id=setup_basic_data["doc_type_id"],
            case_entity_id=setup_basic_data["case_entity_id"],
            judge_id=setup_basic_data["judge_id"],
            court_id=setup_basic_data["court_id"],
            text="Текст решения"
        )
        db.add(document)
        db.commit()

        doc_file = DocumentFile(
            document_id=document.id,
            file_data=b"%PDF-1.4 test content",
            file_size=1024,
            mime_type="application/pdf"
        )
        db.add(doc_file)
        db.commit()
        document_id = document.id
    finally:
        db.close()

    response = client.get(
        f"/api/admin/documents/{document_id}/document_file",
        headers=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_download_document_file_without_file(auth_headers, setup_basic_data):
    """Документ есть, но файл не прикреплён — 204 No Content."""
    db = SessionLocal()
    try:
        document = Document(
            title="Документ без файла",
            date=date(2024, 3, 10),
            type_id=setup_basic_data["doc_type_id"],
            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(document)
        db.commit()
        document_id = document.id
    finally:
        db.close()

    response = client.get(
        f"/api/admin/documents/{document_id}/document_file",
        headers=auth_headers
    )
    assert response.status_code == 204
    assert response.content == b""


def test_download_document_file_without_auth():
    """Без авторизации — 401."""
    response = client.get("/api/admin/documents/1/document_file")
    assert response.status_code == 401