arm-core/api/tests/test_api_files_upload.py

"""Загрузка файлов организации (spec primary-files «Загрузка», «Дедуп»; audit-trail «События файлов и документов»).

Сценарии поверх фикстуры `app` (транзакция теста с SAVEPOINT): запрос — `begin_nested()`, каждый файл — ещё один
SAVEPOINT внутри него. Реальная ОСВ и параллельный дубль идут на `committed_*` (временная БД с зафиксированным seed,
сессия запроса как в проде): публикатор outbox и второй запрос работают в своих транзакциях и видят только
зафиксированное.
"""

import asyncio
import hashlib
import json
import uuid
from collections.abc import Sequence
from datetime import timedelta
from typing import Any

import httpx
import pytest
from fastapi import FastAPI
from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from src.api.dependencies.directories import get_app_settings
from src.core.config import get_settings
from src.models.access import AppUser
from src.models.audit_ex import AuditEvent
from src.models.directories import LegalEntity, RegionalOffice
from src.models.enums import (
    AuditAction,
    AuditSection,
    FileStatus,
    FileType,
    RecognitionPipeline,
    RecognitionStatus,
    S3BucketRole,
)
from src.models.files import RecognitionJob, S3Object, StoredFile
from src.services.recognition_exchange import publish_pending
from tests.conftest import AuthClientFactory, FakeStorage
from tests.factories import make_company_group, make_legal_entity
from tests.test_recognition_exchange import REQUEST_VALIDATOR, FakeNats

HTTP_CREATED = 201
HTTP_FORBIDDEN = 403
HTTP_NOT_FOUND = 404
HTTP_UNPROCESSABLE = 422
HTTP_UNAVAILABLE = 503

KA_RF = "ka_rf@armka.local"  # Ростовский филиал
CLIENT_MANAGER = "client_manager@armka.local"  # Краснодарский филиал, право DOCUMENT_EDIT есть
CURATOR_GO = "curator_go@armka.local"  # права DOCUMENT_EDIT нет
ROSTOV = "Ростовский региональный филиал"

REAL_OSV = "ОСВ_66 за 2024.xls"
REAL_OSV_PATH = ("osv", "rb", "2024-FY", REAL_OSV)

BIFF8 = b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1"
ZIP = b"PK\x03\x04"
PDF = b"%PDF-1.7\n"
XLS_TYPE = "application/vnd.ms-excel"
XLSX_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
TINY_LIMIT = 10  # байт: «файл 51 МиБ» без 51 МиБ в памяти теста

Upload = tuple[str, bytes]


def _files_url(entity_id: uuid.UUID) -> str:
    return f"/legal-entities/{entity_id}/files"


def _xls(tag: str) -> bytes:
    """Байты с сигнатурой BIFF8 и уникальным хвостом (разный sha256 — не дубль)."""
    return BIFF8 + tag.encode()


async def _post(client: httpx.AsyncClient, entity_id: uuid.UUID, *uploads: Upload) -> httpx.Response:
    files = [("files", (name, content, "application/octet-stream")) for name, content in uploads]
    return await client.post(_files_url(entity_id), files=files)


async def _rostov_entity(session: AsyncSession, name: str = "ООО Загрузка") -> LegalEntity:
    office = (await session.execute(select(RegionalOffice).where(RegionalOffice.name == ROSTOV))).scalar_one()
    group = await make_company_group(session, name=f"ГК {name}", office=office)
    return await make_legal_entity(session, name=name, group=group)


async def _user_id(session: AsyncSession, email: str) -> uuid.UUID:
    return (await session.execute(select(AppUser.id).where(AppUser.email == email))).scalar_one()


async def _counts(session: AsyncSession) -> dict[str, int]:
    return {
        model.__tablename__: (await session.execute(select(func.count()).select_from(model))).scalar_one()
        for model in (S3Object, StoredFile, RecognitionJob)
    } | {
        "file_upload": (
            await session.execute(
                select(func.count()).select_from(AuditEvent).where(AuditEvent.action == AuditAction.FILE_UPLOAD)
            )
        ).scalar_one()
    }


async def _upload_events(session: AsyncSession) -> Sequence[AuditEvent]:
    stmt = select(AuditEvent).where(AuditEvent.action == AuditAction.FILE_UPLOAD).order_by(AuditEvent.occurred_at)
    return (await session.execute(stmt)).scalars().all()


@pytest.fixture
def storage(app: FastAPI) -> FakeStorage:
    fake = FakeStorage()
    app.state.storage = fake
    return fake


class FlakyStorage(FakeStorage):
    """S3, у которого падает запись объектов с заданным суффиксом ключа."""

    def __init__(self, failing_suffix: str) -> None:
        super().__init__()
        self.failing_suffix = failing_suffix

    async def put_object(self, obj_name: str, file_data: bytes) -> None:
        if obj_name.endswith(self.failing_suffix):
            raise ConnectionError("S3 недоступно")
        await super().put_object(obj_name, file_data)


class CapturedPublish:
    """Делегат `FakeNats`: запоминает тела опубликованных запросов."""

    def __init__(self) -> None:
        self.payloads: list[dict[str, Any]] = []

    async def publish(self, subject: str, payload: bytes, msg_id: str) -> None:  # noqa: ARG002 - протокол издателя
        self.payloads.append(json.loads(payload))


# --- «Реальная ОСВ принята и ушла в outbox» ----------------------------------------------------------------------


@pytest.mark.integration
@pytest.mark.fixtures
async def test_real_osv_is_accepted_and_published_from_outbox(
    fixtures_dir: Any,
    committed_client: AuthClientFactory,
    committed_sessionmaker: async_sessionmaker[AsyncSession],
    committed_storage: FakeStorage,
) -> None:
    content = fixtures_dir.joinpath(*REAL_OSV_PATH).read_bytes()
    async with committed_sessionmaker() as session, session.begin():
        entity = await _rostov_entity(session, "АО Русский Бисквит")
        user_id = await _user_id(session, KA_RF)
    client = await committed_client(KA_RF)

    response = await _post(client, entity.id, (REAL_OSV, content))

    assert response.status_code == HTTP_CREATED, response.text
    body = response.json()
    assert body["rejected"] == []
    (accepted,) = body["accepted"]
    file_id, job_id = uuid.UUID(accepted["file_id"]), uuid.UUID(accepted["job_id"])
    assert accepted["name"] == REAL_OSV
    key = f"uploads/{file_id}/{job_id}/{REAL_OSV}"
    assert committed_storage.objects == {key: content}

    async with committed_sessionmaker() as session:
        file = await session.get(StoredFile, file_id)
        job = await session.get(RecognitionJob, job_id)
        assert file is not None and job is not None
        s3 = await session.get(S3Object, file.s3_object_id)
        assert s3 is not None
    sha256 = hashlib.sha256(content).hexdigest()
    assert (file.status, file.file_type, file.content_hash, file.version_no) == (
        FileStatus.UPLOADED,
        FileType.XLS,
        sha256,
        1,
    )
    assert (file.legal_entity_id, file.uploaded_by, file.original_name) == (entity.id, user_id, REAL_OSV)
    assert file.current_recognition_job_id == job.id
    assert (job.status, job.pipeline, job.max_attempts, job.file_id) == (
        RecognitionStatus.queued,
        RecognitionPipeline.tabular,
        3,
        file_id,
    )
    assert (s3.bucket_role, s3.key, s3.sha256, s3.size_bytes) == (S3BucketRole.uploads, key, sha256, len(content))

    capture = CapturedPublish()
    report = await publish_pending(
        committed_sessionmaker,
        FakeNats(delegate=capture),
        ack_timeout=timedelta(seconds=30),
        max_republish=3,
    )

    assert report.published == 1
    (request,) = capture.payloads
    REQUEST_VALIDATOR.validate(request)
    assert request["s3"] == {"bucket": committed_storage.bucket, "key": key}
    assert (request["job_id"], request["file_id"], request["file_type"]) == (str(job_id), str(file_id), "XLS")
    assert request["original_name"] == REAL_OSV


# --- поля записей, имя файла, аудит ------------------------------------------------------------------------------


@pytest.mark.integration
async def test_hostile_name_gives_flat_key_and_full_records(
    auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
) -> None:
    """Review Focus 4: кириллица, пробелы, `..` и обратные слэши — ключ S3 без путей, имя без каталогов."""
    entity = await _rostov_entity(seeded)
    user_id = await _user_id(seeded, KA_RF)
    client = await auth_client(KA_RF)
    content = _xls("hostile")

    response = await _post(client, entity.id, ("../..\\ОСВ 60 (2).xls", content))

    assert response.status_code == HTTP_CREATED, response.text
    (accepted,) = response.json()["accepted"]
    file_id, job_id = uuid.UUID(accepted["file_id"]), uuid.UUID(accepted["job_id"])
    assert accepted["name"] == "ОСВ 60 (2).xls"
    key = f"uploads/{file_id}/{job_id}/ОСВ 60 (2).xls"
    assert list(storage.objects) == [key]
    assert ".." not in key
    assert "\\" not in key
    assert key.count("/") == 3  # noqa: PLR2004 - uploads/{file_id}/{job_id}/{имя}

    file = await seeded.get(StoredFile, file_id)
    job = await seeded.get(RecognitionJob, job_id)
    assert file is not None and job is not None
    s3 = await seeded.get(S3Object, file.s3_object_id)
    assert s3 is not None
    sha256 = hashlib.sha256(content).hexdigest()
    assert file.original_name == "ОСВ 60 (2).xls"
    assert (file.status, file.version_no, file.content_hash, file.size_bytes) == (
        FileStatus.UPLOADED,
        1,
        sha256,
        len(content),
    )
    assert file.uploaded_by == user_id
    assert file.uploaded_at is not None
    assert file.current_recognition_job_id == job_id
    assert job.request_subject == "recognition.tabular.request"
    assert job.schema_version == "1.0"
    assert (job.status, job.attempt, job.max_attempts) == (RecognitionStatus.queued, 1, 3)
    assert job.requested_at is not None
    assert (s3.bucket_role, s3.bucket, s3.key, s3.content_type) == (
        S3BucketRole.uploads,
        storage.bucket,
        key,
        XLS_TYPE,
    )
    assert (s3.sha256, s3.size_bytes, s3.created_by) == (sha256, len(content), f"user:{user_id}")


@pytest.mark.integration
async def test_image_file_goes_to_image_pipeline_with_one_attempt(
    auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
) -> None:
    entity = await _rostov_entity(seeded)
    client = await auth_client(KA_RF)

    response = await _post(client, entity.id, ("ББ 2024.pdf", PDF + b"scan"))

    assert response.status_code == HTTP_CREATED, response.text
    job = await seeded.get(RecognitionJob, uuid.UUID(response.json()["accepted"][0]["job_id"]))
    assert job is not None
    assert (job.pipeline, job.max_attempts, job.request_subject) == (
        RecognitionPipeline.image,
        1,
        "recognition.image.request",
    )
    assert len(storage.objects) == 1


@pytest.mark.integration
async def test_two_files_write_two_upload_events(
    auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
) -> None:
    """audit-trail «Загрузка двух файлов»: два `FILE_UPLOAD` с `object_id` каждого файла."""
    entity = await _rostov_entity(seeded)
    user_id = await _user_id(seeded, KA_RF)
    client = await auth_client(KA_RF)
    first, second = _xls("first"), ZIP + b"second"

    response = await _post(client, entity.id, ("ОСВ 60.xls", first), ("ОСВ 62.xlsx", second))

    assert response.status_code == HTTP_CREATED, response.text
    accepted = response.json()["accepted"]
    assert [item["name"] for item in accepted] == ["ОСВ 60.xls", "ОСВ 62.xlsx"]
    events = await _upload_events(seeded)
    assert len(events) == len(accepted)
    by_object = {event.object_id: event for event in events}
    for item, content, file_type in ((accepted[0], first, "XLS"), (accepted[1], second, "XLSX")):
        event = by_object[uuid.UUID(item["file_id"])]
        assert (event.section, event.object_type, event.user_id) == (
            AuditSection.PRIMARY_DOCUMENTS,
            "stored_file",
            user_id,
        )
        assert event.after_value == {
            "original_name": item["name"],
            "content_hash": hashlib.sha256(content).hexdigest(),
            "job_id": item["job_id"],
            "file_type": file_type,
        }
    assert len(storage.objects) == len(accepted)


# --- «Смесь принятых и отклонённых», «Расширение не совпадает с содержимым» ---------------------------------------


@pytest.mark.integration
async def test_mixed_request_accepts_valid_and_rejects_others(
    app: FastAPI, auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
) -> None:
    app.dependency_overrides[get_app_settings] = lambda: get_settings().model_copy(
        update={"UPLOAD_MAX_BYTES": TINY_LIMIT}
    )
    entity = await _rostov_entity(seeded)
    client = await auth_client(KA_RF)
    before = await _counts(seeded)
    xlsx = ZIP + b"ok"  # 6 байт — в пределах лимита
    big = BIFF8 + b"overflow"  # больше лимита — как файл 51 МиБ при 50 МиБ

    response = await _post(client, entity.id, ("osv.xlsx", xlsx), ("report.docx", ZIP + b"d"), ("big.xls", big))

    assert response.status_code == HTTP_CREATED, response.text
    body = response.json()
    assert [item["name"] for item in body["accepted"]] == ["osv.xlsx"]
    assert {item["name"]: item["code"] for item in body["rejected"]} == {
        "report.docx": "UNSUPPORTED_FILE",
        "big.xls": "FILE_TOO_LARGE",
    }
    assert all(item["message"] and item["file_id"] is None for item in body["rejected"])
    after = await _counts(seeded)
    assert {name: after[name] - before[name] for name in after} == {
        "s3_object": 1,
        "stored_file": 1,
        "recognition_job": 1,
        "file_upload": 1,
    }
    (key,) = storage.objects
    assert key.endswith("/osv.xlsx")
    s3 = (await seeded.execute(select(S3Object).where(S3Object.key == key))).scalar_one()
    assert s3.content_type == XLSX_TYPE


@pytest.mark.integration
async def test_signature_mismatch_is_rejected_with_422(
    auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
) -> None:
    entity = await _rostov_entity(seeded)
    client = await auth_client(KA_RF)
    before = await _counts(seeded)

    response = await _post(client, entity.id, ("osv.xlsx", _xls("biff")))

    assert response.status_code == HTTP_UNPROCESSABLE, response.text
    assert response.json()["accepted"] == []
    (rejected,) = response.json()["rejected"]
    assert (rejected["name"], rejected["code"]) == ("osv.xlsx", "UNSUPPORTED_FILE")
    assert await _counts(seeded) == before
    assert storage.objects == {}


@pytest.mark.integration
async def test_empty_filename_is_unsupported(
    auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
) -> None:
    """Review Focus 4: имя `..` (после очистки — пустое) — `UNSUPPORTED_FILE`, а не файл без имени."""
    entity = await _rostov_entity(seeded)
    client = await auth_client(KA_RF)

    response = await _post(client, entity.id, ("..", _xls("noname")))

    assert response.status_code == HTTP_UNPROCESSABLE, response.text
    (rejected,) = response.json()["rejected"]
    assert (rejected["name"], rejected["code"]) == ("", "UNSUPPORTED_FILE")
    assert storage.objects == {}


# --- «Чужое ЮЛ», права, S3 --------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_foreign_entity_is_404_like_missing(
    auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
) -> None:
    entity = await _rostov_entity(seeded)
    client = await auth_client(CLIENT_MANAGER)
    before = await _counts(seeded)

    foreign = await _post(client, entity.id, ("osv.xls", _xls("foreign")))
    missing = await _post(client, uuid.uuid4(), ("osv.xls", _xls("missing")))

    assert foreign.status_code == missing.status_code == HTTP_NOT_FOUND, foreign.text
    assert foreign.json() == missing.json()
    assert foreign.json()["detail"]["code"] == "NOT_FOUND"
    assert await _counts(seeded) == before
    assert storage.objects == {}


@pytest.mark.integration
async def test_upload_requires_document_edit(
    auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
) -> None:
    entity = await _rostov_entity(seeded)
    client = await auth_client(CURATOR_GO)

    response = await _post(client, entity.id, ("osv.xls", _xls("curator")))

    assert response.status_code == HTTP_FORBIDDEN, response.text
    assert "DOCUMENT_EDIT" in response.json()["detail"]
    assert storage.objects == {}


@pytest.mark.integration
async def test_without_storage_is_503(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    entity = await _rostov_entity(seeded)  # `app.state.storage` фикстуры `app` — None
    client = await auth_client(KA_RF)
    before = await _counts(seeded)

    response = await _post(client, entity.id, ("osv.xls", _xls("nostorage")))

    assert response.status_code == HTTP_UNAVAILABLE, response.text
    assert response.json()["detail"]["code"] == "STORAGE_UNAVAILABLE"
    assert await _counts(seeded) == before


@pytest.mark.integration
async def test_storage_failure_rejects_only_that_file(
    app: FastAPI, auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    """Отказ S3 на одном файле откатывает только его SAVEPOINT: принятый файл и его записи остаются."""
    storage = FlakyStorage(failing_suffix="/broken.xls")
    app.state.storage = storage
    entity = await _rostov_entity(seeded)
    client = await auth_client(KA_RF)
    before = await _counts(seeded)

    response = await _post(client, entity.id, ("good.xls", _xls("good")), ("broken.xls", _xls("broken")))

    assert response.status_code == HTTP_CREATED, response.text
    body = response.json()
    assert [item["name"] for item in body["accepted"]] == ["good.xls"]
    (rejected,) = body["rejected"]
    assert (rejected["name"], rejected["code"]) == ("broken.xls", "STORAGE_UNAVAILABLE")
    after = await _counts(seeded)
    assert {name: after[name] - before[name] for name in after} == {
        "s3_object": 1,
        "stored_file": 1,
        "recognition_job": 1,
        "file_upload": 1,
    }
    assert [key.rsplit("/", 1)[1] for key in storage.objects] == ["good.xls"]


# --- «Дедуп по содержимому» --------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_duplicate_in_same_entity_is_rejected_other_entity_accepts(
    auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
) -> None:
    entity = await _rostov_entity(seeded)
    other = await _rostov_entity(seeded, "ООО Другое")
    client = await auth_client(KA_RF)
    content = _xls("dup")
    first = await _post(client, entity.id, ("ОСВ 60.xls", content))
    assert first.status_code == HTTP_CREATED, first.text
    first_id = first.json()["accepted"][0]["file_id"]
    before = await _counts(seeded)

    again = await _post(client, entity.id, ("ОСВ 60 копия.xls", content))

    assert again.status_code == HTTP_UNPROCESSABLE, again.text
    (rejected,) = again.json()["rejected"]
    assert (rejected["code"], rejected["file_id"], rejected["name"]) == ("DUPLICATE_FILE", first_id, "ОСВ 60 копия.xls")
    assert await _counts(seeded) == before
    assert len(storage.objects) == 1

    elsewhere = await _post(client, other.id, ("ОСВ 60.xls", content))
    assert elsewhere.status_code == HTTP_CREATED, elsewhere.text


@pytest.mark.integration
async def test_superseded_file_does_not_block_same_content(
    auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
) -> None:
    entity = await _rostov_entity(seeded)
    client = await auth_client(KA_RF)
    content = _xls("superseded")
    first = await _post(client, entity.id, ("ОСВ 60.xls", content))
    first_id = uuid.UUID(first.json()["accepted"][0]["file_id"])
    await seeded.execute(update(StoredFile).where(StoredFile.id == first_id).values(status=FileStatus.SUPERSEDED))

    again = await _post(client, entity.id, ("ОСВ 60.xls", content))

    assert again.status_code == HTTP_CREATED, again.text
    assert again.json()["accepted"][0]["file_id"] != str(first_id)
    assert len(storage.objects) == 2  # noqa: PLR2004 - старый и новый объект


@pytest.mark.integration
async def test_concurrent_duplicates_accept_only_one(
    committed_client: AuthClientFactory,
    committed_sessionmaker: async_sessionmaker[AsyncSession],
    committed_storage: FakeStorage,
) -> None:
    """Двойной клик: два запроса с одним содержимым одновременно — принят один, второй `DUPLICATE_FILE`."""
    async with committed_sessionmaker() as session, session.begin():
        entity = await _rostov_entity(session)
    first_client, second_client = await committed_client(KA_RF), await committed_client(KA_RF)
    content = _xls("race")

    responses = await asyncio.gather(
        _post(first_client, entity.id, ("ОСВ 60.xls", content)),
        _post(second_client, entity.id, ("ОСВ 60.xls", content)),
    )

    assert sorted(response.status_code for response in responses) == [HTTP_CREATED, HTTP_UNPROCESSABLE]
    accepted = next(r for r in responses if r.status_code == HTTP_CREATED).json()["accepted"][0]
    rejected = next(r for r in responses if r.status_code == HTTP_UNPROCESSABLE).json()["rejected"][0]
    assert (rejected["code"], rejected["file_id"]) == ("DUPLICATE_FILE", accepted["file_id"])
    async with committed_sessionmaker() as session:
        files = (await session.execute(select(func.count()).select_from(StoredFile))).scalar_one()
    assert files == 1
    assert len(committed_storage.objects) == 1