RSHB/arm-core
9
0
Fork 0
Code Issues 25 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: src/core/nats.py — фоновое подключение к core NATS, publish с Nats-Msg-Id, подписка recognition.result в queue group arm-core

Browse Source 
- 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 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-10-03 23:47:39 +04:00
parent dcbeea5756
commit 4e5e1a7b5b
7 changed files with 968 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
README.md
View File

@ -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/ доступ к данным

87
api/src/app.py
View File

@ -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()

310
api/src/core/nats.py Normal file
View File

@ -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: <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 закрыто")

14
api/src/services/recognition_exchange.py Normal file
View File

@ -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))

115
api/tests/test_app.py
View File

@ -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()

379
api/tests/test_nats_core.py Normal file
View File

@ -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 == []

169
api/tests/test_nats_integration.py Normal file
View File

@ -0,0 +1,169 @@
"""Интеграционные тесты `src/core/nats.py` против NATS без JetStream из `.env.test` (spec recognition-job-exchange,
«Соединение с NATS по шаблону, без JetStream»; design D-6).
Subject'ы — уникальные на тест (`test.<uuid>`): общий `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"}