diff --git a/api/pyproject.toml b/api/pyproject.toml index cea3ce9..6546cb8 100644 --- a/api/pyproject.toml +++ b/api/pyproject.toml @@ -25,7 +25,7 @@ explicit_package_bases = true # Строгий режим для собственного кода сервиса (`src`). `strict = true` в overrides mypy (2.x) применяется глобально, # а не к модулю, поэтому флаги семейства strict перечислены явно. [[tool.mypy.overrides]] -module = ["src.contract.*", "src.gateway.*", "src.normalizer.*", "src.scan_emulator.*"] +module = ["src.contract.*", "src.core.clients", "src.core.nats", "src.gateway.*", "src.normalizer.*", "src.scan_emulator.*"] disallow_any_generics = true disallow_subclassing_any = true disallow_untyped_calls = true diff --git a/api/src/core/nats.py b/api/src/core/nats.py new file mode 100644 index 0000000..f72608a --- /dev/null +++ b/api/src/core/nats.py @@ -0,0 +1,309 @@ +"""Соединение scan-gateway с NATS: core NATS без JetStream (design D-2, D-7, D-8; spec image-recognition-gateway, +«Соединение с NATS без JetStream»; docs/context/03-architecture.md, «Контракт обмена через NATS»). + +- Аутентификация как в шаблоне: заданы `NATS_SEED` и `NATS_JWT` — nkeys/JWT; оба пусты — без учётных данных; задан + только один или seed не читается — `NatsConfigError` (подключиться «как-нибудь» без учётных данных нельзя). +- `NATS_ENABLED=false` — `connect()` ничего не делает и возвращает `False` (dev без NATS). +- Переподключение — встроенное в nats-py (`max_reconnect_attempts=-1`): клиент сам повторяет и первое подключение, + поэтому `connect()` возвращается только при успехе (вызывающий оборачивает его в задачу или `wait_for`). + После переподключения nats-py сам отправляет SUB для всех подписок с прежней queue group — подписку повторно + оформлять не нужно. +- `subscribe_requests(cb)` — подписка на `recognition.image.request` в queue group `scan-gateway`: из нескольких + инстансов сообщение получает ровно один. `cb` получает исходное сообщение nats-py (`msg.data`, `msg.headers`); + исключения `cb` логируются, подписка живёт. +- `publish(subject, payload, msg_id)` — публикация с заголовком `Nats-Msg-Id` и `flush` (подтверждение сервером) с + повторами на временных ошибках; исчерпаны — `PublishFailed`. Воркер логирует error и идёт дальше: ядро закроет + задание по `reply_deadline` (D-7). +- Stream'ов и consumer'ов JetStream нет: `jetstream()` не вызывается. +""" + +import asyncio +import base64 +import json +from collections.abc import Awaitable, Callable +from contextlib import suppress +from typing import Any + +import nkeys +from loguru import logger +from nats import errors as nats_errors +from nats.aio.client import Client as NATS +from nats.aio.msg import Msg + +from src.contract import ( + IMAGE_REQUEST_SUBJECT, + NATS_MSG_ID_HEADER, + RESULT_SUBJECT, + MessageStatus, + RecognitionResultMessage, + ack_msg_id, + result_msg_id, +) +from src.core.config import NATSSettings, Settings + +REQUEST_QUEUE_GROUP = "scan-gateway" +MAX_RECONNECT_ATTEMPTS = -1 # переподключаться бесконечно +RECONNECT_TIME_WAIT = 2 # секунд между попытками переподключения nats-py +CONNECT_TIMEOUT = 2 # секунд на одну попытку TCP-подключения +FLUSH_TIMEOUT = 5 # секунд на подтверждение публикации сервером (PING/PONG) +RETRY_DELAY = 0.5 # секунд между попытками публикации + +MessageHandler = Callable[[Msg], Awaitable[None]] +SleepFn = Callable[[float], Awaitable[None]] + +# Временные ошибки публикации: соединение закрыто/переподключается/переполнен буфер на время переподключения, +# сервер не подтвердил flush. Повтор имеет смысл. Остальное (MaxPayloadError, BadSubjectError, ошибки программы) +# повтором не лечится — `PublishFailed` сразу. +TRANSIENT_PUBLISH_ERRORS: tuple[type[BaseException], ...] = ( + nats_errors.ConnectionClosedError, + nats_errors.ConnectionReconnectingError, + nats_errors.ConnectionDrainingError, + nats_errors.OutboundBufferLimitError, + nats_errors.StaleConnectionError, + nats_errors.NoServersError, + TimeoutError, # nats.errors.TimeoutError / FlushTimeoutError — его подклассы + ConnectionError, +) + + +class NatsError(Exception): + """Базовая ошибка слоя NATS сервиса.""" + + +class NatsConfigError(NatsError, RuntimeError): + """Учётные данные NATS заданы некорректно: подключение не выполняется.""" + + +class NatsNotConnectedError(NatsError): + """Операция требует соединения с NATS.""" + + +class PublishFailed(NatsError): + """Публикация не удалась: попытки исчерпаны или ошибка не временная. Причина последней попытки — в `__cause__`.""" + + def __init__(self, message: str, *, subject: str, msg_id: str, attempts: int) -> None: + super().__init__(message) + self.subject = subject + self.msg_id = msg_id + self.attempts = attempts + + +def build_nats_auth(settings: NATSSettings) -> dict[str, Any]: + """Параметры аутентификации для `Client.connect`: `{}` без учётных данных или колбэки nkeys/JWT шаблона.""" + seed = settings.NATS_SEED.strip() + jwt = settings.NATS_JWT.strip() + if not seed and not jwt: + return {} + if not seed or not jwt: + raise NatsConfigError( + "Неполные учётные данные NATS: NATS_SEED и NATS_JWT задаются вместе или не задаются вовсе" + ) + try: + keypair = nkeys.from_seed(seed.encode("utf-8")) + except Exception as exc: + raise NatsConfigError(f"Некорректный NATS_SEED: {type(exc).__name__}") from None + + def user_jwt_cb() -> bytes: + return jwt.encode("utf-8") + + def signature_cb(nonce: str | bytes) -> bytes: + raw = nonce.encode("utf-8") if isinstance(nonce, str) else nonce + signature: bytes = keypair.sign(raw) + return base64.b64encode(signature) + + return {"user_jwt_cb": user_jwt_cb, "signature_cb": signature_cb} + + +def message_msg_id(message: RecognitionResultMessage, attempt: int) -> str: + """`Nats-Msg-Id` конверта: `ack::` для `processing`, иначе `result::`.""" + if message.status is MessageStatus.PROCESSING: + return ack_msg_id(message.job_id, attempt) + return result_msg_id(message.job_id, attempt) + + +class NatsGateway: + """Соединение сервиса с NATS: подключение, подписка на запросы, публикация с `Nats-Msg-Id` и повторами. + + `client_factory`, `sleep`, `retry_delay`, `flush_timeout` — подмены для тестов и тайминги. + """ + + def __init__( # noqa: PLR0913 - все параметры, кроме settings, именованные: подмены для тестов и тайминги + self, + settings: Settings, + *, + client_factory: Callable[[], Any] = NATS, + sleep: SleepFn = asyncio.sleep, + retry_delay: float = RETRY_DELAY, + flush_timeout: int = FLUSH_TIMEOUT, + ) -> None: + self._settings = settings + self._client_factory = client_factory + self._sleep = sleep + self._retry_delay = retry_delay + self._flush_timeout = flush_timeout + self._client: Any = None + self._closing = False + + # --- состояние и жизненный цикл --------------------------------------------------------------------------- + @property + def is_connected(self) -> bool: + """Фактическое состояние: `False` до подключения, при `NATS_ENABLED=false`, во время переподключения и после + `close()`. Источник поля `nats` в `/healthcheck`.""" + return self._client is not None and bool(self._client.is_connected) + + async def connect(self) -> bool: + """Подключается к `NATS_URL`; `False` без подключения, если `NATS_ENABLED=false`. + + Ждёт сервер сколь угодно долго (nats-py повторяет и первое подключение). Ошибка учётных данных — + `NatsConfigError` до попытки подключения. + """ + if not self._settings.NATS_ENABLED: + logger.info("NATS выключен (NATS_ENABLED=false): подключение не выполняется") + return False + nats_settings = self._settings.OPENBAO.NATS + auth = build_nats_auth(nats_settings) + client = self._client_factory() + self._closing = False + try: + await client.connect( + nats_settings.NATS_URL, + name=self._settings.SERVICE_NAME, + error_cb=self._on_error, + disconnected_cb=self._on_disconnected, + reconnected_cb=self._on_reconnected, + closed_cb=self._on_closed, + max_reconnect_attempts=MAX_RECONNECT_ATTEMPTS, + reconnect_time_wait=RECONNECT_TIME_WAIT, + connect_timeout=CONNECT_TIMEOUT, + **auth, + ) + except BaseException: + with suppress(Exception): + await client.close() + raise + self._client = client + logger.info("Подключение к NATS установлено. url={url}", url=nats_settings.NATS_URL) + return True + + async def close(self) -> None: + """Подключённый клиент — `drain` (дообработать полученное и отправить буфер), иначе `close`; без исключений.""" + client = self._client + if client is None or client.is_closed: + return + self._closing = True + try: + if client.is_connected: + await client.drain() + else: + await client.close() + except Exception as exc: + logger.warning( + "Ошибка при закрытии соединения NATS. error_type={error_type}, error={error}", + error_type=type(exc).__name__, + error=repr(exc), + ) + with suppress(Exception): + await client.close() + + # --- подписка --------------------------------------------------------------------------------------------- + async def subscribe_requests(self, cb: MessageHandler, queue: str = REQUEST_QUEUE_GROUP) -> Any: + """Подписка на `recognition.image.request` в queue group `queue`; возвращает подписку nats-py после + подтверждения сервером (`flush`). + + `cb` получает исходное `Msg` (`data`, `headers`); его исключения логируются, подписка не падает. + """ + if self._client is None: + raise NatsNotConnectedError(f"Нет соединения с NATS: подписка на {IMAGE_REQUEST_SUBJECT} невозможна") + + async def safe_cb(msg: Msg) -> None: + try: + await cb(msg) + except Exception as exc: + logger.exception( + "Ошибка обработки сообщения NATS. subject={subject}, error_type={error_type}, error={error}", + subject=msg.subject, + error_type=type(exc).__name__, + error=repr(exc), + ) + + subscription = await self._client.subscribe(IMAGE_REQUEST_SUBJECT, queue=queue, cb=safe_cb) + await self._client.flush(timeout=self._flush_timeout) # SUB принят сервером: подписка действует по возврату + logger.info( + "Подписка NATS активна. subject={subject}, queue={queue}", subject=IMAGE_REQUEST_SUBJECT, queue=queue + ) + return subscription + + # --- публикация ------------------------------------------------------------------------------------------- + async def publish(self, subject: str, payload: bytes, msg_id: str, *, retries: int | None = None) -> None: + """Публикует `payload` с заголовком `Nats-Msg-Id: ` и ждёт подтверждения сервера (`flush`). + + `retries` — число попыток **всего** (по умолчанию `PUBLISH_RETRIES` = 3: первая и до двух повторов), между + попытками — пауза `retry_delay`. Повторяются только временные ошибки (`TRANSIENT_PUBLISH_ERRORS`). Попытки + исчерпаны, ошибка не временная или соединения не было вовсе (`attempts = 0`) — `PublishFailed`. + Дубликаты при повторе безопасны: ядро дедуплицирует по `Nats-Msg-Id`. + """ + attempts_total = self._settings.PUBLISH_RETRIES if retries is None else retries + if attempts_total < 1: + raise ValueError(f"retries должно быть ≥ 1, получено {attempts_total}") + if self._client is None: + raise PublishFailed( + f"Нет соединения с NATS: {msg_id} в {subject} не опубликовано", + subject=subject, + msg_id=msg_id, + attempts=0, + ) + headers = {NATS_MSG_ID_HEADER: msg_id} + for attempt in range(1, attempts_total + 1): + try: + await self._client.publish(subject, payload, headers=headers) + await self._client.flush(timeout=self._flush_timeout) + return + except TRANSIENT_PUBLISH_ERRORS as exc: + if attempt == attempts_total: + raise PublishFailed( + f"Публикация {msg_id} в {subject} не удалась за {attempt} попыток ({type(exc).__name__})", + subject=subject, + msg_id=msg_id, + attempts=attempt, + ) from exc + logger.warning( + "Публикация в NATS не удалась, повтор. subject={subject}, msg_id={msg_id}, attempt={attempt}, " + "error_type={error_type}", + subject=subject, + msg_id=msg_id, + attempt=attempt, + error_type=type(exc).__name__, + ) + await self._sleep(self._retry_delay) + except Exception as exc: + raise PublishFailed( + f"Публикация {msg_id} в {subject} не удалась, ошибка не временная ({type(exc).__name__})", + subject=subject, + msg_id=msg_id, + attempts=attempt, + ) from exc + + async def publish_result(self, message: RecognitionResultMessage, attempt: int) -> None: + """Публикует конверт в `recognition.result`: тело — `message.to_wire()` (JSON, UTF-8), `Nats-Msg-Id` — + `ack:…` для `processing`, `result:…` для итога; `attempt` — из запроса. Ошибки — как у `publish`.""" + payload = json.dumps(message.to_wire(), ensure_ascii=False).encode("utf-8") + await self.publish(RESULT_SUBJECT, payload, message_msg_id(message, attempt)) + + # --- колбэки nats-py -------------------------------------------------------------------------------------- + async def _on_error(self, exc: Exception) -> None: + logger.warning( + "Ошибка соединения NATS. error_type={error_type}, error={error}", + error_type=type(exc).__name__, + error=repr(exc), + ) + + async def _on_disconnected(self) -> None: + if self._closing: + return + logger.warning("Соединение с NATS потеряно, nats-py переподключается") + + async def _on_reconnected(self) -> None: + logger.info("Соединение с NATS восстановлено, подписки восстановлены клиентом") + + async def _on_closed(self) -> None: + logger.info("Соединение с NATS закрыто") diff --git a/api/tests/integration/test_nats_minio_integration.py b/api/tests/integration/test_nats_minio_integration.py new file mode 100644 index 0000000..d1f0810 --- /dev/null +++ b/api/tests/integration/test_nats_minio_integration.py @@ -0,0 +1,213 @@ +"""Интеграция `src.core.nats` и `src.core.clients` с compose-инфраструктурой (`make up`): NATS без JetStream на +127.0.0.1:4222, MinIO на 127.0.0.1:9000 (бакет `arm-ka`); параметры — `api/.env.test`. + +Недоступная инфраструктура — падение с понятным сообщением, не skip. Общий контейнер NATS не перезапускается: +сценарий переподключения покрыт unit-тестом колбэков (tests/unit/test_nats.py). Чужой трафик на тех же subject'ах +отсекается уникальным `job_id` теста; ключи MinIO — под уникальным префиксом `itest//` и удаляются. +""" + +import asyncio +import json +import socket +import uuid +from collections.abc import AsyncIterator +from typing import Any +from urllib.parse import urlparse + +import pytest +from nats.aio.client import Client as NATS +from nats.aio.msg import Msg + +from src.contract import ( + IMAGE_REQUEST_SUBJECT, + NATS_MSG_ID_HEADER, + RESULT_SUBJECT, + ServiceInfo, + build_ack_message, +) +from src.core.clients import S3ObjectNotFoundError, S3Storage +from src.core.config import Settings, get_settings +from src.core.nats import REQUEST_QUEUE_GROUP, NatsGateway + +pytestmark = pytest.mark.integration + +CONNECT_WAIT = 5.0 # секунд на подключение к NATS (nats-py ждёт сервер бесконечно — ограничиваем снаружи) +DELIVERY_WAIT = 5.0 + + +def _require_port(host: str, port: int, what: str) -> None: + try: + with socket.create_connection((host, port), timeout=2): + return + except OSError as exc: + pytest.fail(f"{what} недоступен на {host}:{port} ({exc!r}): поднимите инфраструктуру `make up`") + + +@pytest.fixture +def settings() -> Settings: + settings = get_settings() + nats_url = urlparse(settings.OPENBAO.NATS.NATS_URL) + _require_port(nats_url.hostname or "127.0.0.1", nats_url.port or 4222, "NATS") + host, _, port = settings.OPENBAO.MINIO.MINIO_URL.partition(":") + _require_port(host, int(port or 9000), "MinIO") + return settings + + +def _no_jetstream_client() -> NATS: + """Настоящий клиент nats-py, у которого вызов `jetstream()` — ошибка теста (D-8: JetStream не используется).""" + client = NATS() + + def forbidden(**_: Any) -> None: + raise AssertionError("JetStream не используется (D-8)") + + client.jetstream = forbidden # type: ignore[method-assign] + return client + + +async def _connected_gateway(settings: Settings) -> NatsGateway: + gateway = NatsGateway(settings, client_factory=_no_jetstream_client, retry_delay=0.1) + try: + connected = await asyncio.wait_for(gateway.connect(), CONNECT_WAIT) + except TimeoutError: + await gateway.close() + pytest.fail(f"Нет подключения к NATS {settings.OPENBAO.NATS.NATS_URL} за {CONNECT_WAIT} с: `make up`") + assert connected is True + assert gateway.is_connected is True + return gateway + + +@pytest.fixture +async def gateways(settings: Settings) -> AsyncIterator[list[NatsGateway]]: + created: list[NatsGateway] = [] + yield created + for gateway in created: + await gateway.close() + + +async def _wait_until(predicate: Any, timeout: float = DELIVERY_WAIT) -> None: + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while not predicate() and loop.time() < deadline: + await asyncio.sleep(0.05) + + +# --- NATS --------------------------------------------------------------------------------------------------------- + + +async def test_subscribe_and_publish_with_msg_id_without_jetstream(settings: Settings, gateways: list[NatsGateway]): + subscriber = await _connected_gateway(settings) + publisher = await _connected_gateway(settings) + gateways += [subscriber, publisher] + job_id = str(uuid.uuid4()) + received: list[Msg] = [] + + async def on_request(msg: Msg) -> None: + if json.loads(msg.data).get("job_id") == job_id: # чужие сообщения на общем subject пропускаем + received.append(msg) + + await subscriber.subscribe_requests(on_request) + await publisher.publish(IMAGE_REQUEST_SUBJECT, json.dumps({"job_id": job_id}).encode(), f"{job_id}:1") + await _wait_until(lambda: len(received) >= 1) + + assert len(received) == 1 + [msg] = received + assert msg.subject == IMAGE_REQUEST_SUBJECT + assert msg.headers is not None + assert msg.headers[NATS_MSG_ID_HEADER] == f"{job_id}:1" + + +async def test_queue_group_delivers_each_message_to_exactly_one_member(settings: Settings, gateways: list[NatsGateway]): + members = [await _connected_gateway(settings), await _connected_gateway(settings)] + publisher = await _connected_gateway(settings) + gateways += [*members, publisher] + run_id = str(uuid.uuid4()) + deliveries: list[tuple[int, str]] = [] + + def collector(member: int) -> Any: + async def on_request(msg: Msg) -> None: + body = json.loads(msg.data) + if body.get("run_id") == run_id: + assert msg.headers is not None + deliveries.append((member, msg.headers[NATS_MSG_ID_HEADER])) + + return on_request + + for index, member in enumerate(members): + await member.subscribe_requests(collector(index), queue=REQUEST_QUEUE_GROUP) + + total = 40 + sent = {f"{run_id}-{n}:1" for n in range(total)} + for msg_id in sorted(sent): + await publisher.publish(IMAGE_REQUEST_SUBJECT, json.dumps({"run_id": run_id}).encode(), msg_id) + await _wait_until(lambda: len(deliveries) >= total) + await asyncio.sleep(0.3) # лишние (дублирующие) доставки успели бы прийти + + delivered_ids = [msg_id for _, msg_id in deliveries] + assert len(deliveries) == total + assert set(delivered_ids) == sent # каждое сообщение — ровно одному участнику группы, без дублей + assert {member for member, _ in deliveries} == {0, 1} # сообщения распределены между обоими участниками + + +async def test_publish_result_reaches_result_subject_with_ack_msg_id(settings: Settings, gateways: list[NatsGateway]): + gateway = await _connected_gateway(settings) + gateways.append(gateway) + listener = NATS() + await asyncio.wait_for(listener.connect(settings.OPENBAO.NATS.NATS_URL), CONNECT_WAIT) + job_id, file_id = uuid.uuid4(), uuid.uuid4() + received: list[Msg] = [] + + async def on_result(msg: Msg) -> None: + if json.loads(msg.data).get("job_id") == str(job_id): + received.append(msg) + + try: + await listener.subscribe(RESULT_SUBJECT, cb=on_result) + await listener.flush() + message = build_ack_message(job_id, file_id, ServiceInfo(name="scan-gateway", version="1.0.0")) + await gateway.publish_result(message, attempt=1) + await _wait_until(lambda: len(received) >= 1) + finally: + await listener.drain() + + assert len(received) == 1 + assert received[0].headers is not None + assert received[0].headers[NATS_MSG_ID_HEADER] == f"ack:{job_id}:1" + assert json.loads(received[0].data) == message.to_wire() + + +# --- S3 ----------------------------------------------------------------------------------------------------------- + + +@pytest.fixture +async def storage(settings: Settings) -> AsyncIterator[tuple[S3Storage, str, list[str]]]: + storage = S3Storage.from_settings(settings.OPENBAO.MINIO) + prefix = f"itest/{uuid.uuid4()}" + written: list[str] = [] + yield storage, prefix, written + for key in written: + await storage.client.remove_object(storage.bucket, key) + await storage.close() + + +async def test_s3_put_and_get_roundtrip(storage: tuple[S3Storage, str, list[str]]): + s3, prefix, written = storage + json_key, bytes_key = f"{prefix}/raw.json", f"{prefix}/file.pdf" + written += [json_key, bytes_key] + payload = {"error": None, "message": "Бухгалтерский баланс", "n": 1} + + await s3.put_json(json_key, payload) + await s3.put_bytes(bytes_key, b"%PDF-1.7 \x00\xff", content_type="application/pdf") + + raw = await s3.get_bytes(json_key) + assert "Бухгалтерский баланс".encode() in raw # UTF-8 без экранирования + assert json.loads(raw) == payload + assert await s3.get_bytes(bytes_key, bucket=s3.bucket) == b"%PDF-1.7 \x00\xff" + stat = await s3.client.stat_object(s3.bucket, json_key) + assert stat.content_type == "application/json" + + +async def test_s3_get_missing_key_raises_not_found(storage: tuple[S3Storage, str, list[str]]): + s3, prefix, _ = storage + with pytest.raises(S3ObjectNotFoundError) as exc_info: + await s3.get_bytes(f"{prefix}/missing.pdf") + assert exc_info.value.key == f"{prefix}/missing.pdf" diff --git a/api/tests/unit/test_nats.py b/api/tests/unit/test_nats.py new file mode 100644 index 0000000..c490ed0 --- /dev/null +++ b/api/tests/unit/test_nats.py @@ -0,0 +1,408 @@ +"""Слой NATS (`src.core.nats`): подключение с/без nkeys, колбэки переподключения, подписка queue group, публикация +с `Nats-Msg-Id` и повторами. Клиент nats-py подменяется фейком — сеть не нужна. Переподключение к живому серверу +не проверяется (общий контейнер NATS не перезапускаем): здесь — только колбэки и восстановление подписки клиентом. +""" + +import base64 +import json +import uuid +from typing import Any + +import nkeys +import pytest +from nats import errors as nats_errors + +from src.contract import ( + IMAGE_REQUEST_SUBJECT, + NATS_MSG_ID_HEADER, + RESULT_SUBJECT, + ErrorCode, + ServiceInfo, + build_ack_message, + build_error_message, +) +from src.core.config import Settings, get_settings +from src.core.nats import ( + REQUEST_QUEUE_GROUP, + NatsConfigError, + NatsGateway, + NatsNotConnectedError, + PublishFailed, +) + + +class FakeMsg: + def __init__(self, data: bytes = b"{}", headers: dict[str, str] | None = None) -> None: + self.subject = IMAGE_REQUEST_SUBJECT + self.data = data + self.headers = headers + + +class FakeNatsClient: + """Минимальная подмена `nats.aio.client.Client`: запоминает вызовы, ошибки публикации задаются очередью.""" + + def __init__(self) -> None: + self.connect_args: tuple[Any, ...] = () + self.connect_kwargs: dict[str, Any] = {} + self.is_connected = False + self.is_closed = False + self.published: list[tuple[str, bytes, dict[str, str] | None]] = [] + self.publish_errors: list[BaseException] = [] + self.flush_calls = 0 + self.subscriptions: list[dict[str, Any]] = [] + self.drained = False + self.closed = False + self.jetstream_calls = 0 + + async def connect(self, *args: Any, **kwargs: Any) -> None: + self.connect_args = args + self.connect_kwargs = kwargs + self.is_connected = True + + async def publish(self, subject: str, payload: bytes = b"", headers: dict[str, str] | None = None) -> None: + self.published.append((subject, payload, headers)) + if self.publish_errors: + raise self.publish_errors.pop(0) + + async def flush(self, timeout: float = 10) -> None: + self.flush_calls += 1 + + async def subscribe(self, subject: str, queue: str = "", cb: Any = None) -> str: + self.subscriptions.append({"subject": subject, "queue": queue, "cb": cb}) + return f"sub-{len(self.subscriptions)}" + + async def drain(self) -> None: + self.drained = True + self.is_connected = False + self.is_closed = True + + async def close(self) -> None: + self.closed = True + self.is_connected = False + self.is_closed = True + + def jetstream(self, **opts: Any) -> None: # pragma: no cover - вызов и есть ошибка + self.jetstream_calls += 1 + raise AssertionError("JetStream не используется (D-8)") + + +def make_settings(monkeypatch: pytest.MonkeyPatch, **env: str) -> Settings: + for name, value in env.items(): + monkeypatch.setenv(name, value) + get_settings.cache_clear() + return get_settings() + + +@pytest.fixture +def settings(monkeypatch: pytest.MonkeyPatch) -> Settings: + return make_settings(monkeypatch) + + +@pytest.fixture +def fake() -> FakeNatsClient: + return FakeNatsClient() + + +@pytest.fixture +def sleeps() -> list[float]: + return [] + + +@pytest.fixture +def gateway(settings: Settings, fake: FakeNatsClient, sleeps: list[float]) -> NatsGateway: + async def fake_sleep(delay: float) -> None: + sleeps.append(delay) + + return NatsGateway(settings, client_factory=lambda: fake, retry_delay=0, sleep=fake_sleep) + + +# --- подключение ------------------------------------------------------------------------------------------------ + + +async def test_connect_without_credentials_when_seed_and_jwt_empty(gateway: NatsGateway, fake: FakeNatsClient): + assert await gateway.connect() is True + + assert fake.connect_args == ("nats://127.0.0.1:4222",) + kwargs = fake.connect_kwargs + assert kwargs["max_reconnect_attempts"] == -1 + assert kwargs["name"] == "scan-gateway" + for callback in ("error_cb", "disconnected_cb", "reconnected_cb", "closed_cb"): + assert callable(kwargs[callback]) + for credential in ("user_jwt_cb", "signature_cb", "nkeys_seed", "nkeys_seed_str", "user_credentials", "token"): + assert credential not in kwargs + assert fake.jetstream_calls == 0 + assert gateway.is_connected is True + + +async def test_connect_with_nkeys_and_jwt_when_both_set(monkeypatch: pytest.MonkeyPatch, fake: FakeNatsClient): + # Фиктивная пара ключей создаётся в тесте: настоящих секретов в репозитории нет + keypair = nkeys.from_seed(nkeys.encode_seed(b"\x01" * 32, nkeys.PREFIX_BYTE_USER)) + seed = keypair.seed.decode() + settings = make_settings(monkeypatch, OPENBAO__NATS__NATS_SEED=seed, OPENBAO__NATS__NATS_JWT="dummy.jwt.token") + gateway = NatsGateway(settings, client_factory=lambda: fake) + + await gateway.connect() + + kwargs = fake.connect_kwargs + assert kwargs["user_jwt_cb"]() == b"dummy.jwt.token" + nonce = "nonce-123" + signature = base64.b64decode(kwargs["signature_cb"](nonce)) + assert keypair.verify(nonce.encode(), signature) is True + assert fake.jetstream_calls == 0 + + +@pytest.mark.parametrize( + "env", + [ + {"OPENBAO__NATS__NATS_SEED": "SUAxxx"}, + {"OPENBAO__NATS__NATS_JWT": "dummy.jwt.token"}, + ], +) +async def test_connect_with_only_one_credential_is_config_error( + monkeypatch: pytest.MonkeyPatch, fake: FakeNatsClient, env: dict[str, str] +): + settings = make_settings(monkeypatch, **env) + gateway = NatsGateway(settings, client_factory=lambda: fake) + + with pytest.raises(NatsConfigError, match="NATS_SEED и NATS_JWT"): + await gateway.connect() + assert fake.connect_kwargs == {} + + +async def test_connect_with_unreadable_seed_is_config_error(monkeypatch: pytest.MonkeyPatch, fake: FakeNatsClient): + settings = make_settings(monkeypatch, OPENBAO__NATS__NATS_SEED="не-seed", OPENBAO__NATS__NATS_JWT="dummy.jwt") + gateway = NatsGateway(settings, client_factory=lambda: fake) + + with pytest.raises(NatsConfigError, match="NATS_SEED"): + await gateway.connect() + + +async def test_nats_disabled_does_not_connect(monkeypatch: pytest.MonkeyPatch, fake: FakeNatsClient): + settings = make_settings(monkeypatch, NATS_ENABLED="false") + created: list[FakeNatsClient] = [] + + def factory() -> FakeNatsClient: + created.append(fake) + return fake + + gateway = NatsGateway(settings, client_factory=factory) + + assert await gateway.connect() is False + assert created == [] + assert gateway.is_connected is False + + +async def test_is_connected_reflects_client_state(gateway: NatsGateway, fake: FakeNatsClient): + assert gateway.is_connected is False # до connect клиента нет + await gateway.connect() + assert gateway.is_connected is True + fake.is_connected = False # nats-py переподключается + assert gateway.is_connected is False + + +async def test_reconnect_callbacks_log_and_keep_state(gateway: NatsGateway, fake: FakeNatsClient): + await gateway.connect() + kwargs = fake.connect_kwargs + + fake.is_connected = False + await kwargs["disconnected_cb"]() + await kwargs["error_cb"](nats_errors.StaleConnectionError()) + assert gateway.is_connected is False + + fake.is_connected = True + await kwargs["reconnected_cb"]() + assert gateway.is_connected is True + await kwargs["closed_cb"]() + + +async def test_close_drains_connected_client(gateway: NatsGateway, fake: FakeNatsClient): + await gateway.connect() + await gateway.close() + assert fake.drained is True + assert gateway.is_connected is False + + +async def test_close_without_connection_is_noop(gateway: NatsGateway, fake: FakeNatsClient): + await gateway.close() + assert fake.drained is False + assert fake.closed is False + + +async def test_close_falls_back_to_close_when_drain_fails(gateway: NatsGateway, fake: FakeNatsClient): + await gateway.connect() + + async def failing_drain() -> None: + raise nats_errors.ConnectionReconnectingError() + + fake.drain = failing_drain # type: ignore[method-assign] + await gateway.close() + assert fake.closed is True + + +# --- подписка --------------------------------------------------------------------------------------------------- + + +async def test_subscribe_requests_uses_image_subject_and_queue_group(gateway: NatsGateway, fake: FakeNatsClient): + received: list[FakeMsg] = [] + + async def on_request(msg: Any) -> None: + received.append(msg) + + await gateway.connect() + await gateway.subscribe_requests(on_request) + + assert REQUEST_QUEUE_GROUP == "scan-gateway" + [subscription] = fake.subscriptions + assert subscription["subject"] == IMAGE_REQUEST_SUBJECT == "recognition.image.request" + assert subscription["queue"] == "scan-gateway" + assert fake.flush_calls == 1 # подписка подтверждена сервером до возврата + + msg = FakeMsg(b'{"job_id": "x"}', {NATS_MSG_ID_HEADER: "x:1"}) + await subscription["cb"](msg) + assert received == [msg] # обработчик получает исходное сообщение nats-py (data, headers) + + +async def test_subscribe_requests_handler_exception_is_logged_not_raised(gateway: NatsGateway, fake: FakeNatsClient): + async def broken(msg: Any) -> None: + raise RuntimeError("сбой обработчика") + + await gateway.connect() + await gateway.subscribe_requests(broken) + await fake.subscriptions[0]["cb"](FakeMsg()) # не бросает: подписка живёт + + +async def test_subscribe_requests_requires_connection(gateway: NatsGateway): + async def on_request(msg: Any) -> None: + return None + + with pytest.raises(NatsNotConnectedError): + await gateway.subscribe_requests(on_request) + + +# --- публикация ------------------------------------------------------------------------------------------------- + + +async def test_publish_sets_msg_id_header_and_flushes(gateway: NatsGateway, fake: FakeNatsClient): + await gateway.connect() + await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1") + + assert fake.published == [(RESULT_SUBJECT, b"payload", {"Nats-Msg-Id": "result:job:1"})] + assert fake.flush_calls == 1 + + +async def test_publish_retries_transient_errors_until_success( + gateway: NatsGateway, fake: FakeNatsClient, sleeps: list[float] +): + await gateway.connect() + fake.publish_errors = [nats_errors.ConnectionReconnectingError(), nats_errors.FlushTimeoutError()] + + await gateway.publish(RESULT_SUBJECT, b"payload", "ack:job:1") + + assert len(fake.published) == 3 # PUBLISH_RETRIES=3 — три попытки всего + assert all(headers == {"Nats-Msg-Id": "ack:job:1"} for _, _, headers in fake.published) + assert sleeps == [0, 0] # пауза между попытками, после последней — нет + + +@pytest.mark.parametrize( + "error", + [ + nats_errors.ConnectionClosedError(), + nats_errors.ConnectionReconnectingError(), + nats_errors.OutboundBufferLimitError(), + nats_errors.StaleConnectionError(), + nats_errors.TimeoutError(), + TimeoutError(), + ConnectionResetError(), + ], +) +async def test_publish_raises_publish_failed_after_exactly_retries_attempts( + gateway: NatsGateway, fake: FakeNatsClient, error: BaseException +): + await gateway.connect() + fake.publish_errors = [error] * 10 + + with pytest.raises(PublishFailed) as exc_info: + await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1") + + assert len(fake.published) == 3 + assert exc_info.value.attempts == 3 + assert exc_info.value.msg_id == "result:job:1" + assert exc_info.value.__cause__ is error + + +async def test_publish_retries_parameter_overrides_settings(gateway: NatsGateway, fake: FakeNatsClient): + await gateway.connect() + fake.publish_errors = [nats_errors.ConnectionClosedError()] * 10 + + with pytest.raises(PublishFailed): + await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1", retries=5) + assert len(fake.published) == 5 + + +async def test_publish_uses_publish_retries_from_settings(monkeypatch: pytest.MonkeyPatch, fake: FakeNatsClient): + settings = make_settings(monkeypatch, PUBLISH_RETRIES="2") + gateway = NatsGateway(settings, client_factory=lambda: fake, retry_delay=0) + await gateway.connect() + fake.publish_errors = [nats_errors.ConnectionClosedError()] * 10 + + with pytest.raises(PublishFailed): + await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1") + assert len(fake.published) == 2 + + +@pytest.mark.parametrize("error", [nats_errors.MaxPayloadError(), nats_errors.BadSubjectError(), ValueError("x")]) +async def test_publish_does_not_retry_non_transient_errors( + gateway: NatsGateway, fake: FakeNatsClient, error: BaseException +): + await gateway.connect() + fake.publish_errors = [error] * 10 + + with pytest.raises(PublishFailed) as exc_info: + await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1") + assert len(fake.published) == 1 + assert exc_info.value.attempts == 1 + assert exc_info.value.__cause__ is error + + +async def test_publish_without_connection_fails_without_attempts(gateway: NatsGateway, fake: FakeNatsClient): + with pytest.raises(PublishFailed) as exc_info: + await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1") + assert exc_info.value.attempts == 0 + assert fake.published == [] + + +async def test_publish_rejects_non_positive_retries(gateway: NatsGateway): + await gateway.connect() + with pytest.raises(ValueError, match="retries"): + await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1", retries=0) + + +# --- публикация конверта ---------------------------------------------------------------------------------------- + +SERVICE = ServiceInfo(name="scan-gateway", version="1.0.0") + + +async def test_publish_result_ack_uses_ack_msg_id(gateway: NatsGateway, fake: FakeNatsClient): + job_id, file_id = uuid.uuid4(), uuid.uuid4() + message = build_ack_message(job_id, file_id, SERVICE) + await gateway.connect() + + await gateway.publish_result(message, attempt=2) + + [(subject, payload, headers)] = fake.published + assert subject == RESULT_SUBJECT + assert headers == {"Nats-Msg-Id": f"ack:{job_id}:2"} + assert json.loads(payload) == message.to_wire() + + +async def test_publish_result_final_uses_result_msg_id_and_utf8(gateway: NatsGateway, fake: FakeNatsClient): + job_id, file_id = uuid.uuid4(), uuid.uuid4() + message = build_error_message(job_id, file_id, SERVICE, ErrorCode.TIMEOUT, "ИС не ответила") + await gateway.connect() + + await gateway.publish_result(message, attempt=1) + + [(_, payload, headers)] = fake.published + assert headers == {"Nats-Msg-Id": f"result:{job_id}:1"} + assert "ИС не ответила".encode() in payload # UTF-8 без \u-экранирования + assert json.loads(payload) == message.to_wire() diff --git a/docs/superpowers/plans/2026-10-03-slice-0-skeleton.md b/docs/superpowers/plans/2026-10-03-slice-0-skeleton.md index 1888cbc..2e9f4cc 100644 --- a/docs/superpowers/plans/2026-10-03-slice-0-skeleton.md +++ b/docs/superpowers/plans/2026-10-03-slice-0-skeleton.md @@ -60,7 +60,7 @@ ### Task 11: [5.2] `src/core/clients.py` (`get_bytes`, `put_json`, `put_bytes`), `src/core/nats.py` -- [ ] `src/core/clients.py` (`get_bytes`, `put_json`, `put_bytes`), `src/core/nats.py` (подключение с/без nkeys, автопереподключение, `subscribe_requests(queue="scan-gateway", cb)`, `publish` с `Nats-Msg-Id` и повторами); проверка: unit с фейками, интеграционный тест подключения к compose-NATS без JetStream +- [x] `src/core/clients.py` (`get_bytes`, `put_json`, `put_bytes`), `src/core/nats.py` (подключение с/без nkeys, автопереподключение, `subscribe_requests(queue="scan-gateway", cb)`, `publish` с `Nats-Msg-Id` и повторами); проверка: unit с фейками, интеграционный тест подключения к compose-NATS без JetStream ### Task 12: [5.3] `src/core/worker.py` diff --git a/openspec/changes/slice-0-skeleton/tasks.md b/openspec/changes/slice-0-skeleton/tasks.md index 3ee86f0..1500bfa 100644 --- a/openspec/changes/slice-0-skeleton/tasks.md +++ b/openspec/changes/slice-0-skeleton/tasks.md @@ -47,7 +47,7 @@ - [x] 5.1 `src/normalizer/__init__.py`: протокол `Normalizer`, `NormalizeOutcome`, `NullNormalizer` (ошибка ИС → код из ответа; успех → `NOT_IMPLEMENTED`; не JSON → `INVALID_FORMAT`); unit-тесты на `aep_5.json`, `aep_8_form4.json`, синтетическом ответе с `error`; проверка: зелёный -- [ ] 5.2 `src/core/clients.py` (`get_bytes`, `put_json`, `put_bytes`), `src/core/nats.py` (подключение с/без nkeys, +- [x] 5.2 `src/core/clients.py` (`get_bytes`, `put_json`, `put_bytes`), `src/core/nats.py` (подключение с/без nkeys, автопереподключение, `subscribe_requests(queue="scan-gateway", cb)`, `publish` с `Nats-Msg-Id` и повторами); проверка: unit с фейками, интеграционный тест подключения к compose-NATS без JetStream - [ ] 5.3 `src/core/worker.py`: `on_request(msg)`: валидация → LRU `job_id` → ack `processing` → очередь; семафор