diff --git a/api/src/core/clients.py b/api/src/core/clients.py index ceb87e5..0035b6a 100644 --- a/api/src/core/clients.py +++ b/api/src/core/clients.py @@ -1,28 +1,105 @@ -"""Клиенты инфраструктуры. Каркас: оставлен S3Storage; outbox-паблишер (PollWorker) реализуется в Срезе 0 -по docs/context/03-architecture.md (референс — reference-old-arm/api/src/core/clients.py). +"""S3-клиент сервиса (MinIO через `miniopy-async`): чтение исходника, запись сырого ответа ИС и файла результата. + +Ошибки не проглатываются и не превращаются в `None`: каждая операция бросает исключение из иерархии +`S3StorageError` с исходной причиной в `__cause__`, чтобы воркер мог классифицировать сбой (design D-7): + +- `S3ObjectNotFoundError` (подкласс `S3ReadError`) — объекта нет (`NoSuchKey`); +- `S3ReadError` — любая другая ошибка чтения (нет доступа, нет бакета, сеть) → воркер отвечает `FILE_READ_ERROR`; +- `S3WriteError` — ошибка записи (`put_bytes` / `put_json`). + +`asyncio.CancelledError` не оборачивается — отмена задачи проходит как есть. """ +import json from io import BytesIO +from typing import Any from miniopy_async import Minio +from miniopy_async.error import S3Error + +from src.core.config import MinioSettings + +OCTET_STREAM = "application/octet-stream" +JSON_CONTENT_TYPE = "application/json" +NOT_FOUND_CODES = frozenset({"NoSuchKey"}) + + +class S3StorageError(Exception): + """Ошибка операции с S3; `bucket` и `key` — объект, на котором она случилась.""" + + def __init__(self, message: str, *, bucket: str, key: str) -> None: + super().__init__(message) + self.bucket = bucket + self.key = key + + +class S3ReadError(S3StorageError): + """Объект не прочитан (нет доступа, нет бакета, сбой сети, ...).""" + + +class S3ObjectNotFoundError(S3ReadError): + """Объекта с таким ключом в бакете нет.""" + + +class S3WriteError(S3StorageError): + """Объект не записан.""" class S3Storage: - def __init__(self, client: Minio, bucket_name: str): + """Операции сервиса над одним бакетом (`OPENBAO.MINIO.MINIO_BUCKET`); чтение может явно указать другой бакет.""" + + def __init__(self, client: Minio, bucket: str) -> None: self.client = client - self.bucket = bucket_name + self.bucket = bucket - async def put_object(self, obj_name, file_data): - try: - stream = BytesIO(file_data) - await self.client.put_object(self.bucket, obj_name, stream, length=len(file_data)) - except: - raise + @classmethod + def from_settings(cls, settings: MinioSettings) -> "S3Storage": + """Клиент MinIO из настроек: `MINIO_SECURE=false` — http (compose), `true` — https (платформа).""" + client = Minio( + settings.MINIO_URL, + access_key=settings.MINIO_ADMIN_LOGIN, + secret_key=settings.MINIO_ADMIN_PASSWORD, + secure=settings.MINIO_SECURE, + ) + return cls(client, settings.MINIO_BUCKET) - async def get_object(self, obj_name): + async def get_bytes(self, key: str, *, bucket: str | None = None) -> bytes: + """Содержимое объекта целиком. `bucket` — из запроса (`s3.bucket`), по умолчанию — бакет сервиса. + + Нет объекта → `S3ObjectNotFoundError`; иная ошибка → `S3ReadError`. + """ + bucket_name = bucket or self.bucket try: - response = await self.client.get_object(self.bucket, obj_name) - # file_bytes = await response.content.read() - return response - except: - raise + response = await self.client.get_object(bucket_name, key) + try: + data: bytes = await response.read() + finally: + response.release() + except S3Error as exc: + error_class = S3ObjectNotFoundError if exc.code in NOT_FOUND_CODES else S3ReadError + raise error_class( + f"S3: объект {bucket_name}/{key} не прочитан ({exc.code})", bucket=bucket_name, key=key + ) from exc + except Exception as exc: + raise S3ReadError( + f"S3: объект {bucket_name}/{key} не прочитан ({type(exc).__name__})", bucket=bucket_name, key=key + ) from exc + return data + + async def put_bytes(self, key: str, data: bytes, content_type: str = OCTET_STREAM) -> None: + """Записывает `data` в бакет сервиса; ошибка → `S3WriteError`.""" + try: + await self.client.put_object(self.bucket, key, BytesIO(data), length=len(data), content_type=content_type) + except Exception as exc: + raise S3WriteError( + f"S3: объект {self.bucket}/{key} не записан ({type(exc).__name__})", bucket=self.bucket, key=key + ) from exc + + async def put_json(self, key: str, obj: Any) -> None: + """Записывает `obj` как JSON в UTF-8 без `\\u`-экранирования, `Content-Type: application/json`.""" + data = json.dumps(obj, ensure_ascii=False).encode("utf-8") + await self.put_bytes(key, data, content_type=JSON_CONTENT_TYPE) + + async def close(self) -> None: + """Закрывает HTTP-сессию клиента (при остановке приложения).""" + await self.client.close_session() diff --git a/api/src/core/nats_callback.py b/api/src/core/nats_callback.py deleted file mode 100644 index e3420b5..0000000 --- a/api/src/core/nats_callback.py +++ /dev/null @@ -1,19 +0,0 @@ -"""Обработчики сообщений NATS. - -Каркас: доменный обработчик удалён. Consumer `recognition.result` (дедуп по recognition_job.result_msg_id, -маппинг JSON из S3 в document/osv_entry/financial_statement_*) реализуется в Срезе 0–2 по -docs/context/03-architecture.md. Для референса — reference-old-arm/api/src/core/nats_callback.py. -""" - -from loguru import logger -from nats.aio.client import Client as NATS - -from src.core.clients import S3Storage - - -def build_result_handler(nc: NATS, storage: S3Storage): - async def handler(msg): - logger.info("NATS message received. subject={subject}, size={size}", subject=msg.subject, size=len(msg.data)) - await msg.ack() - - return handler diff --git a/api/tests/unit/test_clients.py b/api/tests/unit/test_clients.py new file mode 100644 index 0000000..df784de --- /dev/null +++ b/api/tests/unit/test_clients.py @@ -0,0 +1,201 @@ +"""S3-клиент сервиса (`src.core.clients.S3Storage`) поверх фейкового `Minio`: аргументы вызовов, JSON в UTF-8, +классификация ошибок чтения/записи. Настоящий MinIO — в tests/integration/test_nats_minio_integration.py. +""" + +import json +from typing import Any + +import pytest + +from src.core import clients +from src.core.clients import ( + S3ObjectNotFoundError, + S3ReadError, + S3Storage, + S3StorageError, + S3WriteError, +) +from src.core.config import get_settings + + +class FakeResponse: + def __init__(self, data: bytes) -> None: + self._data = data + self.released = False + + async def read(self) -> bytes: + return self._data + + def release(self) -> None: + self.released = True + + +class FakeS3Error(Exception): + """Подмена `miniopy_async.error.S3Error`: у настоящего — атрибут `code` (`NoSuchKey`, `AccessDenied`, ...).""" + + def __init__(self, code: str) -> None: + super().__init__(code) + self.code = code + + +class FakeMinio: + def __init__(self) -> None: + self.objects: dict[tuple[str, str], tuple[bytes, str]] = {} + self.get_calls: list[tuple[str, str]] = [] + self.put_calls: list[dict[str, Any]] = [] + self.responses: list[FakeResponse] = [] + self.get_error: BaseException | None = None + self.put_error: BaseException | None = None + self.session_closed = False + + async def get_object(self, bucket_name: str, object_name: str) -> FakeResponse: + self.get_calls.append((bucket_name, object_name)) + if self.get_error is not None: + raise self.get_error + response = FakeResponse(self.objects[(bucket_name, object_name)][0]) + 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() + self.put_calls.append( + {"bucket": bucket_name, "key": object_name, "data": body, "length": length, "content_type": content_type} + ) + if self.put_error is not None: + raise self.put_error + self.objects[(bucket_name, object_name)] = (body, content_type) + + async def close_session(self) -> None: + self.session_closed = True + + +@pytest.fixture +def minio() -> FakeMinio: + return FakeMinio() + + +@pytest.fixture +def storage(minio: FakeMinio) -> S3Storage: + return S3Storage(minio, "arm-ka") + + +async def test_put_bytes_passes_bucket_key_length_and_content_type(storage: S3Storage, minio: FakeMinio): + await storage.put_bytes("uploads/a.pdf", b"%PDF-1.7", content_type="application/pdf") + + assert minio.put_calls == [ + { + "bucket": "arm-ka", + "key": "uploads/a.pdf", + "data": b"%PDF-1.7", + "length": 8, + "content_type": "application/pdf", + } + ] + + +async def test_put_bytes_default_content_type_is_octet_stream(storage: S3Storage, minio: FakeMinio): + await storage.put_bytes("k", b"x") + assert minio.put_calls[0]["content_type"] == "application/octet-stream" + + +async def test_put_json_writes_utf8_without_ascii_escaping(storage: S3Storage, minio: FakeMinio): + payload = {"error": {"code": "INVALID_DOCUMENT", "message": "Документ не распознан"}, "n": 1} + + await storage.put_json("raw/job.json", payload) + + [call] = minio.put_calls + assert call["key"] == "raw/job.json" + assert call["content_type"] == "application/json" + assert "Документ не распознан".encode() in call["data"] + assert call["length"] == len(call["data"]) + assert json.loads(call["data"].decode("utf-8")) == payload + + +async def test_get_bytes_reads_and_releases_response(storage: S3Storage, minio: FakeMinio): + minio.objects[("arm-ka", "uploads/a.pdf")] = (b"%PDF", "application/pdf") + + assert await storage.get_bytes("uploads/a.pdf") == b"%PDF" + assert minio.get_calls == [("arm-ka", "uploads/a.pdf")] + assert minio.responses[0].released is True + + +async def test_get_bytes_from_explicit_bucket(storage: S3Storage, minio: FakeMinio): + minio.objects[("other", "k")] = (b"data", "") + assert await storage.get_bytes("k", bucket="other") == b"data" + assert minio.get_calls == [("other", "k")] + + +async def test_get_bytes_missing_key_raises_not_found( + storage: S3Storage, minio: FakeMinio, monkeypatch: pytest.MonkeyPatch +): + monkeypatch.setattr(clients, "S3Error", FakeS3Error) + minio.get_error = FakeS3Error("NoSuchKey") + + with pytest.raises(S3ObjectNotFoundError) as exc_info: + await storage.get_bytes("uploads/missing.pdf") + + assert isinstance(exc_info.value, S3ReadError) + assert exc_info.value.key == "uploads/missing.pdf" + assert exc_info.value.bucket == "arm-ka" + assert exc_info.value.__cause__ is minio.get_error + + +@pytest.mark.parametrize("error", [FakeS3Error("AccessDenied"), ConnectionRefusedError(), RuntimeError("x")]) +async def test_get_bytes_other_errors_raise_read_error( + storage: S3Storage, minio: FakeMinio, monkeypatch: pytest.MonkeyPatch, error: BaseException +): + monkeypatch.setattr(clients, "S3Error", FakeS3Error) + minio.get_error = error + + with pytest.raises(S3ReadError) as exc_info: + await storage.get_bytes("uploads/a.pdf") + + assert not isinstance(exc_info.value, S3ObjectNotFoundError) + assert exc_info.value.__cause__ is error + + +async def test_put_errors_raise_write_error(storage: S3Storage, minio: FakeMinio): + minio.put_error = ConnectionRefusedError() + + with pytest.raises(S3WriteError) as exc_info: + await storage.put_json("raw/job.json", {"a": 1}) + + assert isinstance(exc_info.value, S3StorageError) + assert exc_info.value.key == "raw/job.json" + assert exc_info.value.__cause__ is minio.put_error + + +async def test_close_closes_http_session(storage: S3Storage, minio: FakeMinio): + await storage.close() + assert minio.session_closed is True + + +def test_from_settings_builds_minio_with_secure_flag(monkeypatch: pytest.MonkeyPatch): + created: list[dict[str, Any]] = [] + + def fake_minio(endpoint: str, **kwargs: Any) -> FakeMinio: + created.append({"endpoint": endpoint, **kwargs}) + return FakeMinio() + + monkeypatch.setattr(clients, "Minio", fake_minio) + settings = get_settings() # .env.test: 127.0.0.1:9000, minioadmin, бакет arm-ka, MINIO_SECURE=false + + storage = S3Storage.from_settings(settings.OPENBAO.MINIO) + + assert created == [ + {"endpoint": "127.0.0.1:9000", "access_key": "minioadmin", "secret_key": "minioadmin", "secure": False} + ] + assert storage.bucket == "arm-ka" + + +def test_from_settings_secure_true(monkeypatch: pytest.MonkeyPatch): + created: list[dict[str, Any]] = [] + monkeypatch.setattr(clients, "Minio", lambda endpoint, **kwargs: created.append(kwargs) or FakeMinio()) + monkeypatch.setenv("OPENBAO__MINIO__MINIO_SECURE", "true") + get_settings.cache_clear() + + S3Storage.from_settings(get_settings().OPENBAO.MINIO) + + assert created[0]["secure"] is True