350 lines
13 KiB
Python
350 lines
13 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 (
|
||
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)
|