bf2f60b3b5
nats-py 2.14 при Permissions Violation не закрывает соединение и не роняет subscribe/flush/publish, а только вызывает error_cb. Раньше это логировалось как «Нет связи с NATS, повторяем подключение» с лимитом раз в минуту. Теперь _on_error распознаёт «permissions violation» и пишет error error_category=nats_permission с subject, каждый раз и без серии ошибок подключения. Исправлены ложные утверждения docstring («ошибки прав проявляются сразу»). connect получает drain_timeout=DRAIN_TIMEOUT (10 с вместо 30 с nats-py): STOP_TIMEOUT воркера + drain укладываются в 30 с платформы. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
477 lines
20 KiB
Python
477 lines
20 KiB
Python
"""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 (
|
||
DRAIN_TIMEOUT,
|
||
REQUEST_QUEUE_GROUP,
|
||
NatsConfigError,
|
||
NatsConnection,
|
||
NatsNotConnectedError,
|
||
NatsPublishError,
|
||
Publisher,
|
||
build_nats_auth,
|
||
)
|
||
from src.core.worker import STOP_TIMEOUT
|
||
|
||
JWT = "header.payload.signature"
|
||
PLATFORM_STOP_GRACE = 30.0 # секунд, которые платформа даёт процессу на завершение
|
||
|
||
|
||
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 _FakeSubscription:
|
||
def __init__(self, client: "_FakeClient", subject: str, queue: str, cb: Any) -> None:
|
||
self.client = client
|
||
self.subject = subject
|
||
self.queue = queue
|
||
self.cb = cb
|
||
self.active = True
|
||
|
||
async def unsubscribe(self) -> None:
|
||
self.active = False
|
||
self.client.calls.append("unsubscribe")
|
||
|
||
|
||
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
|
||
self.calls: list[str] = [] # порядок subscribe/flush/unsubscribe
|
||
|
||
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) -> _FakeSubscription:
|
||
self.calls.append("subscribe")
|
||
subscription = _FakeSubscription(self, subject, queue, cb)
|
||
self.subscriptions.append({"subject": subject, "queue": queue, "cb": cb, "sub": subscription})
|
||
return subscription
|
||
|
||
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:
|
||
self.calls.append("flush")
|
||
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:
|
||
if self.world.hang_close:
|
||
await asyncio.Event().wait() # медленная очистка соединения
|
||
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, hang_close: bool = False
|
||
) -> None:
|
||
self.fail_times = fail_times
|
||
self.flush_failures = flush_failures
|
||
self.hang_connect = hang_connect
|
||
self.hang_close = hang_close
|
||
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"
|
||
# drain при остановке ограничен, чтобы остановка уложилась в 30 с платформы (STOP_TIMEOUT воркера + drain)
|
||
assert options["drain_timeout"] == DRAIN_TIMEOUT
|
||
assert STOP_TIMEOUT + DRAIN_TIMEOUT <= PLATFORM_STOP_GRACE
|
||
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_stop_propagates_cancellation_of_its_caller():
|
||
"""T8.5: отмена задачи, вызвавшей stop(), не проглатывается (подавляется только отмена самого supervisor'а)."""
|
||
world = _FakeWorld(hang_connect=True, hang_close=True)
|
||
conn = NatsConnection(_nats_settings(), client_factory=world, sleep=_RecordingSleep())
|
||
await conn.start()
|
||
await asyncio.sleep(0)
|
||
|
||
stop_task = asyncio.create_task(conn.stop())
|
||
for _ in range(5):
|
||
await asyncio.sleep(0) # supervisor отменён и висит в медленном close()
|
||
assert not stop_task.done()
|
||
stop_task.cancel()
|
||
|
||
# asyncio.wait, а не await: без исправления stop() проглатывает отмену и виснет в close()
|
||
await asyncio.wait({stop_task}, timeout=1)
|
||
assert stop_task.cancelled()
|
||
|
||
|
||
async def test_on_closed_logs_error_when_not_stopping_and_info_on_stop(log_records: list[Any]):
|
||
"""T8.6: самопроизвольное закрытие клиента — error; закрытие в stop() — info."""
|
||
world = _FakeWorld()
|
||
conn = await _connected(world, _RecordingSleep())
|
||
|
||
await conn._on_closed()
|
||
await conn.stop()
|
||
await conn._on_closed()
|
||
|
||
closed = [r for r in log_records if "Соединение с NATS закрыто" in r["message"]]
|
||
assert [r["level"].name for r in closed] == ["ERROR", "INFO"]
|
||
|
||
|
||
# --- подписка -----------------------------------------------------------------------------------
|
||
async def test_permission_violation_is_logged_as_permission_error_not_reconnect(log_records: list[Any]):
|
||
"""nats-py 2.14 при Permissions Violation не рвёт соединение, только вызывает error_cb (текст в нижнем регистре)."""
|
||
conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld(), sleep=_RecordingSleep())
|
||
error = Exception('nats: permissions violation for subscription to "recognition.tabular.request"')
|
||
|
||
repeats = 2 # повтор не глушится лимитом ошибок подключения
|
||
for _ in range(repeats):
|
||
await conn._on_error(error)
|
||
|
||
permission = [r for r in log_records if "error_category=nats_permission" in r["message"]]
|
||
assert len(permission) == repeats
|
||
assert all(r["level"].name == "ERROR" for r in permission)
|
||
assert all("subject=recognition.tabular.request" in r["message"] for r in permission)
|
||
assert not any("повторяем подключение" in r["message"] for r in log_records)
|
||
assert conn._error_streak == 0 # серия ошибок подключения не затронута
|
||
|
||
|
||
async def test_connection_error_in_error_cb_keeps_rate_limited_reconnect_warning(log_records: list[Any]):
|
||
conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld(), sleep=_RecordingSleep())
|
||
|
||
await conn._on_error(OSError("Connection refused"))
|
||
await conn._on_error(OSError("Connection refused"))
|
||
|
||
warnings = [r for r in log_records if "повторяем подключение" in r["message"]]
|
||
assert len(warnings) == 1 # вторая ошибка серии — не чаще раза в минуту
|
||
assert warnings[0]["level"].name == "WARNING"
|
||
assert not any("nats_permission" in r["message"] for r in log_records)
|
||
|
||
|
||
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_flushes_after_subscribe():
|
||
"""T8.3: подписка считается активной, когда сервер обработал SUB (flush после subscribe)."""
|
||
world = _FakeWorld()
|
||
conn = await _connected(world, _RecordingSleep())
|
||
|
||
await conn.subscribe_requests(_noop_handler)
|
||
|
||
assert world.clients[-1].calls == ["subscribe", "flush"]
|
||
await conn.stop()
|
||
|
||
|
||
async def test_on_connected_failure_is_retried_without_duplicate_subscriptions(log_records: list[Any]):
|
||
"""T8.4: on_connected упал после подписки — подписка снимается, повтор с паузой, в итоге одна активная подписка."""
|
||
world = _FakeWorld()
|
||
sleep = _RecordingSleep()
|
||
calls = 0
|
||
|
||
async def on_connected(conn: NatsConnection) -> None:
|
||
nonlocal calls
|
||
calls += 1
|
||
await conn.subscribe_requests(_noop_handler)
|
||
if calls == 1:
|
||
raise RuntimeError("сбой после подписки")
|
||
|
||
conn = NatsConnection(_nats_settings(), on_connected=on_connected, client_factory=world, sleep=sleep)
|
||
await conn.start()
|
||
assert await conn.wait_ready(timeout=2)
|
||
|
||
assert calls == 2 # noqa: PLR2004
|
||
assert sleep.delays == [1]
|
||
subscriptions = [s["sub"] for s in world.clients[-1].subscriptions]
|
||
assert [s.active for s in subscriptions] == [False, True]
|
||
assert any("Не удалось выполнить подписки NATS" in r["message"] for r in log_records)
|
||
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")
|
||
|
||
# PUBLISH_RETRIES=3 — три повтора после первой попытки (R-16): 4 попытки, 3 паузы, один и тот же Nats-Msg-Id
|
||
published = world.clients[-1].published
|
||
assert len(published) == 4 # noqa: PLR2004
|
||
assert {p["headers"][NATS_MSG_ID_HEADER] for p in published} == {"result:j-1:1"}
|
||
assert sleep.delays == [0.5, 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) == 4 # 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, 2.0] # 1 попытка + 2 повтора
|
||
|
||
|
||
async def test_nats_connection_satisfies_publisher_protocol():
|
||
publisher: Publisher = NatsConnection(_nats_settings(), client_factory=_FakeWorld())
|
||
assert isinstance(publisher, Publisher)
|