4e5e1a7b5b
- 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>
380 lines
14 KiB
Python
380 lines
14 KiB
Python
"""Быстрые тесты `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 == []
|