feat(files): GET /files/{id}/content — исходный файл потоком из S3, RFC 5987 (D-10)
Поток из S3 порциями без чтения в память; нет объекта — 404 FILE_CONTENT_NOT_FOUND до заголовков; при обрыве потребителя ответ S3 закрывается и отпускается. Content-Disposition вынесен в common/filenames. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
Andrey Epifancev
parent
68b8a3d2f7
commit
310e63224d
@ -755,6 +755,7 @@ email без учёта регистра проверяет только сер
|
||||
| `FILE_SUPERSEDED` | 409 | подтверждение: документ файла, заменённого новой версией |
|
||||
| `DOCUMENT_NOT_CONFIRMABLE` | 409 | подтверждение: статус документа не `RECOGNIZED`/`NEEDS_CHECK` (например, `ERROR`) |
|
||||
| `FILE_HAS_CONFIRMED_DOCUMENTS` | 409 | удаление файла с `CONFIRMED` документом |
|
||||
| `FILE_CONTENT_NOT_FOUND` | 404 | `GET /files/{id}/content`: файл есть, объекта нет в S3 (истёк или пропал) |
|
||||
| `FILE_SUPERSEDES_OTHERS` | 409 | удаление файла, на который ссылаются `superseded_by_file_id` других файлов |
|
||||
| `CONSOLIDATION_EXISTS` | 409 | консолидация этой ГК за год и квартал уже создана (в т. ч. гонка двух `POST /consolidations`) |
|
||||
| `PERIMETER_TOO_SMALL` | 422 | в ГСК (при создании) или в новом периметре (`PUT …/members`) меньше двух ЮЛ; ничего не создаётся и не меняется |
|
||||
|
||||
@ -5,19 +5,20 @@
|
||||
|
||||
import uuid
|
||||
from typing import Annotated, Any
|
||||
from urllib.parse import quote
|
||||
|
||||
from fastapi import APIRouter, Depends, Query, Response, status
|
||||
|
||||
from src.api.dependencies.audit import RequestAudit, get_request_audit
|
||||
from src.api.dependencies.auth import require
|
||||
from src.api.dependencies.consolidations import ConsolidatedOsvServiceDep
|
||||
from src.common.filenames import content_disposition
|
||||
from src.schemas.common import ErrorDetail
|
||||
from src.schemas.consolidations import OsvFilters
|
||||
|
||||
router = APIRouter(prefix="/consolidations/versions")
|
||||
|
||||
XLSX_MEDIA_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
||||
EXPORT_ASCII_NAME = "svod-osv.xlsx"
|
||||
VIEW = [Depends(require("CONSOLIDATION_VIEW"))]
|
||||
RESPONSES: dict[int | str, dict[str, Any]] = {
|
||||
status.HTTP_200_OK: {
|
||||
@ -28,12 +29,6 @@ RESPONSES: dict[int | str, dict[str, Any]] = {
|
||||
}
|
||||
|
||||
|
||||
def content_disposition(filename: str) -> str:
|
||||
"""Заголовок вложения: ASCII-имя по умолчанию и `filename*` (RFC 5987) с полным percent-кодированием имени —
|
||||
пробел, кавычки, `;`, `/`, переводы строк и кириллица экранируются, заголовок остаётся чистым ASCII."""
|
||||
return f"attachment; filename=\"svod-osv.xlsx\"; filename*=UTF-8''{quote(filename, safe='')}"
|
||||
|
||||
|
||||
@router.get(
|
||||
"/{version_id}/osv/export",
|
||||
response_class=Response,
|
||||
@ -50,6 +45,5 @@ async def export_osv(
|
||||
"""Файл `svod-osv_<ГК>_<год>-<квартал>_v<n>.xlsx`: строки свода после фильтров (как у `GET …/osv`, без
|
||||
пагинации, в том же порядке) и лист пробелов по своду целиком. Пишет `audit_event(EXPORT)`."""
|
||||
content, filename = await service.export(version_id, filters, audit)
|
||||
return Response(
|
||||
content=content, media_type=XLSX_MEDIA_TYPE, headers={"Content-Disposition": content_disposition(filename)}
|
||||
)
|
||||
header = content_disposition(filename, ascii_fallback=EXPORT_ASCII_NAME)
|
||||
return Response(content=content, media_type=XLSX_MEDIA_TYPE, headers={"Content-Disposition": header})
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
"""`GET /legal-entities/{legal_entity_id}/files`, `GET /files/{file_id}` — файлы организации и карточка файла
|
||||
(spec documents-api «Файлы организации и карточка файла»; design D-11); `DELETE /files/{file_id}` — удаление файла
|
||||
(spec primary-files «Удаление файла»; design D-10).
|
||||
(spec primary-files «Удаление файла»; design D-10); `GET /files/{file_id}/content` — исходный файл потоком из S3
|
||||
(spec documents-api «Исходный файл»; design D-10).
|
||||
|
||||
Право `DOCUMENT_VIEW` (удаление — `DOCUMENT_EDIT`) проверяется до видимости; ЮЛ или файл вне видимости — тот же 404,
|
||||
что несуществующий.
|
||||
@ -8,9 +9,10 @@
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from typing import Annotated, Any
|
||||
from typing import Annotated, Any, Literal
|
||||
|
||||
from fastapi import APIRouter, Depends, Query, status
|
||||
from fastapi.responses import StreamingResponse
|
||||
|
||||
from src.api.dependencies.auth import require
|
||||
from src.api.dependencies.files import PrimaryFilesServiceDep
|
||||
@ -66,6 +68,36 @@ async def get_file(file_id: uuid.UUID, service: PrimaryFilesServiceDep) -> FileO
|
||||
return await service.get_file(file_id)
|
||||
|
||||
|
||||
CONTENT_RESPONSES: dict[int | str, dict[str, Any]] = {
|
||||
status.HTTP_200_OK: {
|
||||
"description": "содержимое исходного файла (тип — по `file_type`)",
|
||||
"content": {"application/octet-stream": {"schema": {"type": "string", "format": "binary"}}},
|
||||
},
|
||||
status.HTTP_404_NOT_FOUND: {
|
||||
"model": ErrorDetail,
|
||||
"description": "файл нет или вне видимости; FILE_CONTENT_NOT_FOUND — объекта нет в хранилище",
|
||||
},
|
||||
status.HTTP_503_SERVICE_UNAVAILABLE: {"model": ErrorDetail, "description": "STORAGE_UNAVAILABLE: S3 не подключено"},
|
||||
}
|
||||
|
||||
|
||||
@router.get(
|
||||
"/files/{file_id}/content",
|
||||
response_class=StreamingResponse,
|
||||
dependencies=VIEW,
|
||||
responses=CONTENT_RESPONSES,
|
||||
summary="Исходный файл",
|
||||
)
|
||||
async def get_file_content(
|
||||
file_id: uuid.UUID,
|
||||
service: PrimaryFilesServiceDep,
|
||||
disposition: Annotated[Literal["attachment", "inline"], Query()] = "attachment",
|
||||
) -> StreamingResponse:
|
||||
"""Содержимое исходного файла из S3 потоком (в память целиком не читается); `inline` — для просмотра в браузере."""
|
||||
stream, media_type, header = await service.content(file_id, disposition)
|
||||
return StreamingResponse(stream, media_type=media_type, headers={"Content-Disposition": header})
|
||||
|
||||
|
||||
@router.delete(
|
||||
"/files/{file_id}",
|
||||
status_code=status.HTTP_204_NO_CONTENT,
|
||||
|
||||
@ -6,6 +6,7 @@
|
||||
|
||||
import unicodedata
|
||||
from pathlib import PurePosixPath
|
||||
from urllib.parse import quote
|
||||
|
||||
from fastapi import HTTPException, UploadFile, status
|
||||
|
||||
@ -50,6 +51,13 @@ def safe_filename(filename: str | None, *, default: str = DEFAULT_FILENAME) -> s
|
||||
return name
|
||||
|
||||
|
||||
def content_disposition(filename: str, *, disposition: str = "attachment", ascii_fallback: str) -> str:
|
||||
"""Заголовок `Content-Disposition`: ASCII-имя `ascii_fallback` и `filename*` (RFC 5987) с полным
|
||||
percent-кодированием `filename` — пробел, кавычки, `;`, `/`, переводы строк и кириллица экранируются,
|
||||
заголовок остаётся чистым ASCII."""
|
||||
return f"{disposition}; filename=\"{ascii_fallback}\"; filename*=UTF-8''{quote(filename, safe='')}"
|
||||
|
||||
|
||||
def extension(name: str) -> str:
|
||||
"""Расширение имени в нижнем регистре без точки (`ОСВ.XLS` → `xls`; без расширения — пустая строка)."""
|
||||
return PurePosixPath(name).suffix.lstrip(".").lower()
|
||||
|
||||
@ -2,6 +2,7 @@
|
||||
по docs/context/03-architecture.md (референс — reference-old-arm/api/src/core/clients.py).
|
||||
"""
|
||||
|
||||
from collections.abc import AsyncIterator
|
||||
from io import BytesIO
|
||||
|
||||
from miniopy_async import Minio
|
||||
@ -45,3 +46,18 @@ class S3Storage:
|
||||
finally:
|
||||
response.close()
|
||||
await response.release()
|
||||
|
||||
async def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncIterator[bytes]:
|
||||
"""Объект порциями; `NoSuchKey` → `ObjectNotFoundError` до первой порции; ответ отпускается и при обрыве."""
|
||||
try:
|
||||
response = await self.client.get_object(self.bucket, obj_name)
|
||||
except S3Error as error:
|
||||
if error.code == NO_SUCH_KEY:
|
||||
raise ObjectNotFoundError(obj_name) from error
|
||||
raise
|
||||
try:
|
||||
async for chunk in response.content.iter_chunked(chunk_size):
|
||||
yield chunk
|
||||
finally:
|
||||
response.close()
|
||||
await response.release()
|
||||
|
||||
@ -4,6 +4,7 @@
|
||||
`services/directory_import.py`: тот же интерфейс нужен загрузке файлов и маппингу результатов (Срез 2c, D-12).
|
||||
"""
|
||||
|
||||
from collections.abc import AsyncIterator
|
||||
from typing import Protocol
|
||||
|
||||
|
||||
@ -12,7 +13,7 @@ class ObjectNotFoundError(Exception):
|
||||
|
||||
|
||||
class ObjectStorage(Protocol):
|
||||
"""Что ядру нужно от S3: бакет, запись и чтение объекта целиком."""
|
||||
"""Что ядру нужно от S3: бакет, запись, чтение объекта целиком и порциями."""
|
||||
|
||||
bucket: str
|
||||
|
||||
@ -21,3 +22,9 @@ class ObjectStorage(Protocol):
|
||||
async def get_bytes(self, obj_name: str) -> bytes:
|
||||
"""Содержимое объекта; отсутствие объекта — `ObjectNotFoundError`."""
|
||||
...
|
||||
|
||||
def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncIterator[bytes]:
|
||||
"""Объект порциями по `chunk_size` байт; отсутствие объекта — `ObjectNotFoundError` до первой порции.
|
||||
|
||||
Закрытие итератора (`aclose`) освобождает соединение с S3."""
|
||||
...
|
||||
|
||||
@ -18,7 +18,9 @@ audit-trail «События файлов и документов»; design D-4)
|
||||
|
||||
import hashlib
|
||||
import uuid
|
||||
from collections.abc import AsyncIterator
|
||||
from dataclasses import dataclass
|
||||
from typing import Literal
|
||||
|
||||
from loguru import logger
|
||||
from sqlalchemy import func, select
|
||||
@ -28,10 +30,10 @@ from starlette import status as http_status
|
||||
from src.api.dependencies.audit import RequestAudit
|
||||
from src.common.clock import utc_now
|
||||
from src.common.exceptions import ConflictError, DomainError, DomainValidationError, NotFoundError
|
||||
from src.common.filenames import safe_filename
|
||||
from src.common.filenames import CONTENT_TYPES, content_disposition, extension, safe_filename
|
||||
from src.contract import CONTRACT_VERSION, request_subject
|
||||
from src.core.config import Settings
|
||||
from src.core.storage import ObjectStorage
|
||||
from src.core.storage import ObjectNotFoundError, ObjectStorage
|
||||
from src.domain.file_types import MAX_ATTEMPTS, DetectedFile, detect_file
|
||||
from src.domain.policy import IN_WORK_STATUSES, VisibilityScope
|
||||
from src.models.consolidation import ConsolidationVersion
|
||||
@ -65,6 +67,7 @@ FILE_TOO_LARGE = "FILE_TOO_LARGE"
|
||||
STORAGE_UNAVAILABLE = "STORAGE_UNAVAILABLE"
|
||||
STORAGE_UNAVAILABLE_MESSAGE = "хранилище файлов недоступно"
|
||||
FILE_NOT_FOUND = "файл не найден" # одинаково для чужого и несуществующего
|
||||
FILE_CONTENT_NOT_FOUND = "FILE_CONTENT_NOT_FOUND"
|
||||
FILE_HAS_CONFIRMED_DOCUMENTS = "FILE_HAS_CONFIRMED_DOCUMENTS"
|
||||
FILE_SUPERSEDES_OTHERS = "FILE_SUPERSEDES_OTHERS"
|
||||
|
||||
@ -76,6 +79,13 @@ class StorageUnavailableError(DomainError):
|
||||
super().__init__(http_status.HTTP_503_SERVICE_UNAVAILABLE, STORAGE_UNAVAILABLE, message)
|
||||
|
||||
|
||||
class FileContentNotFoundError(DomainError):
|
||||
"""404 `FILE_CONTENT_NOT_FOUND`: файл есть в БД, но объекта нет в S3 (истёк или пропал)."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__(http_status.HTTP_404_NOT_FOUND, FILE_CONTENT_NOT_FOUND, "исходный файл недоступен в хранилище")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AcceptedFile:
|
||||
"""Принятый файл: `stored_file.id`, `recognition_job.id`, очищенное имя."""
|
||||
@ -106,6 +116,18 @@ class RejectedFile:
|
||||
)
|
||||
|
||||
|
||||
async def _chain(first: bytes, rest: AsyncIterator[bytes]) -> AsyncIterator[bytes]:
|
||||
"""Первая (уже прочитанная) порция и остальной поток; при завершении, ошибке и обрыве потребителя поток S3
|
||||
закрывается (`aclose` — там `close` + `release` ответа), соединение не остаётся занятым."""
|
||||
try:
|
||||
if first:
|
||||
yield first
|
||||
async for chunk in rest:
|
||||
yield chunk
|
||||
finally:
|
||||
await rest.aclose() # type: ignore[attr-defined]
|
||||
|
||||
|
||||
def upload_key(file_id: uuid.UUID, job_id: uuid.UUID, name: str) -> str:
|
||||
"""Ключ исходного файла в S3: `uploads/{file_id}/{job_id}/{имя}` (`03-architecture.md`, схема ключей)."""
|
||||
return f"{UPLOADS_PREFIX}/{file_id}/{job_id}/{name}"
|
||||
@ -231,6 +253,36 @@ class PrimaryFilesService:
|
||||
(card,) = await self._cards([file])
|
||||
return card
|
||||
|
||||
async def content(
|
||||
self, file_id: uuid.UUID, disposition: Literal["attachment", "inline"] = "attachment"
|
||||
) -> tuple[AsyncIterator[bytes], str, str]:
|
||||
"""Исходный файл потоком: `(поток, Content-Type, Content-Disposition)` (spec documents-api, D-10).
|
||||
|
||||
Файл любого статуса (в т. ч. `SUPERSEDED`) в области видимости; чужой и несуществующий — 404. Ключ и имя
|
||||
читаются здесь, тело потока сессию не держит (ответ уходит после коммита запроса). Первая порция берётся до
|
||||
отправки заголовков: отсутствие объекта в S3 — 404 `FILE_CONTENT_NOT_FOUND`, а не оборванный ответ.
|
||||
Аудит не пишется."""
|
||||
file = await files_repo.get_file(self._session, self._scope, file_id)
|
||||
if file is None:
|
||||
raise NotFoundError(FILE_NOT_FOUND)
|
||||
s3_object = await self._session.get(S3Object, file.s3_object_id)
|
||||
if s3_object is None:
|
||||
raise FileContentNotFoundError()
|
||||
storage = self.require_storage()
|
||||
iterator = storage.stream(s3_object.key)
|
||||
try:
|
||||
first = await anext(iterator, b"")
|
||||
except ObjectNotFoundError:
|
||||
raise FileContentNotFoundError() from None
|
||||
except BaseException:
|
||||
await iterator.aclose() # type: ignore[attr-defined]
|
||||
raise
|
||||
name = file.original_name or ""
|
||||
suffix = extension(name) or file.file_type.value.lower()
|
||||
header = content_disposition(name or f"file.{suffix}", disposition=disposition, ascii_fallback=f"file.{suffix}")
|
||||
media_type = CONTENT_TYPES.get(file.file_type.value.lower(), "application/octet-stream")
|
||||
return _chain(first, iterator), media_type, header
|
||||
|
||||
async def delete(self, file_id: uuid.UUID) -> None:
|
||||
"""Удалить видимый файл (spec primary-files «Удаление файла»; audit-trail `FILE_DELETE`; design D-10).
|
||||
|
||||
|
||||
@ -207,7 +207,7 @@ async def auth_client(app: FastAPI, seeded: AsyncSession) -> AsyncIterator[AuthC
|
||||
|
||||
|
||||
class FakeStorage:
|
||||
"""Заглушка `S3Storage` (протокол `ObjectStorage`): `bucket`/`put_object`/`get_bytes`, объекты — в словаре."""
|
||||
"""Заглушка `S3Storage` (протокол `ObjectStorage`): `put_object`/`get_bytes`/`stream`, объекты — в словаре."""
|
||||
|
||||
def __init__(self, bucket: str = "arm-ka") -> None:
|
||||
self.bucket = bucket
|
||||
@ -222,6 +222,14 @@ class FakeStorage:
|
||||
except KeyError:
|
||||
raise ObjectNotFoundError(obj_name) from None
|
||||
|
||||
async def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncIterator[bytes]:
|
||||
try:
|
||||
data = self.objects[obj_name]
|
||||
except KeyError:
|
||||
raise ObjectNotFoundError(obj_name) from None
|
||||
for start in range(0, len(data), chunk_size):
|
||||
yield data[start : start + chunk_size]
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def committed_sessionmaker() -> AsyncIterator[async_sessionmaker[AsyncSession]]:
|
||||
|
||||
167
api/tests/test_api_files_content.py
Normal file
167
api/tests/test_api_files_content.py
Normal file
@ -0,0 +1,167 @@
|
||||
"""Исходный файл потоком (spec documents-api «Исходный файл»; design D-10)."""
|
||||
|
||||
from collections.abc import AsyncIterator
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.core.storage import ObjectNotFoundError
|
||||
from src.models.enums import FileStatus, FileType, RecognitionStatus
|
||||
from src.models.files import S3Object, StoredFile
|
||||
from src.services.primary_files import _chain
|
||||
from tests.conftest import AuthClientFactory, FakeStorage
|
||||
from tests.factories import make_stored_file
|
||||
from tests.test_api_documents import CLIENT_MANAGER, CURATOR_GO, KA_RF, RKA, _rostov_entity
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
PDF = b"%PDF-1.4\n" + bytes(range(256)) * 4096 # ~1 МиБ: проверка порционной отдачи
|
||||
HTTP_OK = 200
|
||||
HTTP_NOT_FOUND = 404
|
||||
HTTP_FORBIDDEN = 403
|
||||
|
||||
|
||||
async def _file(
|
||||
session: AsyncSession, storage: FakeStorage, name: str, file_type: FileType, *, entity_name: str = "ООО Файлы"
|
||||
) -> StoredFile:
|
||||
entity = await _rostov_entity(session, name=entity_name)
|
||||
file, _ = await make_stored_file(
|
||||
session, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done, original_name=name
|
||||
)
|
||||
file.file_type = file_type
|
||||
key = (await session.execute(select(S3Object.key).where(S3Object.id == file.s3_object_id))).scalar_one()
|
||||
storage.objects[key] = PDF
|
||||
await session.flush()
|
||||
return file
|
||||
|
||||
|
||||
async def test_inline_pdf(app, seeded: AsyncSession, auth_client: AuthClientFactory) -> None: # noqa: ANN001
|
||||
storage = FakeStorage()
|
||||
app.state.storage = storage
|
||||
file = await _file(seeded, storage, "aep_6.pdf", FileType.PDF)
|
||||
client: AsyncClient = await auth_client(KA_RF)
|
||||
response = await client.get(f"/files/{file.id}/content", params={"disposition": "inline"})
|
||||
assert response.status_code == HTTP_OK
|
||||
assert response.headers["content-type"] == "application/pdf"
|
||||
assert response.headers["content-disposition"] == "inline; filename=\"file.pdf\"; filename*=UTF-8''aep_6.pdf"
|
||||
assert response.content == PDF
|
||||
|
||||
|
||||
async def test_cyrillic_name_and_missing(app, seeded: AsyncSession, auth_client: AuthClientFactory) -> None: # noqa: ANN001
|
||||
storage = FakeStorage()
|
||||
app.state.storage = storage
|
||||
file = await _file(seeded, storage, "ОСВ_66 за 2024.xls", FileType.XLS)
|
||||
client = await auth_client(KA_RF)
|
||||
response = await client.get(f"/files/{file.id}/content")
|
||||
assert response.headers["content-type"] == "application/vnd.ms-excel"
|
||||
assert response.headers["content-disposition"].startswith(
|
||||
"attachment; filename=\"file.xls\"; filename*=UTF-8''%D0%9E"
|
||||
)
|
||||
storage.objects.clear()
|
||||
missing = await client.get(f"/files/{file.id}/content")
|
||||
assert missing.status_code == HTTP_NOT_FOUND and missing.json()["detail"]["code"] == "FILE_CONTENT_NOT_FOUND"
|
||||
|
||||
|
||||
async def test_superseded_file_is_served(app, seeded: AsyncSession, auth_client: AuthClientFactory) -> None: # noqa: ANN001
|
||||
storage = FakeStorage()
|
||||
app.state.storage = storage
|
||||
file = await _file(seeded, storage, "old.xlsx", FileType.XLSX)
|
||||
file.status = FileStatus.SUPERSEDED
|
||||
await seeded.flush()
|
||||
client = await auth_client(KA_RF)
|
||||
response = await client.get(f"/files/{file.id}/content")
|
||||
assert response.status_code == HTTP_OK and response.content == PDF
|
||||
assert response.headers["content-type"] == ("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
|
||||
|
||||
|
||||
async def test_foreign_unknown_and_permission(app, seeded: AsyncSession, auth_client: AuthClientFactory) -> None: # noqa: ANN001
|
||||
storage = FakeStorage()
|
||||
app.state.storage = storage
|
||||
file = await _file(seeded, storage, "a.pdf", FileType.PDF)
|
||||
# RKA — только DOCUMENT_VIEW: доступ есть (тот же филиал); без права — 403 до видимости
|
||||
assert (await (await auth_client(RKA)).get(f"/files/{file.id}/content")).status_code == HTTP_OK
|
||||
assert (await (await auth_client(CURATOR_GO)).get(f"/files/{file.id}/content")).status_code == HTTP_FORBIDDEN
|
||||
foreign = await (await auth_client(CLIENT_MANAGER)).get(f"/files/{file.id}/content") # другой филиал
|
||||
assert foreign.status_code == HTTP_NOT_FOUND and foreign.json()["detail"]["code"] == "NOT_FOUND"
|
||||
client = await auth_client(KA_RF)
|
||||
unknown = await client.get("/files/00000000-0000-0000-0000-000000000000/content")
|
||||
assert unknown.status_code == HTTP_NOT_FOUND
|
||||
|
||||
|
||||
async def test_stream_error_before_first_chunk_is_object_not_found() -> None:
|
||||
storage = FakeStorage()
|
||||
with pytest.raises(ObjectNotFoundError):
|
||||
await storage.stream("missing").__anext__()
|
||||
|
||||
|
||||
async def test_stream_releases_on_error() -> None:
|
||||
"""Review Focus 4: при обрыве потребителя ответ S3 закрывается и отпускается."""
|
||||
from src.core.clients import S3Storage # noqa: PLC0415
|
||||
|
||||
class Content:
|
||||
async def iter_chunked(self, size: int): # noqa: ANN202
|
||||
yield b"a" * size
|
||||
yield b"b" * size
|
||||
|
||||
class Response:
|
||||
def __init__(self) -> None:
|
||||
self.content = Content()
|
||||
self.closed = self.released = False
|
||||
|
||||
def close(self) -> None:
|
||||
self.closed = True
|
||||
|
||||
async def release(self) -> None:
|
||||
self.released = True
|
||||
|
||||
response = Response()
|
||||
|
||||
class Client:
|
||||
async def get_object(self, bucket: str, name: str) -> Response:
|
||||
return response
|
||||
|
||||
storage = S3Storage(Client(), "arm-ka") # type: ignore[arg-type]
|
||||
stream = storage.stream("uploads/x", chunk_size=4)
|
||||
assert await stream.__anext__() == b"aaaa"
|
||||
await stream.aclose() # потребитель оборвал чтение
|
||||
assert response.closed and response.released
|
||||
|
||||
|
||||
class _Source:
|
||||
"""Источник порций с признаком закрытия (`finally` асинхронного генератора)."""
|
||||
|
||||
def __init__(self, fail_after: int | None = None) -> None:
|
||||
self.closed = False
|
||||
self._fail_after = fail_after
|
||||
self.iterator = self._run()
|
||||
|
||||
async def _run(self) -> AsyncIterator[bytes]:
|
||||
try:
|
||||
for index in range(3):
|
||||
if self._fail_after == index:
|
||||
raise ConnectionError("обрыв S3")
|
||||
yield b"x"
|
||||
finally:
|
||||
self.closed = True
|
||||
|
||||
|
||||
async def test_chain_closes_source_when_consumer_stops() -> None:
|
||||
"""Обрыв клиента после первой порции: `aclose` обёртки закрывает исходный поток."""
|
||||
source = _Source()
|
||||
first = await anext(source.iterator, b"")
|
||||
chain = _chain(first, source.iterator)
|
||||
assert await chain.__anext__() == b"x"
|
||||
await chain.aclose()
|
||||
assert source.closed
|
||||
|
||||
|
||||
async def test_chain_closes_source_on_error_and_yields_all() -> None:
|
||||
source = _Source(fail_after=2)
|
||||
chain = _chain(await anext(source.iterator, b""), source.iterator)
|
||||
with pytest.raises(ConnectionError):
|
||||
_ = [chunk async for chunk in chain]
|
||||
assert source.closed
|
||||
complete = _Source()
|
||||
assert b"".join([chunk async for chunk in _chain(await anext(complete.iterator, b""), complete.iterator)]) == b"xxx"
|
||||
assert complete.closed
|
||||
@ -1,6 +1,7 @@
|
||||
"""`S3Storage.get_bytes` на заглушке Minio (без сети):
|
||||
чтение, закрытие response, `NoSuchKey` → `ObjectNotFoundError`."""
|
||||
|
||||
from collections.abc import AsyncIterator
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
@ -11,11 +12,21 @@ from src.core.clients import S3Storage
|
||||
from src.core.storage import ObjectNotFoundError
|
||||
|
||||
|
||||
class StubContent:
|
||||
def __init__(self, data: bytes) -> None:
|
||||
self._data = data
|
||||
|
||||
async def iter_chunked(self, size: int) -> AsyncIterator[bytes]:
|
||||
for start in range(0, len(self._data), size):
|
||||
yield self._data[start : start + size]
|
||||
|
||||
|
||||
class StubResponse:
|
||||
"""Ответ `aiohttp` с флагами закрытия."""
|
||||
|
||||
def __init__(self, data: bytes, *, fail_read: bool = False) -> None:
|
||||
self._data = data
|
||||
self.content = StubContent(data)
|
||||
self._fail_read = fail_read
|
||||
self.closed = False
|
||||
self.released = False
|
||||
@ -83,3 +94,24 @@ async def test_get_bytes_other_s3_error_is_not_swallowed() -> None:
|
||||
with pytest.raises(S3Error) as caught:
|
||||
await _storage(StubMinio(error=_s3_error("AccessDenied"))).get_bytes("k")
|
||||
assert caught.value.code == "AccessDenied"
|
||||
|
||||
|
||||
async def test_stream_yields_chunks_and_releases_response() -> None:
|
||||
response = StubResponse(b"abcdefghij")
|
||||
|
||||
chunks = [chunk async for chunk in _storage(StubMinio(response)).stream("k", chunk_size=4)]
|
||||
|
||||
assert chunks == [b"abcd", b"efgh", b"ij"]
|
||||
assert response.closed is True
|
||||
assert response.released is True
|
||||
|
||||
|
||||
async def test_stream_no_such_key_is_object_not_found_before_first_chunk() -> None:
|
||||
with pytest.raises(ObjectNotFoundError):
|
||||
await _storage(StubMinio(error=_s3_error("NoSuchKey"))).stream("missing").__anext__()
|
||||
|
||||
|
||||
async def test_stream_other_s3_error_is_not_swallowed() -> None:
|
||||
with pytest.raises(S3Error) as caught:
|
||||
await _storage(StubMinio(error=_s3_error("AccessDenied"))).stream("k").__anext__()
|
||||
assert caught.value.code == "AccessDenied"
|
||||
|
||||
@ -48,6 +48,7 @@ EXPECTED_ROUTES: dict[str, set[str]] = {
|
||||
# primary-files, documents-api (Срез 2c)
|
||||
"/api/v1/legal-entities/{legal_entity_id}/files": {"get", "post"},
|
||||
"/api/v1/files/{file_id}": {"get", "delete"},
|
||||
"/api/v1/files/{file_id}/content": {"get"},
|
||||
"/api/v1/documents/{document_id}": {"get"},
|
||||
"/api/v1/documents/{document_id}/entries": {"get"},
|
||||
"/api/v1/documents/{document_id}/confirm": {"post"},
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user