6c287ef4bc
Решение владельца 08.10.2026: recognition.tabular.request → files.tabular_uploaded, recognition.image.request → files.img_uploaded, recognition.result → files.tabular_parsed (общий subject ack и результатов; files.img_parsed не используется). Тела сообщений, Nats-Msg-Id, queue group'ы и schema_version не меняются. Константы api/src/contract/subjects.py, докстринги, тесты, README, CLAUDE.md; копии схем api/contracts/ из docs/contracts/ побайтно; change nats-files-subjects с дельтой спеки. Сквозной change — RSHB/arm-ka-docs#74. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
432 lines
16 KiB
Python
432 lines
16 KiB
Python
"""Слой NATS (`src.core.nats`): подключение с/без nkeys, колбэки переподключения, подписка queue group, публикация
|
||
с `Nats-Msg-Id` и повторами. Клиент nats-py подменяется фейком — сеть не нужна. Переподключение к живому серверу
|
||
не проверяется (общий контейнер NATS не перезапускаем): здесь — только колбэки и восстановление подписки клиентом.
|
||
"""
|
||
|
||
import base64
|
||
import json
|
||
import uuid
|
||
from typing import Any
|
||
|
||
import nkeys
|
||
import pytest
|
||
from nats import errors as nats_errors
|
||
|
||
from src.contract import (
|
||
IMAGE_REQUEST_SUBJECT,
|
||
NATS_MSG_ID_HEADER,
|
||
RESULT_SUBJECT,
|
||
ErrorCode,
|
||
ServiceInfo,
|
||
build_ack_message,
|
||
build_error_message,
|
||
)
|
||
from src.core.config import Settings, get_settings
|
||
from src.core.nats import (
|
||
REQUEST_QUEUE_GROUP,
|
||
NatsConfigError,
|
||
NatsGateway,
|
||
NatsNotConnectedError,
|
||
PublishFailed,
|
||
)
|
||
|
||
|
||
class FakeMsg:
|
||
def __init__(self, data: bytes = b"{}", headers: dict[str, str] | None = None) -> None:
|
||
self.subject = IMAGE_REQUEST_SUBJECT
|
||
self.data = data
|
||
self.headers = headers
|
||
|
||
|
||
class FakeNatsClient:
|
||
"""Минимальная подмена `nats.aio.client.Client`: запоминает вызовы, ошибки публикации задаются очередью."""
|
||
|
||
def __init__(self) -> None:
|
||
self.connect_args: tuple[Any, ...] = ()
|
||
self.connect_kwargs: dict[str, Any] = {}
|
||
self.is_connected = False
|
||
self.is_closed = False
|
||
self.published: list[tuple[str, bytes, dict[str, str] | None]] = []
|
||
self.publish_errors: list[BaseException] = []
|
||
self.flush_calls = 0
|
||
self.subscriptions: list[dict[str, Any]] = []
|
||
self.drained = False
|
||
self.closed = False
|
||
self.jetstream_calls = 0
|
||
|
||
async def connect(self, *args: Any, **kwargs: Any) -> None:
|
||
self.connect_args = args
|
||
self.connect_kwargs = kwargs
|
||
self.is_connected = True
|
||
|
||
async def publish(self, subject: str, payload: bytes = b"", headers: dict[str, str] | None = None) -> None:
|
||
self.published.append((subject, payload, headers))
|
||
if self.publish_errors:
|
||
raise self.publish_errors.pop(0)
|
||
|
||
async def flush(self, timeout: float = 10) -> None:
|
||
self.flush_calls += 1
|
||
|
||
async def subscribe(self, subject: str, queue: str = "", cb: Any = None) -> str:
|
||
self.subscriptions.append({"subject": subject, "queue": queue, "cb": cb})
|
||
return f"sub-{len(self.subscriptions)}"
|
||
|
||
async def drain(self) -> None:
|
||
self.drained = True
|
||
self.is_connected = False
|
||
self.is_closed = True
|
||
|
||
async def close(self) -> None:
|
||
self.closed = True
|
||
self.is_connected = False
|
||
self.is_closed = True
|
||
|
||
def jetstream(self, **opts: Any) -> None: # pragma: no cover - вызов и есть ошибка
|
||
self.jetstream_calls += 1
|
||
raise AssertionError("JetStream не используется (D-8)")
|
||
|
||
|
||
def make_settings(monkeypatch: pytest.MonkeyPatch, **env: str) -> Settings:
|
||
for name, value in env.items():
|
||
monkeypatch.setenv(name, value)
|
||
get_settings.cache_clear()
|
||
return get_settings()
|
||
|
||
|
||
@pytest.fixture
|
||
def settings(monkeypatch: pytest.MonkeyPatch) -> Settings:
|
||
return make_settings(monkeypatch)
|
||
|
||
|
||
@pytest.fixture
|
||
def fake() -> FakeNatsClient:
|
||
return FakeNatsClient()
|
||
|
||
|
||
@pytest.fixture
|
||
def sleeps() -> list[float]:
|
||
return []
|
||
|
||
|
||
@pytest.fixture
|
||
def gateway(settings: Settings, fake: FakeNatsClient, sleeps: list[float]) -> NatsGateway:
|
||
async def fake_sleep(delay: float) -> None:
|
||
sleeps.append(delay)
|
||
|
||
return NatsGateway(settings, client_factory=lambda: fake, retry_delay=0, sleep=fake_sleep)
|
||
|
||
|
||
# --- подключение ------------------------------------------------------------------------------------------------
|
||
|
||
|
||
async def test_connect_without_credentials_when_seed_and_jwt_empty(gateway: NatsGateway, fake: FakeNatsClient):
|
||
assert await gateway.connect() is True
|
||
|
||
assert fake.connect_args == ("nats://127.0.0.1:4222",)
|
||
kwargs = fake.connect_kwargs
|
||
assert kwargs["max_reconnect_attempts"] == -1
|
||
assert kwargs["name"] == "scan-gateway"
|
||
for callback in ("error_cb", "disconnected_cb", "reconnected_cb", "closed_cb"):
|
||
assert callable(kwargs[callback])
|
||
for credential in ("user_jwt_cb", "signature_cb", "nkeys_seed", "nkeys_seed_str", "user_credentials", "token"):
|
||
assert credential not in kwargs
|
||
assert fake.jetstream_calls == 0
|
||
assert gateway.is_connected is True
|
||
|
||
|
||
async def test_connect_with_nkeys_and_jwt_when_both_set(monkeypatch: pytest.MonkeyPatch, fake: FakeNatsClient):
|
||
# Фиктивная пара ключей создаётся в тесте: настоящих секретов в репозитории нет
|
||
keypair = nkeys.from_seed(nkeys.encode_seed(b"\x01" * 32, nkeys.PREFIX_BYTE_USER))
|
||
seed = keypair.seed.decode()
|
||
settings = make_settings(monkeypatch, OPENBAO__NATS__NATS_SEED=seed, OPENBAO__NATS__NATS_JWT="dummy.jwt.token")
|
||
gateway = NatsGateway(settings, client_factory=lambda: fake)
|
||
|
||
await gateway.connect()
|
||
|
||
kwargs = fake.connect_kwargs
|
||
assert kwargs["user_jwt_cb"]() == b"dummy.jwt.token"
|
||
nonce = "nonce-123"
|
||
signature = base64.b64decode(kwargs["signature_cb"](nonce))
|
||
assert keypair.verify(nonce.encode(), signature) is True
|
||
assert fake.jetstream_calls == 0
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"env",
|
||
[
|
||
{"OPENBAO__NATS__NATS_SEED": "SUAxxx"},
|
||
{"OPENBAO__NATS__NATS_JWT": "dummy.jwt.token"},
|
||
],
|
||
)
|
||
async def test_connect_with_only_one_credential_is_config_error(
|
||
monkeypatch: pytest.MonkeyPatch, fake: FakeNatsClient, env: dict[str, str]
|
||
):
|
||
settings = make_settings(monkeypatch, **env)
|
||
gateway = NatsGateway(settings, client_factory=lambda: fake)
|
||
|
||
with pytest.raises(NatsConfigError, match="NATS_SEED и NATS_JWT"):
|
||
await gateway.connect()
|
||
assert fake.connect_kwargs == {}
|
||
|
||
|
||
async def test_connect_with_unreadable_seed_is_config_error(monkeypatch: pytest.MonkeyPatch, fake: FakeNatsClient):
|
||
settings = make_settings(monkeypatch, OPENBAO__NATS__NATS_SEED="не-seed", OPENBAO__NATS__NATS_JWT="dummy.jwt")
|
||
gateway = NatsGateway(settings, client_factory=lambda: fake)
|
||
|
||
with pytest.raises(NatsConfigError, match="NATS_SEED"):
|
||
await gateway.connect()
|
||
|
||
|
||
async def test_nats_disabled_does_not_connect(monkeypatch: pytest.MonkeyPatch, fake: FakeNatsClient):
|
||
settings = make_settings(monkeypatch, NATS_ENABLED="false")
|
||
created: list[FakeNatsClient] = []
|
||
|
||
def factory() -> FakeNatsClient:
|
||
created.append(fake)
|
||
return fake
|
||
|
||
gateway = NatsGateway(settings, client_factory=factory)
|
||
|
||
assert await gateway.connect() is False
|
||
assert created == []
|
||
assert gateway.is_connected is False
|
||
|
||
|
||
async def test_is_connected_reflects_client_state(gateway: NatsGateway, fake: FakeNatsClient):
|
||
assert gateway.is_connected is False # до connect клиента нет
|
||
await gateway.connect()
|
||
assert gateway.is_connected is True
|
||
fake.is_connected = False # nats-py переподключается
|
||
assert gateway.is_connected is False
|
||
|
||
|
||
async def test_reconnect_callbacks_log_and_keep_state(gateway: NatsGateway, fake: FakeNatsClient):
|
||
await gateway.connect()
|
||
kwargs = fake.connect_kwargs
|
||
|
||
fake.is_connected = False
|
||
await kwargs["disconnected_cb"]()
|
||
await kwargs["error_cb"](nats_errors.StaleConnectionError())
|
||
assert gateway.is_connected is False
|
||
|
||
fake.is_connected = True
|
||
await kwargs["reconnected_cb"]()
|
||
assert gateway.is_connected is True
|
||
await kwargs["closed_cb"]()
|
||
|
||
|
||
async def test_close_drains_connected_client(gateway: NatsGateway, fake: FakeNatsClient):
|
||
await gateway.connect()
|
||
await gateway.close()
|
||
assert fake.drained is True
|
||
assert gateway.is_connected is False
|
||
|
||
|
||
async def test_close_without_connection_is_noop(gateway: NatsGateway, fake: FakeNatsClient):
|
||
await gateway.close()
|
||
assert fake.drained is False
|
||
assert fake.closed is False
|
||
|
||
|
||
async def test_close_falls_back_to_close_when_drain_fails(gateway: NatsGateway, fake: FakeNatsClient):
|
||
await gateway.connect()
|
||
|
||
async def failing_drain() -> None:
|
||
raise nats_errors.ConnectionReconnectingError()
|
||
|
||
fake.drain = failing_drain # type: ignore[method-assign]
|
||
await gateway.close()
|
||
assert fake.closed is True
|
||
|
||
|
||
# --- подписка ---------------------------------------------------------------------------------------------------
|
||
|
||
|
||
async def test_subscribe_requests_uses_image_subject_and_queue_group(gateway: NatsGateway, fake: FakeNatsClient):
|
||
received: list[FakeMsg] = []
|
||
|
||
async def on_request(msg: Any) -> None:
|
||
received.append(msg)
|
||
|
||
await gateway.connect()
|
||
await gateway.subscribe_requests(on_request)
|
||
|
||
assert REQUEST_QUEUE_GROUP == "scan-gateway"
|
||
[subscription] = fake.subscriptions
|
||
assert subscription["subject"] == IMAGE_REQUEST_SUBJECT == "files.img_uploaded"
|
||
assert subscription["queue"] == "scan-gateway"
|
||
assert fake.flush_calls == 1 # подписка подтверждена сервером до возврата
|
||
|
||
msg = FakeMsg(b'{"job_id": "x"}', {NATS_MSG_ID_HEADER: "x:1"})
|
||
await subscription["cb"](msg)
|
||
assert received == [msg] # обработчик получает исходное сообщение nats-py (data, headers)
|
||
|
||
|
||
async def test_subscribe_requests_handler_exception_is_logged_not_raised(gateway: NatsGateway, fake: FakeNatsClient):
|
||
async def broken(msg: Any) -> None:
|
||
raise RuntimeError("сбой обработчика")
|
||
|
||
await gateway.connect()
|
||
await gateway.subscribe_requests(broken)
|
||
await fake.subscriptions[0]["cb"](FakeMsg()) # не бросает: подписка живёт
|
||
|
||
|
||
async def test_subscribe_requests_requires_connection(gateway: NatsGateway):
|
||
async def on_request(msg: Any) -> None:
|
||
return None
|
||
|
||
with pytest.raises(NatsNotConnectedError):
|
||
await gateway.subscribe_requests(on_request)
|
||
|
||
|
||
# --- публикация -------------------------------------------------------------------------------------------------
|
||
|
||
|
||
async def test_publish_sets_msg_id_header_and_flushes(gateway: NatsGateway, fake: FakeNatsClient):
|
||
await gateway.connect()
|
||
await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1")
|
||
|
||
assert fake.published == [(RESULT_SUBJECT, b"payload", {"Nats-Msg-Id": "result:job:1"})]
|
||
assert fake.flush_calls == 1
|
||
|
||
|
||
async def test_publish_retries_transient_errors_until_success(
|
||
gateway: NatsGateway, fake: FakeNatsClient, sleeps: list[float]
|
||
):
|
||
await gateway.connect()
|
||
fake.publish_errors = [nats_errors.ConnectionReconnectingError(), nats_errors.FlushTimeoutError()]
|
||
|
||
await gateway.publish(RESULT_SUBJECT, b"payload", "ack:job:1")
|
||
|
||
assert len(fake.published) == 3 # первая попытка + два повтора из PUBLISH_RETRIES=3
|
||
assert all(headers == {"Nats-Msg-Id": "ack:job:1"} for _, _, headers in fake.published)
|
||
assert sleeps == [0, 0] # пауза перед каждым повтором
|
||
|
||
|
||
async def test_publish_succeeds_on_last_retry(gateway: NatsGateway, fake: FakeNatsClient, sleeps: list[float]):
|
||
await gateway.connect()
|
||
fake.publish_errors = [nats_errors.ConnectionClosedError()] * 3
|
||
|
||
await gateway.publish(RESULT_SUBJECT, b"payload", "ack:job:1")
|
||
|
||
assert len(fake.published) == 4 # 1 + 3 повтора: последний повтор успешен
|
||
assert sleeps == [0, 0, 0]
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"error",
|
||
[
|
||
nats_errors.ConnectionClosedError(),
|
||
nats_errors.ConnectionReconnectingError(),
|
||
nats_errors.OutboundBufferLimitError(),
|
||
nats_errors.StaleConnectionError(),
|
||
nats_errors.TimeoutError(),
|
||
TimeoutError(),
|
||
ConnectionResetError(),
|
||
],
|
||
)
|
||
async def test_publish_raises_publish_failed_after_initial_attempt_and_all_retries(
|
||
gateway: NatsGateway, fake: FakeNatsClient, error: BaseException
|
||
):
|
||
await gateway.connect()
|
||
fake.publish_errors = [error] * 10
|
||
|
||
with pytest.raises(PublishFailed) as exc_info:
|
||
await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1")
|
||
|
||
assert len(fake.published) == 4 # PUBLISH_RETRIES=3: первая попытка + 3 повтора
|
||
assert exc_info.value.attempts == 4
|
||
assert exc_info.value.msg_id == "result:job:1"
|
||
assert exc_info.value.__cause__ is error
|
||
|
||
|
||
async def test_publish_retries_parameter_overrides_settings(gateway: NatsGateway, fake: FakeNatsClient):
|
||
await gateway.connect()
|
||
fake.publish_errors = [nats_errors.ConnectionClosedError()] * 10
|
||
|
||
with pytest.raises(PublishFailed):
|
||
await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1", retries=5)
|
||
assert len(fake.published) == 6
|
||
|
||
|
||
async def test_publish_zero_retries_makes_exactly_one_attempt(
|
||
gateway: NatsGateway, fake: FakeNatsClient, sleeps: list[float]
|
||
):
|
||
await gateway.connect()
|
||
fake.publish_errors = [nats_errors.ConnectionClosedError()] * 10
|
||
|
||
with pytest.raises(PublishFailed) as exc_info:
|
||
await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1", retries=0)
|
||
assert len(fake.published) == 1
|
||
assert exc_info.value.attempts == 1
|
||
assert sleeps == []
|
||
|
||
|
||
async def test_publish_uses_publish_retries_from_settings(monkeypatch: pytest.MonkeyPatch, fake: FakeNatsClient):
|
||
settings = make_settings(monkeypatch, PUBLISH_RETRIES="2")
|
||
gateway = NatsGateway(settings, client_factory=lambda: fake, retry_delay=0)
|
||
await gateway.connect()
|
||
fake.publish_errors = [nats_errors.ConnectionClosedError()] * 10
|
||
|
||
with pytest.raises(PublishFailed):
|
||
await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1")
|
||
assert len(fake.published) == 3 # PUBLISH_RETRIES=2: 1 + 2 повтора
|
||
|
||
|
||
@pytest.mark.parametrize("error", [nats_errors.MaxPayloadError(), nats_errors.BadSubjectError(), ValueError("x")])
|
||
async def test_publish_does_not_retry_non_transient_errors(
|
||
gateway: NatsGateway, fake: FakeNatsClient, error: BaseException
|
||
):
|
||
await gateway.connect()
|
||
fake.publish_errors = [error] * 10
|
||
|
||
with pytest.raises(PublishFailed) as exc_info:
|
||
await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1")
|
||
assert len(fake.published) == 1
|
||
assert exc_info.value.attempts == 1
|
||
assert exc_info.value.__cause__ is error
|
||
|
||
|
||
async def test_publish_without_connection_fails_without_attempts(gateway: NatsGateway, fake: FakeNatsClient):
|
||
with pytest.raises(PublishFailed) as exc_info:
|
||
await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1")
|
||
assert exc_info.value.attempts == 0
|
||
assert fake.published == []
|
||
|
||
|
||
async def test_publish_rejects_negative_retries(gateway: NatsGateway):
|
||
await gateway.connect()
|
||
with pytest.raises(ValueError, match="retries"):
|
||
await gateway.publish(RESULT_SUBJECT, b"payload", "result:job:1", retries=-1)
|
||
|
||
|
||
# --- публикация конверта ----------------------------------------------------------------------------------------
|
||
|
||
SERVICE = ServiceInfo(name="scan-gateway", version="1.0.0")
|
||
|
||
|
||
async def test_publish_result_ack_uses_ack_msg_id(gateway: NatsGateway, fake: FakeNatsClient):
|
||
job_id, file_id = uuid.uuid4(), uuid.uuid4()
|
||
message = build_ack_message(job_id, file_id, SERVICE)
|
||
await gateway.connect()
|
||
|
||
await gateway.publish_result(message, attempt=2)
|
||
|
||
[(subject, payload, headers)] = fake.published
|
||
assert subject == RESULT_SUBJECT
|
||
assert headers == {"Nats-Msg-Id": f"ack:{job_id}:2"}
|
||
assert json.loads(payload) == message.to_wire()
|
||
|
||
|
||
async def test_publish_result_final_uses_result_msg_id_and_utf8(gateway: NatsGateway, fake: FakeNatsClient):
|
||
job_id, file_id = uuid.uuid4(), uuid.uuid4()
|
||
message = build_error_message(job_id, file_id, SERVICE, ErrorCode.TIMEOUT, "ИС не ответила")
|
||
await gateway.connect()
|
||
|
||
await gateway.publish_result(message, attempt=1)
|
||
|
||
[(_, payload, headers)] = fake.published
|
||
assert headers == {"Nats-Msg-Id": f"result:{job_id}:1"}
|
||
assert "ИС не ответила".encode() in payload # UTF-8 без \u-экранирования
|
||
assert json.loads(payload) == message.to_wire()
|