feat: слой NATS src/core/nats.py — фоновое подключение (nkeys/без), подписка на запросы в queue group, публикация с повторами
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
4c32cd36ac
commit
ff08905580
335
api/src/core/nats.py
Normal file
335
api/src/core/nats.py
Normal file
@ -0,0 +1,335 @@
|
||||
"""Соединение сервиса с NATS: core NATS без JetStream (design D-2, D-5, D-6; spec «Соединение с NATS по правилам
|
||||
шаблона, без JetStream», «Подтверждение запроса до обработки», «Результат по контракту»). Сторона сервиса
|
||||
распознавания; образец — `arm-core/api/src/core/nats.py` (сторона ядра).
|
||||
|
||||
- Аутентификация как в шаблоне: заданы `NATS_SEED` и `NATS_JWT` — nkeys/JWT; оба пусты — без учётных данных; задан
|
||||
только один — `NatsConfigError` (подключение не выполняется, ошибка в логе).
|
||||
- Первое подключение выполняет фоновый supervisor (`start()`): старт приложения не ждёт NATS. nats-py 2.14 с
|
||||
`max_reconnect_attempts=-1` сам повторяет и первое подключение (параметра `retry_on_failed_connect` в 2.14 нет:
|
||||
`connect` просто не возвращается, пока сервер недоступен), поэтому ждать его в lifespan нельзя. Если `connect` всё же
|
||||
завершился исключением, supervisor повторяет попытку с паузой 1 → 2 → 4 → … ≤ 30 с. После подключения один раз
|
||||
вызывается `on_connected(conn)` (подписки воркера); подписки после разрыва nats-py восстанавливает сам.
|
||||
- `subscribe_requests(cb)` — подписка на `recognition.tabular.request` в queue group `tabular-parser`: из нескольких
|
||||
инстанций сообщение получает одна. Исключения обработчика логируются, подписка живёт.
|
||||
- `publish(subject, payload, msg_id)` — заголовок `Nats-Msg-Id` и `flush` (подтверждение сервером); до
|
||||
`publish_retries` попыток с паузой `publish_retry_pause` между ними (D-5: 3 попытки, 2 с), затем `NatsPublishError` —
|
||||
воркер логирует ошибку, задание дойдёт до `timeout` в ядре.
|
||||
- Ошибки подключения логируются без шума: первая ошибка серии — warning, дальше не чаще раза в минуту.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import base64
|
||||
import time
|
||||
from collections.abc import Awaitable, Callable
|
||||
from contextlib import suppress
|
||||
from typing import Any, Protocol, runtime_checkable
|
||||
|
||||
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, TABULAR_REQUEST_SUBJECT
|
||||
from src.core.config import NATSSettings
|
||||
|
||||
REQUEST_QUEUE_GROUP = "tabular-parser"
|
||||
CLIENT_NAME = "tabular-parser"
|
||||
MAX_RECONNECT_ATTEMPTS = -1 # переподключаться бесконечно
|
||||
RECONNECT_TIME_WAIT = 2 # секунд между попытками переподключения nats-py
|
||||
CONNECT_TIMEOUT = 2 # секунд на одну попытку TCP-подключения
|
||||
FLUSH_TIMEOUT = 5 # секунд на подтверждение публикации сервером (PING/PONG)
|
||||
PUBLISH_RETRIES = 3 # попыток публикации (settings.PUBLISH_RETRIES)
|
||||
PUBLISH_RETRY_PAUSE = 2.0 # секунд между попытками публикации (D-5)
|
||||
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):
|
||||
"""Публикация не подтверждена за все попытки."""
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class Publisher(Protocol):
|
||||
"""Публикация в NATS, нужная воркеру (design D-7): исключение `NatsPublishError` — все попытки исчерпаны."""
|
||||
|
||||
async def publish(self, subject: str, payload: bytes, msg_id: str) -> None: ...
|
||||
|
||||
|
||||
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: фоновое подключение, подписка на запросы, публикация с повторами.
|
||||
|
||||
Реализует `Publisher`. `client_factory` и `sleep` подменяются в тестах; `sleep` используется и для пауз
|
||||
переподключения supervisor'а, и для пауз между попытками публикации.
|
||||
"""
|
||||
|
||||
def __init__( # noqa: PLR0913 - все параметры, кроме settings, именованные: подмены для тестов и тайминги
|
||||
self,
|
||||
settings: NATSSettings,
|
||||
*,
|
||||
on_connected: Callable[["NatsConnection"], Awaitable[None]] | None = None,
|
||||
publish_retries: int = PUBLISH_RETRIES,
|
||||
publish_retry_pause: float = PUBLISH_RETRY_PAUSE,
|
||||
client_factory: Callable[[], Any] = NATS,
|
||||
sleep: SleepFn = asyncio.sleep,
|
||||
backoff_initial: float = BACKOFF_INITIAL,
|
||||
backoff_max: float = BACKOFF_MAX,
|
||||
flush_timeout: int = FLUSH_TIMEOUT,
|
||||
) -> None:
|
||||
if publish_retries < 1:
|
||||
raise ValueError("publish_retries должен быть ≥ 1")
|
||||
self._settings = settings
|
||||
self._on_connected = on_connected
|
||||
self._publish_retries = publish_retries
|
||||
self._publish_retry_pause = publish_retry_pause
|
||||
self._client_factory = client_factory
|
||||
self._sleep = sleep
|
||||
self._backoff_initial = backoff_initial
|
||||
self._backoff_max = backoff_max
|
||||
self._flush_timeout = flush_timeout
|
||||
self._client: Any = None
|
||||
self._supervisor: asyncio.Task[None] | None = None
|
||||
self._ready = asyncio.Event()
|
||||
self._stopping = False
|
||||
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
|
||||
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 subscribe_requests(
|
||||
self, cb: MessageHandler, *, queue: str = REQUEST_QUEUE_GROUP, subject: str = TABULAR_REQUEST_SUBJECT
|
||||
) -> Any:
|
||||
"""Подписка на запросы (`recognition.tabular.request`) в queue group; исключения `cb` логируются."""
|
||||
if not self.is_connected:
|
||||
raise NatsNotConnectedError(f"Нет соединения с NATS: подписка на {subject} невозможна")
|
||||
|
||||
async def safe_cb(msg: Msg) -> None:
|
||||
try:
|
||||
await cb(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(subject, queue=queue, cb=safe_cb)
|
||||
logger.info("Подписка NATS активна. subject={subject}, queue={queue}", subject=subject, queue=queue)
|
||||
return subscription
|
||||
|
||||
async def publish(self, subject: str, payload: bytes, msg_id: str) -> None:
|
||||
"""Публикует с `Nats-Msg-Id: <msg_id>`; до `publish_retries` попыток с паузой, затем `NatsPublishError`."""
|
||||
last_error: str = ""
|
||||
for attempt in range(1, self._publish_retries + 1):
|
||||
if attempt > 1:
|
||||
await self._sleep(self._publish_retry_pause)
|
||||
try:
|
||||
await self._publish_once(subject, payload, msg_id)
|
||||
return
|
||||
except NatsPublishError as exc:
|
||||
last_error = str(exc)
|
||||
logger.warning(
|
||||
"Публикация в NATS не удалась. error_category=nats_error, subject={subject}, msg_id={msg_id}, "
|
||||
"attempt={attempt}, attempts={attempts}, error={error}",
|
||||
subject=subject,
|
||||
msg_id=msg_id,
|
||||
attempt=attempt,
|
||||
attempts=self._publish_retries,
|
||||
error=last_error,
|
||||
)
|
||||
raise NatsPublishError(
|
||||
f"Публикация {msg_id} в {subject} не удалась за {self._publish_retries} попыток: {last_error}"
|
||||
)
|
||||
|
||||
async def _publish_once(self, subject: str, payload: bytes, msg_id: str) -> None:
|
||||
if not self.is_connected:
|
||||
raise NatsPublishError("Нет соединения с NATS")
|
||||
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"публикация не подтверждена ({type(exc).__name__})") from exc
|
||||
|
||||
# --- колбэки 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 >= 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 not self._stopping:
|
||||
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 закрыто")
|
||||
349
api/tests/unit/test_nats.py
Normal file
349
api/tests/unit/test_nats.py
Normal file
@ -0,0 +1,349 @@
|
||||
"""Unit-тесты `src/core/nats.py` без сервера NATS: аутентификация, фоновое подключение, подписка, публикация.
|
||||
|
||||
Клиент nats-py подменяется фабрикой (`client_factory`), паузы — инжектируемым `sleep` (design D-5, D-6; spec
|
||||
«Соединение с NATS по правилам шаблона, без JetStream», «Результат по контракту» — повтор публикации).
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import base64
|
||||
import os
|
||||
from collections.abc import Iterator
|
||||
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, TABULAR_REQUEST_SUBJECT
|
||||
from src.core.config import NATSSettings
|
||||
from src.core.nats import (
|
||||
REQUEST_QUEUE_GROUP,
|
||||
NatsConfigError,
|
||||
NatsConnection,
|
||||
NatsNotConnectedError,
|
||||
NatsPublishError,
|
||||
Publisher,
|
||||
build_nats_auth,
|
||||
)
|
||||
|
||||
JWT = "header.payload.signature"
|
||||
|
||||
|
||||
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", JWT))
|
||||
|
||||
|
||||
def test_build_nats_auth_valid_credentials_give_callbacks():
|
||||
seed = _valid_seed()
|
||||
auth = build_nats_auth(_nats_settings(seed, JWT))
|
||||
|
||||
assert set(auth) == {"user_jwt_cb", "signature_cb"}
|
||||
assert auth["user_jwt_cb"]() == JWT.encode()
|
||||
signature = auth["signature_cb"]("nonce")
|
||||
# Подпись проверяется публичным ключом той же пары
|
||||
keypair = nkeys.from_seed(seed.encode())
|
||||
assert keypair.verify(b"nonce", base64.b64decode(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.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.hang_connect:
|
||||
await asyncio.Event().wait() # NATS недоступен: nats-py 2.14 повторяет подключение бесконечно
|
||||
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.world.flush_failures > 0:
|
||||
self.world.flush_failures -= 1
|
||||
raise TimeoutError("flush timeout")
|
||||
|
||||
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` подключений падают; `flush_failures` первых flush — таймаут."""
|
||||
|
||||
def __init__(self, fail_times: int = 0, flush_failures: int = 0, hang_connect: bool = False) -> None:
|
||||
self.fail_times = fail_times
|
||||
self.flush_failures = flush_failures
|
||||
self.hang_connect = hang_connect
|
||||
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`: запоминает паузы и не ждёт."""
|
||||
|
||||
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 _noop_handler(msg: Any) -> None:
|
||||
return None
|
||||
|
||||
|
||||
async def _connected(world: _FakeWorld, sleep: _RecordingSleep, **kwargs: Any) -> NatsConnection:
|
||||
conn = NatsConnection(_nats_settings(), client_factory=world, sleep=sleep, **kwargs)
|
||||
await conn.start()
|
||||
assert await conn.wait_ready(timeout=2)
|
||||
return conn
|
||||
|
||||
|
||||
# --- подключение ------------------------------------------------------------------------------
|
||||
async def test_connect_without_credentials_infinite_reconnect_and_callbacks():
|
||||
world = _FakeWorld()
|
||||
conn = await _connected(world, _RecordingSleep())
|
||||
|
||||
assert conn.is_connected
|
||||
options = world.clients[-1].connect_options
|
||||
assert options["servers"] == "nats://localhost:4222"
|
||||
assert options["max_reconnect_attempts"] == -1
|
||||
assert options["name"] == "tabular-parser"
|
||||
for callback in ("error_cb", "disconnected_cb", "reconnected_cb", "closed_cb"):
|
||||
assert callable(options[callback])
|
||||
assert "user_jwt_cb" not in options
|
||||
assert "signature_cb" not in options
|
||||
await conn.stop()
|
||||
|
||||
|
||||
async def test_connect_with_credentials_passes_nkeys_callbacks():
|
||||
world = _FakeWorld()
|
||||
conn = NatsConnection(_nats_settings(_valid_seed(), JWT), client_factory=world, sleep=_RecordingSleep())
|
||||
await conn.start()
|
||||
assert await conn.wait_ready(timeout=2)
|
||||
|
||||
options = world.clients[-1].connect_options
|
||||
assert options["user_jwt_cb"]() == JWT.encode()
|
||||
assert callable(options["signature_cb"])
|
||||
await conn.stop()
|
||||
|
||||
|
||||
async def test_supervisor_retries_with_backoff_then_connects_and_runs_on_connected_once(log_records: list[Any]):
|
||||
world = _FakeWorld(fail_times=7)
|
||||
sleep = _RecordingSleep()
|
||||
called: list[NatsConnection] = []
|
||||
|
||||
async def on_connected(conn: NatsConnection) -> None:
|
||||
called.append(conn)
|
||||
await conn.subscribe_requests(_noop_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 world.connect_calls == 8 # noqa: PLR2004
|
||||
assert sleep.delays == [1, 2, 4, 8, 16, 30, 30]
|
||||
assert called == [conn]
|
||||
# Первая ошибка серии — 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"]
|
||||
await conn.stop()
|
||||
|
||||
|
||||
async def test_start_does_not_wait_for_unavailable_nats_and_stop_is_clean():
|
||||
"""Регрессия O-3: NATS недоступен при старте — start() возвращается сразу, stop() отменяет подключение."""
|
||||
world = _FakeWorld(hang_connect=True)
|
||||
conn = NatsConnection(_nats_settings(), client_factory=world, sleep=_RecordingSleep())
|
||||
|
||||
await asyncio.wait_for(conn.start(), timeout=0.5)
|
||||
await asyncio.sleep(0)
|
||||
assert not conn.is_connected
|
||||
assert not await conn.wait_ready(timeout=0.05)
|
||||
|
||||
await asyncio.wait_for(conn.stop(), timeout=1)
|
||||
assert world.clients[-1].is_closed
|
||||
|
||||
|
||||
async def test_partial_credentials_stop_supervisor_with_config_error(log_records: list[Any]):
|
||||
world = _FakeWorld()
|
||||
conn = NatsConnection(_nats_settings(seed="", jwt=JWT), client_factory=world, sleep=_RecordingSleep())
|
||||
|
||||
await conn.start()
|
||||
assert not await conn.wait_ready(timeout=0.1)
|
||||
|
||||
assert world.connect_calls == 0
|
||||
errors = [r for r in log_records if r["level"].name == "ERROR"]
|
||||
assert any("error_category=config_error" in r["message"] for r in errors)
|
||||
await conn.stop()
|
||||
|
||||
|
||||
async def test_stop_drains_connected_client():
|
||||
world = _FakeWorld()
|
||||
conn = await _connected(world, _RecordingSleep())
|
||||
|
||||
await conn.stop()
|
||||
|
||||
assert world.clients[-1].drained
|
||||
assert not conn.is_connected
|
||||
|
||||
|
||||
# --- подписка -----------------------------------------------------------------------------------
|
||||
async def test_subscribe_requests_uses_tabular_subject_and_queue_group():
|
||||
world = _FakeWorld()
|
||||
conn = await _connected(world, _RecordingSleep())
|
||||
|
||||
await conn.subscribe_requests(_noop_handler)
|
||||
|
||||
assert REQUEST_QUEUE_GROUP == "tabular-parser"
|
||||
subscription = world.clients[-1].subscriptions[0]
|
||||
assert (subscription["subject"], subscription["queue"]) == (TABULAR_REQUEST_SUBJECT, "tabular-parser")
|
||||
await conn.stop()
|
||||
|
||||
|
||||
async def test_subscribe_requests_handler_errors_are_logged_not_raised(log_records: list[Any]):
|
||||
world = _FakeWorld()
|
||||
conn = await _connected(world, _RecordingSleep())
|
||||
received: list[bytes] = []
|
||||
|
||||
async def handler(msg: Any) -> None:
|
||||
if msg.data == b"boom":
|
||||
raise ValueError("сломанное сообщение")
|
||||
received.append(msg.data)
|
||||
|
||||
await conn.subscribe_requests(handler)
|
||||
safe_cb = world.clients[-1].subscriptions[0]["cb"]
|
||||
await safe_cb(SimpleNamespace(subject=TABULAR_REQUEST_SUBJECT, data=b"boom"))
|
||||
await safe_cb(SimpleNamespace(subject=TABULAR_REQUEST_SUBJECT, data=b"ok"))
|
||||
|
||||
assert received == [b"ok"]
|
||||
assert any("error_category=handler_error" in r["message"] for r in log_records)
|
||||
await conn.stop()
|
||||
|
||||
|
||||
async def test_subscribe_requests_without_connection_raises():
|
||||
conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld(), sleep=_RecordingSleep())
|
||||
|
||||
with pytest.raises(NatsNotConnectedError):
|
||||
await conn.subscribe_requests(_noop_handler)
|
||||
|
||||
|
||||
# --- публикация ---------------------------------------------------------------------------------
|
||||
async def test_publish_adds_nats_msg_id_header_and_flushes():
|
||||
world = _FakeWorld()
|
||||
sleep = _RecordingSleep()
|
||||
conn = await _connected(world, sleep)
|
||||
|
||||
await conn.publish(RESULT_SUBJECT, b'{"status": "processing"}', "ack:j-1:1")
|
||||
|
||||
assert world.clients[-1].published == [
|
||||
{
|
||||
"subject": RESULT_SUBJECT,
|
||||
"payload": b'{"status": "processing"}',
|
||||
"headers": {NATS_MSG_ID_HEADER: "ack:j-1:1"},
|
||||
}
|
||||
]
|
||||
assert sleep.delays == []
|
||||
await conn.stop()
|
||||
|
||||
|
||||
async def test_publish_retries_with_pause_then_succeeds_with_same_msg_id():
|
||||
world = _FakeWorld(flush_failures=1)
|
||||
sleep = _RecordingSleep()
|
||||
conn = await _connected(world, sleep)
|
||||
|
||||
await conn.publish(RESULT_SUBJECT, b"{}", "result:j-1:1")
|
||||
|
||||
published = world.clients[-1].published
|
||||
assert len(published) == 2 # noqa: PLR2004 - первая попытка не подтверждена flush, вторая успешна
|
||||
assert {p["headers"][NATS_MSG_ID_HEADER] for p in published} == {"result:j-1:1"}
|
||||
assert sleep.delays == [2.0]
|
||||
await conn.stop()
|
||||
|
||||
|
||||
async def test_publish_retries_exhausted_raises_publish_error(log_records: list[Any]):
|
||||
world = _FakeWorld(flush_failures=10)
|
||||
sleep = _RecordingSleep()
|
||||
conn = await _connected(world, sleep, publish_retries=3, publish_retry_pause=0.5)
|
||||
|
||||
with pytest.raises(NatsPublishError, match="result:j-1:1"):
|
||||
await conn.publish(RESULT_SUBJECT, b"{}", "result:j-1:1")
|
||||
|
||||
assert len(world.clients[-1].published) == 3 # noqa: PLR2004
|
||||
assert sleep.delays == [0.5, 0.5] # пауза только между попытками
|
||||
warnings = [r for r in log_records if r["level"].name == "WARNING" and "result:j-1:1" in r["message"]]
|
||||
assert len(warnings) == 3 # noqa: PLR2004
|
||||
await conn.stop()
|
||||
|
||||
|
||||
async def test_publish_without_connection_retries_then_raises():
|
||||
sleep = _RecordingSleep()
|
||||
conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld(), sleep=sleep, publish_retries=2)
|
||||
|
||||
with pytest.raises(NatsPublishError, match="Нет соединения"):
|
||||
await conn.publish(RESULT_SUBJECT, b"{}", "ack:j-1:1")
|
||||
|
||||
assert sleep.delays == [2.0]
|
||||
|
||||
|
||||
async def test_nats_connection_satisfies_publisher_protocol():
|
||||
publisher: Publisher = NatsConnection(_nats_settings(), client_factory=_FakeWorld())
|
||||
assert isinstance(publisher, Publisher)
|
||||
Loading…
x
Reference in New Issue
Block a user