From 4e5e1a7b5be2d60823215bc9c96b689d599f0016 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sat, 3 Oct 2026 23:47:39 +0400 Subject: [PATCH] =?UTF-8?q?feat:=20src/core/nats.py=20=E2=80=94=20=D1=84?= =?UTF-8?q?=D0=BE=D0=BD=D0=BE=D0=B2=D0=BE=D0=B5=20=D0=BF=D0=BE=D0=B4=D0=BA?= =?UTF-8?q?=D0=BB=D1=8E=D1=87=D0=B5=D0=BD=D0=B8=D0=B5=20=D0=BA=20core=20NA?= =?UTF-8?q?TS,=20publish=20=D1=81=20Nats-Msg-Id,=20=D0=BF=D0=BE=D0=B4?= =?UTF-8?q?=D0=BF=D0=B8=D1=81=D0=BA=D0=B0=20recognition.result=20=D0=B2=20?= =?UTF-8?q?queue=20group=20arm-core?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - NatsConnection: supervisor первого подключения в фоне (старт приложения не ждёт NATS), nats-py переподключается сам (max_reconnect_attempts=-1, reconnect_time_wait=2) и восстанавливает подписки; экспоненциальный backoff 1→30 с, если connect завершился исключением; логирование ошибок без шума (первая ошибка серии — warning, дальше не чаще раза в минуту). - publish(subject, payload, msg_id): заголовок Nats-Msg-Id + flush; нет соединения или нет подтверждения — NatsPublishError. subscribe_results(handler): queue group arm-core, исключения обработчика логируются. - build_nats_auth перенесён из app.py; ошибка учётных данных — NatsConfigError, категория config_error, без повторов (T4.1). Первое подключение повторяется (T4.2): прежний await connect в lifespan при недоступном NATS висел бесконечно (nats-py с max_reconnect_attempts=-1 повторяет первое подключение сам). - app.py: подключение переехало в core/nats.py (R-3), lifespan запускает/останавливает NatsConnection, подписка — заглушка services/recognition_exchange.handle_message (полная обработка — задача 24). - Тесты: юнит (поддельный клиент, инжектируемый sleep, отсутствие JetStream в src/) и интеграционные против NATS без JetStream (заголовок, queue group без дублей, недоступный сервер). Co-Authored-By: Claude Fable 5.1 --- README.md | 9 +- api/src/app.py | 87 ++---- api/src/core/nats.py | 310 ++++++++++++++++++ api/src/services/recognition_exchange.py | 14 + api/tests/test_app.py | 115 ++++--- api/tests/test_nats_core.py | 379 +++++++++++++++++++++++ api/tests/test_nats_integration.py | 169 ++++++++++ 7 files changed, 968 insertions(+), 115 deletions(-) create mode 100644 api/src/core/nats.py create mode 100644 api/src/services/recognition_exchange.py create mode 100644 api/tests/test_nats_core.py create mode 100644 api/tests/test_nats_integration.py diff --git a/README.md b/README.md index 0f36f1a..593359b 100644 --- a/README.md +++ b/README.md @@ -165,8 +165,10 @@ Dev-пользователи (`api/seed/dev/users.yaml`, создаются `mak слушает `http`. S3 в Срезе 0 не используется (`app.state.storage` остаётся пустым). Это до исправления в задаче 22 (отдельная настройка схемы подключения к MinIO). -Недоступность SMB, S3 или NATS при старте приложение не роняет: ошибка логируется, состояние NATS видно в -`/healthcheck`. +Недоступность SMB, S3 или NATS при старте приложение не роняет и не задерживает: ошибка логируется, состояние NATS +видно в `/healthcheck`. К NATS ядро подключается в фоне и повторяет попытки само (`nats=down`, пока связи нет; warning +в логе — на первую ошибку и дальше не чаще раза в минуту); после разрыва nats-py переподключается и восстанавливает +подписку. Предупреждение «Включён dev-эмитент токенов … только для разработки» появляется не при старте, а при первом запросе к защищённому эндпоинту (`/api/v1/me`): проверяющий токены создаётся лениво (штатно, см. «Безопасность @@ -292,7 +294,8 @@ api/ app.py FastAPI-приложение: lifespan (S3, NATS, SMB), /healthcheck, раздача SPA api/endpoints/ эндпоинты (me.py, dev_auth.py); router.py собирает /api/v1 api/dependencies/ зависимости FastAPI (current_user, проверка прав) - core/ config, auth (JWT), clients (S3), nats_callback, logger, exception_handlers + core/ config, auth (JWT), clients (S3), nats (подключение, publish, подписка), nats_callback, + logger, exception_handlers db/ движок и сессия (схема dkb) models/ ORM-модели, 1:1 с DBML repositories/ доступ к данным diff --git a/api/src/app.py b/api/src/app.py index 370099a..2a39269 100644 --- a/api/src/app.py +++ b/api/src/app.py @@ -1,21 +1,18 @@ """Точка входа FastAPI. Каркас из шаблона команды: инфраструктура без доменного кода. -Что здесь есть: настройки, подключение S3 (MinIO), NATS (nkeys JWT; пустые NATS_SEED/NATS_JWT — без учётных данных; -подключение при NATS_ENABLED=true независимо от DEBUG), SMB (горячие папки ИС Сканирование), middleware (gzip, proxy -headers, CORS, РАИСА при DEBUG=False), healthcheck (uptime, version, nats: ok|down|off), раздача собранного SPA. -Недоступность S3 или NATS при старте не роняет приложение: ошибка логируется, healthcheck показывает состояние. -Что добавляет Срез 0: роутеры в src/api/router.py, outbox-паблишер и consumer recognition.result -(docs/context/03-architecture.md), зависимость require(permission). +Что здесь есть: настройки, подключение S3 (MinIO), NATS (`src/core/nats.py`: core +NATS без JetStream, подключение в фоне при NATS_ENABLED=true независимо от DEBUG, подписка `recognition.result` в queue +group `arm-core`), SMB (горячие папки ИС Сканирование), middleware (gzip, proxy headers, CORS, РАИСА при DEBUG=False), +healthcheck (uptime, version, nats: ok|down|off), раздача собранного SPA. +Недоступность S3 или NATS при старте не роняет и не задерживает приложение: ошибка логируется, healthcheck показывает +состояние. Outbox-паблишер и вахтёр сроков — задача 23 (docs/context/03-architecture.md). """ -import base64 import os from contextlib import asynccontextmanager from datetime import UTC, datetime from http import HTTPStatus -from typing import Any -import nkeys import smbclient from fastapi import FastAPI, Request from fastapi.exceptions import HTTPException, StarletteHTTPException @@ -23,14 +20,15 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from loguru import logger from miniopy_async import Minio -from nats.aio.client import Client as NATS from starlette.middleware.gzip import GZipMiddleware from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware from src.api.router import API_PREFIX, router from src.core.clients import S3Storage -from src.core.config import NATSSettings, get_settings +from src.core.config import get_settings from src.core.exception_handlers import register_exception_handlers +from src.core.nats import NatsConnection +from src.services.recognition_exchange import handle_message settings = get_settings() @@ -38,34 +36,6 @@ VERSION = settings.VERSION UP_TIME = datetime.now(UTC) -# --- NATS: аутентификация по nkeys/JWT --------------------------------------------------------- -def _build_nats_auth(nats_settings: NATSSettings) -> dict[str, Any]: - """Параметры аутентификации для ``NATS.connect``. - - Пустые NATS_SEED и NATS_JWT — подключение без учётных данных (docker compose, тесты): ``{}``. Заданы оба — - колбэки nkeys/JWT, как в шаблоне. Задан только один — ошибка конфигурации. Подключение переедет в core/nats.py - (задача 22). - """ - seed = nats_settings.NATS_SEED - jwt = nats_settings.NATS_JWT - if not seed and not jwt: - return {} - if not seed or not jwt: - raise RuntimeError("Неполные учётные данные NATS: NATS_SEED и NATS_JWT задаются вместе или не задаются вовсе") - - keypair = nkeys.from_seed(seed.encode("utf-8")) - - def user_jwt_cb() -> bytes: - return jwt.encode("utf-8") - - def signature_cb(nonce: str | bytes) -> bytes: - if isinstance(nonce, str): - nonce = nonce.encode("utf-8") - return base64.b64encode(keypair.sign(nonce)) - - return {"user_jwt_cb": user_jwt_cb, "signature_cb": signature_cb} - - # --- РАИСА: защищённый API (только вне DEBUG) --------------------------------------------------- protected_middleware_available = False if not settings.DEBUG: @@ -118,33 +88,21 @@ async def _connect_s3() -> S3Storage | None: return S3Storage(client=s3_client, bucket_name=settings.OPENBAO.MINIO.MINIO_BUCKET) -async def _connect_nats() -> NATS | None: - """Подключение к NATS при NATS_ENABLED=true; при ошибке — ``None`` (healthcheck покажет ``down``).""" +async def _subscribe_results(nats: NatsConnection) -> None: + """Подписки после первого подключения к NATS: `recognition.result` в queue group `arm-core`.""" + await nats.subscribe_results(handle_message) + + +def _create_nats() -> NatsConnection | None: + """Соединение NATS при NATS_ENABLED=true; при false — ``None`` (healthcheck покажет ``off``).""" if not settings.NATS_ENABLED: logger.info("NATS выключен (NATS_ENABLED=false)") return None - nc = NATS() - try: - await nc.connect( - settings.OPENBAO.NATS.NATS_URL, - max_reconnect_attempts=-1, - reconnect_time_wait=2, - **_build_nats_auth(settings.OPENBAO.NATS), - ) - except Exception as exc: - logger.exception( - "Не удалось подключиться к NATS. error_category=nats_error, error_type={error_type}, error={error}", - error_type=type(exc).__name__, - error=repr(exc), - ) - return None - logger.info("Подключение к NATS произошло") - return nc + return NatsConnection(settings.OPENBAO.NATS, on_connected=_subscribe_results) @asynccontextmanager async def lifespan(app: FastAPI): - app.state.nats_subs = [] app.state.storage = None app.state.nats = None @@ -157,18 +115,17 @@ async def lifespan(app: FastAPI): try: app.state.storage = await _connect_s3() - app.state.nats = await _connect_nats() - # Подписки (recognition.result) и outbox-паблишер — Срез 0, см. docs/context/03-architecture.md + app.state.nats = _create_nats() + if app.state.nats is not None: + await app.state.nats.start() # подключение в фоне: старт приложения не ждёт NATS logger.info("Приложение запущено") yield finally: logger.info("Приложение закрывается") smbclient.reset_connection_cache() - for sub in app.state.nats_subs: - await sub.unsubscribe() - if app.state.nats: - await app.state.nats.close() + if app.state.nats is not None: + await app.state.nats.stop() if app.state.storage: await app.state.storage.client.close_session() diff --git a/api/src/core/nats.py b/api/src/core/nats.py new file mode 100644 index 0000000..a798c0c --- /dev/null +++ b/api/src/core/nats.py @@ -0,0 +1,310 @@ +"""Соединение с NATS: core NATS без JetStream (design D-6; spec recognition-job-exchange, «Соединение с NATS по +шаблону, без JetStream»; docs/context/03-architecture.md, «Контракт обмена через NATS»). + +- Аутентификация как в шаблоне: заданы `NATS_SEED` и `NATS_JWT` — nkeys/JWT; оба пусты — без учётных данных. +- `publish(subject, payload, msg_id)` — публикация с заголовком `Nats-Msg-Id` и `flush`: ошибка соединения проявляется + сразу исключением `NatsPublishError` (outbox оставляет задание в `queued`). +- `subscribe_results(handler)` — подписка на `recognition.result` в queue group `arm-core`: из нескольких инстансов ядра + сообщение получает ровно один. Исключения обработчика логируются, подписка не падает. +- Stream'ов и consumer'ов JetStream нет. Переподключение после разрыва — встроенное в nats-py + (`max_reconnect_attempts=-1`, `reconnect_time_wait=2`), подписки nats-py восстанавливает сам. +- Первое подключение выполняет фоновый supervisor (`start()`): старт приложения не ждёт NATS. nats-py 2.14 с + `max_reconnect_attempts=-1` сам повторяет и первое подключение (раз в `reconnect_time_wait`), поэтому `connect` + возвращается только при успехе; если он всё же завершился исключением, supervisor повторяет попытку с + экспоненциальной паузой 1 → 2 → 4 → … ≤ 30 с. Ошибки подключения логируются без шума: первая ошибка серии — warning + с причиной, дальше не чаще раза в минуту; серия сбрасывается при (пере)подключении. +""" + +import asyncio +import base64 +import time +from collections.abc import Awaitable, Callable +from contextlib import suppress +from typing import Any + +import nkeys +from loguru import logger +from nats.aio.client import Client as NATS +from nats.aio.msg import Msg + +from src.contract.subjects import NATS_MSG_ID_HEADER, RESULT_SUBJECT +from src.core.config import NATSSettings + +RESULT_QUEUE_GROUP = "arm-core" +CLIENT_NAME = "arm-core" +MAX_RECONNECT_ATTEMPTS = -1 # переподключаться бесконечно +RECONNECT_TIME_WAIT = 2 # секунд между попытками переподключения nats-py +CONNECT_TIMEOUT = 2 # секунд на одну попытку TCP-подключения +FLUSH_TIMEOUT = 5 # секунд на подтверждение публикации сервером (PING/PONG) +BACKOFF_INITIAL = 1.0 +BACKOFF_MAX = 30.0 +ERROR_LOG_INTERVAL = 60.0 # не чаще раза в минуту повторяем warning об ошибках одной серии + +MessageHandler = Callable[[Msg], Awaitable[None]] +SleepFn = Callable[[float], Awaitable[None]] + + +class NatsError(Exception): + """Базовая ошибка слоя NATS ядра.""" + + +class NatsConfigError(NatsError, RuntimeError): + """Ошибка конфигурации учётных данных NATS: повтор подключения бесполезен.""" + + +class NatsNotConnectedError(NatsError): + """Операция требует установленного соединения с NATS.""" + + +class NatsPublishError(NatsError): + """Публикация не подтверждена: соединения нет или сервер не ответил на flush.""" + + +def build_nats_auth(settings: NATSSettings) -> dict[str, Any]: + """Параметры аутентификации для `Client.connect`. + + Пустые `NATS_SEED` и `NATS_JWT` — подключение без учётных данных (docker compose, тесты): `{}`. Заданы оба — + колбэки nkeys/JWT, как в шаблоне. Задан только один или seed не читается — `NatsConfigError`. + """ + seed = settings.NATS_SEED + jwt = settings.NATS_JWT + 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: + if isinstance(nonce, str): + nonce = nonce.encode("utf-8") + return base64.b64encode(keypair.sign(nonce)) + + return {"user_jwt_cb": user_jwt_cb, "signature_cb": signature_cb} + + +class NatsConnection: + """Соединение ядра с NATS: фоновое первое подключение, публикация с `Nats-Msg-Id`, подписка на результаты. + + `on_connected(conn)` вызывается один раз после первого успешного подключения (подписки); при исключении + повторяется с той же экспоненциальной паузой. `result_subject` и `queue` параметризуются для тестов — + по умолчанию `recognition.result` и `arm-core`. + """ + + def __init__( # noqa: PLR0913 - все параметры, кроме settings, именованные: подмены для тестов и тайминги + self, + settings: NATSSettings, + *, + on_connected: Callable[["NatsConnection"], Awaitable[None]] | None = None, + result_subject: str = RESULT_SUBJECT, + queue: str = RESULT_QUEUE_GROUP, + client_factory: Callable[[], Any] = NATS, + sleep: SleepFn = asyncio.sleep, + backoff_initial: float = BACKOFF_INITIAL, + backoff_max: float = BACKOFF_MAX, + error_log_interval: float = ERROR_LOG_INTERVAL, + flush_timeout: int = FLUSH_TIMEOUT, + ) -> None: + self._settings = settings + self._on_connected = on_connected + self.result_subject = result_subject + self.queue = queue + self._client_factory = client_factory + self._sleep = sleep + self._backoff_initial = backoff_initial + self._backoff_max = backoff_max + self._error_log_interval = error_log_interval + self._flush_timeout = flush_timeout + self._client: Any = None + self._supervisor: asyncio.Task[None] | None = None + self._ready = asyncio.Event() + self._stopping = False + # Серия ошибок подключения: сколько подавлено и когда последний раз писали warning + self._error_streak = 0 + self._last_error_log = 0.0 + + # --- состояние ------------------------------------------------------------------------------ + @property + def is_connected(self) -> bool: + """Фактическое состояние соединения (False до первого подключения, во время переподключения и после stop).""" + return self._client is not None and bool(self._client.is_connected) + + async def wait_ready(self, timeout: float) -> bool: + """Ждёт первого подключения и выполнения `on_connected`; `False`, если не дождались за `timeout` секунд.""" + try: + await asyncio.wait_for(self._ready.wait(), timeout) + except TimeoutError: + return False + return True + + # --- жизненный цикл ------------------------------------------------------------------------- + async def start(self) -> None: + """Запускает фоновый supervisor первого подключения; не ждёт NATS.""" + if self._supervisor is None: + self._supervisor = asyncio.create_task(self._supervise(), name="nats-supervisor") + + async def stop(self) -> None: + """Останавливает supervisor и закрывает соединение; исключения наружу не выходят.""" + self._stopping = True + if self._supervisor is not None: + self._supervisor.cancel() + with suppress(asyncio.CancelledError, Exception): + await self._supervisor + self._supervisor = None + await self.close() + + async def close(self) -> None: + """Подключённый клиент — `drain` (дообработать полученное), иначе `close`; без исключений наружу.""" + client = self._client + if client is None or client.is_closed: + return + try: + if client.is_connected: + await client.drain() + else: + await client.close() + except Exception as exc: + logger.warning( + "Ошибка при закрытии соединения NATS. error_category=nats_error, error_type={error_type}, " + "error={error}", + error_type=type(exc).__name__, + error=repr(exc), + ) + with suppress(Exception): + await client.close() + + async def connect(self) -> None: + """Одна попытка подключения новым клиентом. nats-py с `max_reconnect_attempts=-1` повторяет её сам.""" + auth = build_nats_auth(self._settings) + client = self._client_factory() + self._client = client + try: + await client.connect( + self._settings.NATS_URL, + name=CLIENT_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 + + async def _supervise(self) -> None: + delay = self._backoff_initial + while True: + try: + await self.connect() + break + except NatsConfigError as exc: + logger.error( + "Конфигурация NATS некорректна, подключение не выполняется. error_category=config_error, " + "error={error}", + error=str(exc), + ) + return + except Exception as exc: + self._report_error(exc) + await self._sleep(delay) + delay = min(delay * 2, self._backoff_max) + self._reset_error_streak() + logger.info("Подключение к NATS установлено. url={url}", url=self._settings.NATS_URL) + + delay = self._backoff_initial + while self._on_connected is not None: + try: + await self._on_connected(self) + break + except Exception as exc: + logger.error( + "Не удалось выполнить подписки NATS. error_category=nats_error, error_type={error_type}, " + "error={error}", + error_type=type(exc).__name__, + error=repr(exc), + ) + await self._sleep(delay) + delay = min(delay * 2, self._backoff_max) + self._ready.set() + + # --- публикация и подписка ------------------------------------------------------------------ + async def publish(self, subject: str, payload: bytes, msg_id: str) -> None: + """Публикует сообщение с заголовком `Nats-Msg-Id: ` и ждёт подтверждения сервера (`flush`).""" + if not self.is_connected: + raise NatsPublishError(f"Нет соединения с NATS: сообщение {msg_id} в {subject} не опубликовано") + try: + await self._client.publish(subject, payload, headers={NATS_MSG_ID_HEADER: msg_id}) + await self._client.flush(timeout=self._flush_timeout) + except Exception as exc: + raise NatsPublishError( + f"Публикация в NATS не подтверждена: {subject}, {msg_id} ({type(exc).__name__})" + ) from exc + + async def subscribe_results(self, handler: MessageHandler) -> Any: + """Подписка на `result_subject` в queue group `queue`; исключения `handler` логируются, подписка живёт.""" + if not self.is_connected: + raise NatsNotConnectedError(f"Нет соединения с NATS: подписка на {self.result_subject} невозможна") + + async def safe_handler(msg: Msg) -> None: + try: + await handler(msg) + except Exception as exc: + logger.exception( + "Ошибка обработки сообщения NATS. error_category=handler_error, subject={subject}, " + "error_type={error_type}, error={error}", + subject=msg.subject, + error_type=type(exc).__name__, + error=repr(exc), + ) + + subscription = await self._client.subscribe(self.result_subject, queue=self.queue, cb=safe_handler) + logger.info( + "Подписка NATS активна. subject={subject}, queue={queue}", subject=self.result_subject, queue=self.queue + ) + return subscription + + # --- колбэки nats-py и логирование без шума ------------------------------------------------- + def _report_error(self, exc: BaseException) -> None: + now = time.monotonic() + self._error_streak += 1 + if self._error_streak == 1 or now - self._last_error_log >= self._error_log_interval: + logger.warning( + "Нет связи с NATS, повторяем подключение. error_category=nats_error, url={url}, " + "error_type={error_type}, error={error}, attempts={attempts}", + url=self._settings.NATS_URL, + error_type=type(exc).__name__, + error=repr(exc), + attempts=self._error_streak, + ) + self._last_error_log = now + + def _reset_error_streak(self) -> None: + self._error_streak = 0 + self._last_error_log = 0.0 + + async def _on_error(self, exc: Exception) -> None: + self._report_error(exc) + + async def _on_disconnected(self) -> None: + if self._stopping: + return + logger.warning("Соединение с NATS потеряно, nats-py переподключается") + + async def _on_reconnected(self) -> None: + self._reset_error_streak() + logger.info("Соединение с NATS восстановлено") + + async def _on_closed(self) -> None: + logger.info("Соединение с NATS закрыто") diff --git a/api/src/services/recognition_exchange.py b/api/src/services/recognition_exchange.py new file mode 100644 index 0000000..3ddb740 --- /dev/null +++ b/api/src/services/recognition_exchange.py @@ -0,0 +1,14 @@ +"""Обмен с сервисами распознавания через `recognition_job` (design D-6; docs/context/03-architecture.md). + +Пока здесь только заглушка обработчика `recognition.result`: подписка в `src/core/nats.py` уже вызывает её. +Полноценная обработка (ack `processing`, результат, дедуп по `result_msg_id`) — задача 24 (рулинг R-12, design #23); +`publish_pending()` и `expire_deadlines()` — задача 23. +""" + +from loguru import logger +from nats.aio.msg import Msg + + +async def handle_message(msg: Msg) -> None: + """Заглушка: логирует subject и размер полученного сообщения, ничего не меняет.""" + logger.info("Получено сообщение NATS. subject={subject}, size={size}", subject=msg.subject, size=len(msg.data)) diff --git a/api/tests/test_app.py b/api/tests/test_app.py index 6dc057b..b0d0fda 100644 --- a/api/tests/test_app.py +++ b/api/tests/test_app.py @@ -1,53 +1,24 @@ -"""Быстрые тесты каркаса приложения: аутентификация NATS, /healthcheck, префикс API (без БД и NATS).""" +"""Быстрые тесты каркаса приложения: lifespan, /healthcheck, префикс API (без БД и NATS). +Аутентификация и supervisor NATS — tests/test_nats_core.py (подключение переехало в src/core/nats.py, рулинг R-3). +""" + +import asyncio import importlib -import os from http import HTTPStatus from types import SimpleNamespace +from typing import Any import httpx -import nkeys import pytest from src.api.router import router -from src.core.config import NATSSettings +from src.core.nats import NatsConnection # src/__init__.py реэкспортирует объект app под именем модуля, поэтому модуль берём через importlib app_module = importlib.import_module("src.app") -def _nats_settings(seed: str = "", jwt: str = "") -> NATSSettings: - return NATSSettings(NATS_URL="nats://localhost:4222", NATS_SEED=seed, NATS_JWT=jwt) - - -def _valid_seed() -> str: - return nkeys.encode_seed(os.urandom(32), nkeys.PREFIX_BYTE_USER).decode("utf-8") - - -def test_build_nats_auth_empty_credentials_gives_no_kwargs(): - assert app_module._build_nats_auth(_nats_settings()) == {} - - -@pytest.mark.parametrize( - ("seed", "jwt"), - [("SU-only-seed", ""), ("", "only-jwt")], -) -def test_build_nats_auth_partial_credentials_is_config_error(seed: str, jwt: str): - with pytest.raises(RuntimeError, match="NATS_SEED"): - app_module._build_nats_auth(_nats_settings(seed, jwt)) - - -def test_build_nats_auth_valid_credentials_give_callbacks(): - seed = _valid_seed() - auth = app_module._build_nats_auth(_nats_settings(seed, "header.payload.signature")) - - assert set(auth) == {"user_jwt_cb", "signature_cb"} - assert auth["user_jwt_cb"]() == b"header.payload.signature" - signature = auth["signature_cb"](b"nonce") - assert isinstance(signature, bytes) - assert signature - - @pytest.fixture async def client(): # Без lifespan: app.state.nats не установлен — healthcheck обязан это пережить @@ -97,10 +68,15 @@ def test_router_prefix(): class _FakeNats: def __init__(self) -> None: - self.closed = False + self.started = False + self.stopped = False + self.is_connected = False - async def close(self) -> None: - self.closed = True + async def start(self) -> None: + self.started = True + + async def stop(self) -> None: + self.stopped = True class _FakeS3Client: @@ -123,7 +99,7 @@ async def test_lifespan_starts_when_s3_and_nats_unavailable(monkeypatch: pytest. return None monkeypatch.setattr(app_module, "_connect_s3", unavailable) - monkeypatch.setattr(app_module, "_connect_nats", unavailable) + monkeypatch.setattr(app_module, "_create_nats", lambda: None) reached_yield = False async with app_module.lifespan(app_module.app): @@ -135,7 +111,7 @@ async def test_lifespan_starts_when_s3_and_nats_unavailable(monkeypatch: pytest. @pytest.mark.usefixtures("_no_smb") -async def test_lifespan_closes_created_resources(monkeypatch: pytest.MonkeyPatch): +async def test_lifespan_starts_and_stops_nats_and_closes_s3(monkeypatch: pytest.MonkeyPatch): fake_nats = _FakeNats() s3_client = _FakeS3Client() storage = SimpleNamespace(client=s3_client) @@ -143,17 +119,62 @@ async def test_lifespan_closes_created_resources(monkeypatch: pytest.MonkeyPatch async def connect_s3() -> SimpleNamespace: return storage - async def connect_nats() -> _FakeNats: - return fake_nats - monkeypatch.setattr(app_module, "_connect_s3", connect_s3) - monkeypatch.setattr(app_module, "_connect_nats", connect_nats) + monkeypatch.setattr(app_module, "_create_nats", lambda: fake_nats) async with app_module.lifespan(app_module.app): assert app_module.app.state.storage is storage assert app_module.app.state.nats is fake_nats - assert not fake_nats.closed + assert fake_nats.started + assert not fake_nats.stopped assert not s3_client.session_closed - assert fake_nats.closed + assert fake_nats.stopped assert s3_client.session_closed + + +class _NeverConnectingClient: + """Клиент nats-py, который никогда не подключается: connect висит (как nats-py без сервера).""" + + is_connected = False + is_closed = False + + async def connect(self, servers: str, **options: Any) -> None: + await asyncio.Event().wait() + + async def close(self) -> None: + self.is_closed = True + + +@pytest.mark.usefixtures("_no_smb") +async def test_lifespan_does_not_wait_for_nats_and_healthcheck_is_down(monkeypatch: pytest.MonkeyPatch): + """Регрессия: недоступный NATS не задерживает старт; до первого подключения healthcheck показывает down.""" + + async def unavailable() -> None: + return None + + nats = NatsConnection(app_module.settings.OPENBAO.NATS, client_factory=_NeverConnectingClient) + monkeypatch.setattr(app_module, "_connect_s3", unavailable) + monkeypatch.setattr(app_module, "_create_nats", lambda: nats) + monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True) + + async with app_module.lifespan(app_module.app): + transport = httpx.ASGITransport(app=app_module.app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: + response = await c.get("/healthcheck") + assert response.json()["nats"] == "down" + + assert not nats.is_connected + + +def test_create_nats_disabled_returns_none(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(app_module.settings, "NATS_ENABLED", False) + assert app_module._create_nats() is None + + +def test_create_nats_enabled_returns_connection_for_results(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True) + nats = app_module._create_nats() + assert isinstance(nats, NatsConnection) + assert (nats.result_subject, nats.queue) == ("recognition.result", "arm-core") + assert not nats.is_connected # создание не подключает: подключение — в start() diff --git a/api/tests/test_nats_core.py b/api/tests/test_nats_core.py new file mode 100644 index 0000000..176d242 --- /dev/null +++ b/api/tests/test_nats_core.py @@ -0,0 +1,379 @@ +"""Быстрые тесты `src/core/nats.py` без сервера NATS: аутентификация, supervisor первого подключения, publish. + +Клиент nats-py подменяется фабрикой (`client_factory`), паузы backoff — инжектируемым `sleep` +(spec recognition-job-exchange, «Соединение с NATS по шаблону, без JetStream»; design D-6). +""" + +import ast +import asyncio +import os +from collections.abc import Iterator +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import nkeys +import pytest +from loguru import logger + +from src.contract.subjects import NATS_MSG_ID_HEADER, RESULT_SUBJECT +from src.core.config import NATSSettings +from src.core.nats import ( + RESULT_QUEUE_GROUP, + NatsConfigError, + NatsConnection, + NatsNotConnectedError, + NatsPublishError, + build_nats_auth, +) + +SRC_DIR = Path(__file__).resolve().parent.parent / "src" + + +def _nats_settings(seed: str = "", jwt: str = "") -> NATSSettings: + return NATSSettings(NATS_URL="nats://localhost:4222", NATS_SEED=seed, NATS_JWT=jwt) + + +def _valid_seed() -> str: + return nkeys.encode_seed(os.urandom(32), nkeys.PREFIX_BYTE_USER).decode("utf-8") + + +@pytest.fixture +def log_records() -> Iterator[list[Any]]: + """Записи loguru, выпущенные за время теста.""" + records: list[Any] = [] + handler_id = logger.add(lambda message: records.append(message.record), level="DEBUG") + yield records + logger.remove(handler_id) + + +# --- build_nats_auth ---------------------------------------------------------------------------- +def test_build_nats_auth_empty_credentials_gives_no_kwargs(): + assert build_nats_auth(_nats_settings()) == {} + + +@pytest.mark.parametrize(("seed", "jwt"), [("SU-only-seed", ""), ("", "only-jwt")]) +def test_build_nats_auth_partial_credentials_is_config_error(seed: str, jwt: str): + with pytest.raises(NatsConfigError, match="NATS_SEED"): + build_nats_auth(_nats_settings(seed, jwt)) + + +def test_build_nats_auth_invalid_seed_is_config_error(): + with pytest.raises(NatsConfigError, match="NATS_SEED"): + build_nats_auth(_nats_settings("not-a-seed", "header.payload.signature")) + + +def test_nats_config_error_is_runtime_error(): + assert issubclass(NatsConfigError, RuntimeError) + + +def test_build_nats_auth_valid_credentials_give_callbacks(): + auth = build_nats_auth(_nats_settings(_valid_seed(), "header.payload.signature")) + + assert set(auth) == {"user_jwt_cb", "signature_cb"} + assert auth["user_jwt_cb"]() == b"header.payload.signature" + signature = auth["signature_cb"]("nonce") + assert isinstance(signature, bytes) + assert signature + + +# --- поддельный клиент nats-py ------------------------------------------------------------------ +class _FakeClient: + """Минимальный двойник `nats.aio.client.Client`: подключение, подписка, публикация, закрытие.""" + + def __init__(self, world: "_FakeWorld") -> None: + self.world = world + self.is_connected = False + self.is_closed = False + self.connect_options: dict[str, Any] = {} + self.subscriptions: list[dict[str, Any]] = [] + self.published: list[dict[str, Any]] = [] + self.flush_error: Exception | None = None + self.drained = False + + async def connect(self, servers: str, **options: Any) -> None: + self.world.connect_calls += 1 + self.connect_options = {"servers": servers, **options} + if self.world.connect_calls <= self.world.fail_times: + raise OSError("Connection refused") + self.is_connected = True + + async def subscribe(self, subject: str, queue: str = "", cb: Any = None) -> SimpleNamespace: + self.subscriptions.append({"subject": subject, "queue": queue, "cb": cb}) + return SimpleNamespace(subject=subject, queue=queue) + + async def publish(self, subject: str, payload: bytes = b"", headers: dict[str, str] | None = None) -> None: + self.published.append({"subject": subject, "payload": payload, "headers": headers}) + + async def flush(self, timeout: int = 10) -> None: + if self.flush_error is not None: + raise self.flush_error + + async def drain(self) -> None: + self.drained = True + await self.close() + + async def close(self) -> None: + self.is_connected = False + self.is_closed = True + + +class _FakeWorld: + """Фабрика клиентов: первые `fail_times` подключений падают, затем успех.""" + + def __init__(self, fail_times: int = 0) -> None: + self.fail_times = fail_times + self.connect_calls = 0 + self.clients: list[_FakeClient] = [] + + def __call__(self) -> _FakeClient: + client = _FakeClient(self) + self.clients.append(client) + return client + + +class _RecordingSleep: + """Подмена `asyncio.sleep`: запоминает интервалы backoff и не ждёт.""" + + def __init__(self) -> None: + self.delays: list[float] = [] + + async def __call__(self, delay: float) -> None: + self.delays.append(delay) + await asyncio.sleep(0) + + +async def _handler(msg: Any) -> None: + return None + + +# --- supervisor первого подключения ------------------------------------------------------------- +async def test_supervisor_retries_with_backoff_then_connects_and_subscribes_once(log_records: list[Any]): + world = _FakeWorld(fail_times=7) + sleep = _RecordingSleep() + subscribed: list[NatsConnection] = [] + + async def on_connected(conn: NatsConnection) -> None: + subscribed.append(conn) + await conn.subscribe_results(_handler) + + conn = NatsConnection(_nats_settings(), on_connected=on_connected, client_factory=world, sleep=sleep) + assert not conn.is_connected + + await conn.start() + assert await conn.wait_ready(timeout=2) + + assert conn.is_connected + assert world.connect_calls == 8 # noqa: PLR2004 + assert sleep.delays == [1, 2, 4, 8, 16, 30, 30] + assert subscribed == [conn] + connected_client = world.clients[-1] + assert [(s["subject"], s["queue"]) for s in connected_client.subscriptions] == [ + (RESULT_SUBJECT, RESULT_QUEUE_GROUP) + ] + options = connected_client.connect_options + assert options["servers"] == "nats://localhost:4222" + assert options["max_reconnect_attempts"] == -1 + assert options["reconnect_time_wait"] == 2 # noqa: PLR2004 + for callback in ("error_cb", "disconnected_cb", "reconnected_cb", "closed_cb"): + assert callable(options[callback]) + + # Первая ошибка — warning с причиной, остальные в пределах минуты подавлены + warnings = [r for r in log_records if r["level"].name == "WARNING"] + assert len(warnings) == 1 + assert "error_category=nats_error" in warnings[0]["message"] + assert "Connection refused" in warnings[0]["message"] + + await conn.stop() + assert connected_client.is_closed + + +async def test_supervisor_connects_first_time_without_sleep(): + world = _FakeWorld() + sleep = _RecordingSleep() + conn = NatsConnection(_nats_settings(), client_factory=world, sleep=sleep) + + await conn.start() + assert await conn.wait_ready(timeout=2) + + assert sleep.delays == [] + assert world.connect_calls == 1 + await conn.stop() + + +async def test_start_does_not_block_and_stop_cancels_supervisor(): + world = _FakeWorld(fail_times=10**6) + never = asyncio.Event() + + async def blocking_sleep(delay: float) -> None: + await never.wait() + + conn = NatsConnection(_nats_settings(), client_factory=world, sleep=blocking_sleep) + await conn.start() + await asyncio.sleep(0.01) + + assert not conn.is_connected + assert world.connect_calls == 1 + + await conn.stop() # без исключений наружу + assert not conn.is_connected + assert not await conn.wait_ready(timeout=0.01) + + +async def test_config_error_logged_as_config_error_and_not_retried(log_records: list[Any]): + world = _FakeWorld() + sleep = _RecordingSleep() + conn = NatsConnection(_nats_settings(seed="SU-only-seed"), client_factory=world, sleep=sleep) + + await conn.start() + assert not await conn.wait_ready(timeout=0.05) + + assert world.connect_calls == 0 + assert sleep.delays == [] + assert not conn.is_connected + errors = [r for r in log_records if r["level"].name == "ERROR"] + assert len(errors) == 1 + assert "error_category=config_error" in errors[0]["message"] + assert "nats_error" not in errors[0]["message"] + await conn.stop() + + +async def test_on_connected_failure_is_retried(log_records: list[Any]): + world = _FakeWorld() + sleep = _RecordingSleep() + calls = 0 + + async def flaky_on_connected(conn: NatsConnection) -> None: + nonlocal calls + calls += 1 + if calls < 3: # noqa: PLR2004 + raise RuntimeError("подписка не удалась") + + conn = NatsConnection(_nats_settings(), on_connected=flaky_on_connected, client_factory=world, sleep=sleep) + await conn.start() + assert await conn.wait_ready(timeout=2) + + assert calls == 3 # noqa: PLR2004 + assert sleep.delays == [1, 2] + await conn.stop() + + +async def test_stop_without_start_is_noop(): + conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld()) + await conn.stop() + assert not conn.is_connected + + +async def test_error_callback_logging_is_throttled_and_reset_on_reconnect(log_records: list[Any]): + world = _FakeWorld() + conn = NatsConnection(_nats_settings(), client_factory=world, sleep=_RecordingSleep()) + await conn.start() + assert await conn.wait_ready(timeout=2) + options = world.clients[-1].connect_options + + await options["disconnected_cb"]() + for _ in range(5): + await options["error_cb"](OSError("Connection refused")) + await options["reconnected_cb"]() + await options["error_cb"](OSError("Connection reset")) + + warnings = [r["message"] for r in log_records if r["level"].name == "WARNING"] + errors_logged = [m for m in warnings if "error_category=nats_error" in m] + assert len(errors_logged) == 2 # noqa: PLR2004 - первая ошибка серии и первая после переподключения + assert any("Connection reset" in m for m in errors_logged) + await conn.stop() + + +# --- publish / subscribe ------------------------------------------------------------------------ +async def test_publish_without_connection_raises_publish_error(): + conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld()) + + with pytest.raises(NatsPublishError, match="NATS"): + await conn.publish("recognition.tabular.request", b"{}", "job:1") + + +async def test_publish_sends_msg_id_header_and_flushes(): + world = _FakeWorld() + conn = NatsConnection(_nats_settings(), client_factory=world, sleep=_RecordingSleep()) + await conn.start() + assert await conn.wait_ready(timeout=2) + + await conn.publish("recognition.tabular.request", b'{"a": 1}', "job:1") + + assert world.clients[-1].published == [ + {"subject": "recognition.tabular.request", "payload": b'{"a": 1}', "headers": {NATS_MSG_ID_HEADER: "job:1"}} + ] + await conn.stop() + + +async def test_publish_flush_failure_raises_publish_error(): + world = _FakeWorld() + conn = NatsConnection(_nats_settings(), client_factory=world, sleep=_RecordingSleep()) + await conn.start() + assert await conn.wait_ready(timeout=2) + world.clients[-1].flush_error = TimeoutError() + + with pytest.raises(NatsPublishError): + await conn.publish("recognition.tabular.request", b"{}", "job:1") + await conn.stop() + + +async def test_subscribe_results_handler_errors_are_logged_not_raised(log_records: list[Any]): + world = _FakeWorld() + conn = NatsConnection( + _nats_settings(), client_factory=world, sleep=_RecordingSleep(), result_subject="test.subject", queue="g" + ) + await conn.start() + assert await conn.wait_ready(timeout=2) + + async def failing_handler(msg: Any) -> None: + raise ValueError("сломанный обработчик") + + await conn.subscribe_results(failing_handler) + subscription = world.clients[-1].subscriptions[0] + assert (subscription["subject"], subscription["queue"]) == ("test.subject", "g") + + await subscription["cb"](SimpleNamespace(subject="test.subject", data=b"xyz", headers={})) + + errors = [r for r in log_records if r["level"].name == "ERROR"] + assert len(errors) == 1 + assert "error_category=handler_error" in errors[0]["message"] + await conn.stop() + + +async def test_subscribe_results_without_connection_raises(): + conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld()) + with pytest.raises(NatsNotConnectedError): + await conn.subscribe_results(_handler) + + +async def test_stop_drains_connected_client(): + world = _FakeWorld() + conn = NatsConnection(_nats_settings(), client_factory=world, sleep=_RecordingSleep()) + await conn.start() + assert await conn.wait_ready(timeout=2) + + await conn.stop() + + assert world.clients[-1].drained + assert not conn.is_connected + + +# --- без JetStream ------------------------------------------------------------------------------- +def test_src_does_not_use_jetstream(): + """JetStream на платформе нет: в src/ нет вызовов `.jetstream()` и импортов `nats.js`.""" + offenders: list[str] = [] + for path in SRC_DIR.rglob("*.py"): + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for node in ast.walk(tree): + if isinstance(node, ast.Attribute) and node.attr.lower() == "jetstream": + offenders.append(f"{path}:{node.lineno} .{node.attr}") + elif isinstance(node, ast.ImportFrom) and (node.module or "").startswith("nats.js"): + offenders.append(f"{path}:{node.lineno} from {node.module}") + elif isinstance(node, ast.Import): + offenders.extend( + f"{path}:{node.lineno} import {alias.name}" + for alias in node.names + if alias.name.startswith("nats.js") + ) + assert offenders == [] diff --git a/api/tests/test_nats_integration.py b/api/tests/test_nats_integration.py new file mode 100644 index 0000000..0c224d3 --- /dev/null +++ b/api/tests/test_nats_integration.py @@ -0,0 +1,169 @@ +"""Интеграционные тесты `src/core/nats.py` против NATS без JetStream из `.env.test` (spec recognition-job-exchange, +«Соединение с NATS по шаблону, без JetStream»; design D-6). + +Subject'ы — уникальные на тест (`test.`): общий `recognition.result` не трогаем, чтобы не пересекаться +с другими запущенными инстансами ядра. +""" + +import asyncio +import time +import uuid +from collections import Counter +from typing import Any + +import pytest +from nats.aio.client import Client as NATS +from nats.aio.msg import Msg + +from src.contract.subjects import NATS_MSG_ID_HEADER +from src.core.config import NATSSettings, get_settings +from src.core.nats import RESULT_QUEUE_GROUP, NatsConnection, NatsPublishError + +pytestmark = pytest.mark.integration + +READY_TIMEOUT = 5.0 +START_MAX_SECONDS = 0.5 # start() не ждёт подключения +STOP_MAX_SECONDS = 2.0 # stop() не висит в цикле подключения nats-py + + +def _nats_settings() -> NATSSettings: + return get_settings().OPENBAO.NATS + + +def _unique_subject() -> str: + return f"test.{uuid.uuid4().hex}" + + +async def _wait_until(predicate: Any, timeout: float = 5.0) -> None: + deadline = time.monotonic() + timeout + while not predicate(): + if time.monotonic() > deadline: + raise AssertionError("условие не выполнено за отведённое время") + await asyncio.sleep(0.02) + + +async def _start(conn: NatsConnection) -> NatsConnection: + await conn.start() + if not await conn.wait_ready(timeout=READY_TIMEOUT): + await conn.stop() + pytest.fail(f"NATS недоступен ({_nats_settings().NATS_URL}, .env.test)", pytrace=False) + return conn + + +async def test_publish_delivers_payload_and_msg_id_header(): + subject = _unique_subject() + received: list[Msg] = [] + + async def handler(msg: Msg) -> None: + received.append(msg) + + async def subscribe(conn: NatsConnection) -> None: + await conn.subscribe_results(handler) + + conn = await _start(NatsConnection(_nats_settings(), on_connected=subscribe, result_subject=subject)) + try: + assert conn.is_connected + await conn.publish(subject, b'{"job": 1}', "job-1:1") + await _wait_until(lambda: len(received) == 1) + finally: + await conn.stop() + + msg = received[0] + assert msg.subject == subject + assert msg.data == b'{"job": 1}' + assert msg.headers == {NATS_MSG_ID_HEADER: "job-1:1"} + + +async def test_queue_group_delivers_each_message_to_exactly_one_instance(): + """Два инстанса ядра в queue group `arm-core`: каждое сообщение обработано ровно одним (без дублей).""" + subject = _unique_subject() + received: dict[str, list[str]] = {"a": [], "b": []} + + def make_on_connected(name: str) -> Any: + async def handler(msg: Msg) -> None: + assert msg.headers is not None + received[name].append(msg.headers[NATS_MSG_ID_HEADER]) + + async def on_connected(conn: NatsConnection) -> None: + await conn.subscribe_results(handler) + + return on_connected + + instance_a = await _start( + NatsConnection(_nats_settings(), on_connected=make_on_connected("a"), result_subject=subject) + ) + instance_b = await _start( + NatsConnection(_nats_settings(), on_connected=make_on_connected("b"), result_subject=subject) + ) + assert instance_a.queue == instance_b.queue == RESULT_QUEUE_GROUP + total = 20 + try: + for i in range(total): + await instance_a.publish(subject, f"result {i}".encode(), f"result:job-{i}:1") + await _wait_until(lambda: len(received["a"]) + len(received["b"]) >= total) + await asyncio.sleep(0.2) # дубль, если бы он был, успел бы прийти + finally: + await instance_a.stop() + await instance_b.stop() + + all_ids = received["a"] + received["b"] + assert len(all_ids) == total + assert Counter(all_ids) == Counter(f"result:job-{i}:1" for i in range(total)) + + +async def test_subscription_survives_handler_error(): + subject = _unique_subject() + received: list[bytes] = [] + + async def handler(msg: Msg) -> None: + if msg.data == b"boom": + raise ValueError("сломанное сообщение") + received.append(msg.data) + + async def subscribe(conn: NatsConnection) -> None: + await conn.subscribe_results(handler) + + conn = await _start(NatsConnection(_nats_settings(), on_connected=subscribe, result_subject=subject)) + try: + await conn.publish(subject, b"boom", "m:1") + await conn.publish(subject, b"ok", "m:2") + await _wait_until(lambda: received == [b"ok"]) + finally: + await conn.stop() + + +async def test_start_against_unreachable_server_does_not_block_and_stops_cleanly(): + """Регрессия T4.2: недоступный NATS при старте — start() не ждёт, is_connected=False, stop() быстрый.""" + settings = NATSSettings(NATS_URL="nats://127.0.0.1:54999") + conn = NatsConnection(settings) + + started = time.monotonic() + await conn.start() + assert time.monotonic() - started < START_MAX_SECONDS + await asyncio.sleep(0.3) + assert not conn.is_connected + with pytest.raises(NatsPublishError): + await conn.publish("test.unreachable", b"{}", "m:1") + + started = time.monotonic() + await conn.stop() + assert time.monotonic() - started < STOP_MAX_SECONDS + assert not conn.is_connected + + +async def test_plain_client_receives_publish_from_connection(): + """Сторонний подписчик (как сервис распознавания) видит заголовок `Nats-Msg-Id` от `NatsConnection.publish`.""" + subject = _unique_subject() + nc = NATS() + await nc.connect(_nats_settings().NATS_URL, connect_timeout=2, max_reconnect_attempts=1) + subscription = await nc.subscribe(subject) + conn = await _start(NatsConnection(_nats_settings())) + try: + await conn.publish(subject, b"payload", "job-x:2") + msg = await subscription.next_msg(timeout=5) + finally: + await conn.stop() + await nc.close() + + assert msg.data == b"payload" + assert msg.headers == {NATS_MSG_ID_HEADER: "job-x:2"}