Срез 0 [5.2]: слой NATS без JetStream (src/core/nats.py), интеграционные тесты, отметка задачи
NatsGateway: подключение с nkeys/JWT или без учётных данных, NATS_ENABLED=false — без подключения, автопереподключение nats-py, subscribe_requests в queue group scan-gateway, publish с Nats-Msg-Id и повторами (PUBLISH_RETRIES попыток всего) -> PublishFailed, publish_result для конвертов ack/result. src.core.clients и src.core.nats — в strict-списке mypy. Интеграционные тесты против compose-NATS/MinIO: queue group, заголовок Nats-Msg-Id, roundtrip S3, отсутствующий ключ. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
c6dfb3e22c
commit
5d82607715
@ -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
|
||||
|
||||
309
api/src/core/nats.py
Normal file
309
api/src/core/nats.py
Normal file
@ -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:<job_id>:<attempt>` для `processing`, иначе `result:<job_id>:<attempt>`."""
|
||||
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: <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 закрыто")
|
||||
213
api/tests/integration/test_nats_minio_integration.py
Normal file
213
api/tests/integration/test_nats_minio_integration.py
Normal file
@ -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/<uuid>/` и удаляются.
|
||||
"""
|
||||
|
||||
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"
|
||||
408
api/tests/unit/test_nats.py
Normal file
408
api/tests/unit/test_nats.py
Normal file
@ -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()
|
||||
@ -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`
|
||||
|
||||
|
||||
@ -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` → очередь; семафор
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user