40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
"""Интеграционный тест `S3Storage` против compose-MinIO (`.env.test`: localhost:9000, бакет `arm-ka`; design D-4).
|
|
|
|
Ключ уникален на прогон (`tests/<uuid>/x.json`) и удаляется после теста.
|
|
"""
|
|
|
|
import json
|
|
import uuid
|
|
|
|
import pytest
|
|
from miniopy_async import Minio
|
|
|
|
from src.core.clients import S3Storage
|
|
from src.core.config import get_settings
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
async def test_put_json_then_get_bytes_round_trip_in_minio():
|
|
minio = get_settings().OPENBAO.MINIO
|
|
client = Minio(
|
|
minio.MINIO_URL,
|
|
access_key=minio.MINIO_ADMIN_LOGIN,
|
|
secret_key=minio.MINIO_ADMIN_PASSWORD,
|
|
secure=minio.MINIO_SECURE,
|
|
)
|
|
storage = S3Storage(client=client, bucket_name=minio.MINIO_BUCKET)
|
|
key = f"tests/{uuid.uuid4()}/x.json"
|
|
obj = {"job_id": "j-1", "сообщение": "разбор XLSX", "amount": "1234.50"}
|
|
try:
|
|
await storage.put_json(minio.MINIO_BUCKET, key, obj)
|
|
|
|
data = await storage.get_bytes(minio.MINIO_BUCKET, key)
|
|
stat = await client.stat_object(minio.MINIO_BUCKET, key)
|
|
finally:
|
|
await client.remove_object(minio.MINIO_BUCKET, key)
|
|
await client.close_session()
|
|
|
|
assert json.loads(data.decode("utf-8")) == obj
|
|
assert stat.content_type == "application/json"
|