109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
"""Unit-тесты `S3Storage` (design D-4): чтение исходника байтами и запись JSON результата — с фейковым MinIO."""
|
||
|
||
import json
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from src.core.clients import S3Storage, Storage
|
||
|
||
|
||
class _FakeResponse:
|
||
"""Двойник `aiohttp.ClientResponse`: тело и признак освобождения соединения."""
|
||
|
||
def __init__(self, body: bytes) -> None:
|
||
self.body = body
|
||
self.released = False
|
||
|
||
async def read(self) -> bytes:
|
||
return self.body
|
||
|
||
def release(self) -> None:
|
||
self.released = True
|
||
|
||
|
||
class _FakeMinio:
|
||
"""Двойник `miniopy_async.Minio`: объекты в словаре `(bucket, key) -> bytes`."""
|
||
|
||
def __init__(self) -> None:
|
||
self.objects: dict[tuple[str, str], bytes] = {}
|
||
self.put_calls: list[dict[str, Any]] = []
|
||
self.responses: list[_FakeResponse] = []
|
||
|
||
async def get_object(self, bucket_name: str, object_name: str) -> _FakeResponse:
|
||
try:
|
||
body = self.objects[(bucket_name, object_name)]
|
||
except KeyError:
|
||
raise FileNotFoundError(f"{bucket_name}/{object_name}") from None
|
||
response = _FakeResponse(body)
|
||
self.responses.append(response)
|
||
return response
|
||
|
||
async def put_object(
|
||
self, bucket_name: str, object_name: str, data: Any, length: int, content_type: str = ""
|
||
) -> None:
|
||
body = data.read()
|
||
assert len(body) == length
|
||
self.objects[(bucket_name, object_name)] = body
|
||
self.put_calls.append(
|
||
{"bucket": bucket_name, "key": object_name, "length": length, "content_type": content_type}
|
||
)
|
||
|
||
|
||
def _storage(client: _FakeMinio) -> S3Storage:
|
||
return S3Storage(client=client, bucket_name="arm-ka") # type: ignore[arg-type]
|
||
|
||
|
||
async def test_get_bytes_reads_object_and_releases_response():
|
||
client = _FakeMinio()
|
||
client.objects[("arm-ka", "uploads/f/osv.xlsx")] = b"\x50\x4b\x03\x04 xlsx"
|
||
|
||
data = await _storage(client).get_bytes("arm-ka", "uploads/f/osv.xlsx")
|
||
|
||
assert data == b"\x50\x4b\x03\x04 xlsx"
|
||
assert client.responses[0].released
|
||
|
||
|
||
async def test_get_bytes_uses_bucket_argument():
|
||
client = _FakeMinio()
|
||
client.objects[("other-bucket", "k")] = b"x"
|
||
|
||
assert await _storage(client).get_bytes("other-bucket", "k") == b"x"
|
||
|
||
|
||
async def test_get_bytes_missing_object_raises():
|
||
with pytest.raises(FileNotFoundError):
|
||
await _storage(_FakeMinio()).get_bytes("arm-ka", "uploads/нет.xlsx")
|
||
|
||
|
||
async def test_put_json_writes_utf8_json_with_content_type():
|
||
client = _FakeMinio()
|
||
obj = {"job_id": "j-1", "сообщение": "разбор", "amount": "1.10"}
|
||
|
||
await _storage(client).put_json("arm-ka", "results/f/j-1/result.json", obj)
|
||
|
||
body = client.objects[("arm-ka", "results/f/j-1/result.json")]
|
||
assert json.loads(body.decode("utf-8")) == obj
|
||
assert "разбор".encode() in body # кириллица без \u-экранирования
|
||
assert client.put_calls == [
|
||
{
|
||
"bucket": "arm-ka",
|
||
"key": "results/f/j-1/result.json",
|
||
"length": len(body),
|
||
"content_type": "application/json",
|
||
}
|
||
]
|
||
|
||
|
||
async def test_put_json_then_get_bytes_round_trip():
|
||
storage = _storage(_FakeMinio())
|
||
|
||
await storage.put_json("arm-ka", "results/x.json", {"a": [1, 2]})
|
||
|
||
assert json.loads(await storage.get_bytes("arm-ka", "results/x.json")) == {"a": [1, 2]}
|
||
|
||
|
||
def test_s3_storage_satisfies_storage_protocol():
|
||
storage: Storage = _storage(_FakeMinio())
|
||
assert isinstance(storage, Storage)
|