2d69a9e867
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
"""`S3Storage.get_bytes` на заглушке Minio (без сети):
|
||
чтение, закрытие response, `NoSuchKey` → `ObjectNotFoundError`."""
|
||
|
||
from typing import Any
|
||
from unittest.mock import MagicMock
|
||
|
||
import pytest
|
||
from miniopy_async.error import S3Error
|
||
|
||
from src.core.clients import S3Storage
|
||
from src.core.storage import ObjectNotFoundError
|
||
|
||
|
||
class StubResponse:
|
||
"""Ответ `aiohttp` с флагами закрытия."""
|
||
|
||
def __init__(self, data: bytes, *, fail_read: bool = False) -> None:
|
||
self._data = data
|
||
self._fail_read = fail_read
|
||
self.closed = False
|
||
self.released = False
|
||
|
||
async def read(self) -> bytes:
|
||
if self._fail_read:
|
||
raise ConnectionError("обрыв чтения")
|
||
return self._data
|
||
|
||
def close(self) -> None:
|
||
self.closed = True
|
||
|
||
async def release(self) -> None:
|
||
self.released = True
|
||
|
||
|
||
class StubMinio:
|
||
def __init__(self, response: StubResponse | None = None, error: Exception | None = None) -> None:
|
||
self._response = response
|
||
self._error = error
|
||
self.calls: list[tuple[str, str]] = []
|
||
|
||
async def get_object(self, bucket: str, obj_name: str) -> StubResponse:
|
||
self.calls.append((bucket, obj_name))
|
||
if self._error is not None:
|
||
raise self._error
|
||
assert self._response is not None # noqa: S101
|
||
return self._response
|
||
|
||
|
||
def _s3_error(code: str) -> S3Error:
|
||
return S3Error(code, "сообщение", None, None, None, MagicMock())
|
||
|
||
|
||
def _storage(minio: StubMinio) -> S3Storage:
|
||
client: Any = minio
|
||
return S3Storage(client, "arm-ka")
|
||
|
||
|
||
async def test_get_bytes_returns_content_and_closes_response() -> None:
|
||
response = StubResponse(b"payload")
|
||
minio = StubMinio(response)
|
||
|
||
assert await _storage(minio).get_bytes("uploads/a/b.xls") == b"payload"
|
||
assert minio.calls == [("arm-ka", "uploads/a/b.xls")]
|
||
assert response.closed is True
|
||
assert response.released is True
|
||
|
||
|
||
async def test_get_bytes_closes_response_when_read_fails() -> None:
|
||
response = StubResponse(b"", fail_read=True)
|
||
|
||
with pytest.raises(ConnectionError):
|
||
await _storage(StubMinio(response)).get_bytes("k")
|
||
assert response.closed is True
|
||
assert response.released is True
|
||
|
||
|
||
async def test_get_bytes_no_such_key_is_object_not_found() -> None:
|
||
with pytest.raises(ObjectNotFoundError):
|
||
await _storage(StubMinio(error=_s3_error("NoSuchKey"))).get_bytes("missing")
|
||
|
||
|
||
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"
|