arm-tabular-parser/api/tests/integration/test_nats_integration.py

"""Интеграционные тесты `src/core/nats.py` против compose-NATS без JetStream (`.env.test`: nats://localhost:4222;
spec «Соединение с NATS по правилам шаблона, без JetStream»; design D-6).

`.env.test` содержит `NATS_ENABLED=false` (для приложения) — тесты строят соединение сами. Subject'ы уникальны на
тест (`test.tabular.<uuid>`): общие `files.*` не трогаем, на шине может работать другой сервис.
"""

import asyncio
import time
import uuid
from collections import Counter
from typing import Any

import pytest
from nats.aio.client import Client as NATS
from nats.aio.msg import Msg

from src.contract.subjects import NATS_MSG_ID_HEADER
from src.core.config import NATSSettings, get_settings
from src.core.nats import REQUEST_QUEUE_GROUP, NatsConnection, NatsPublishError

pytestmark = pytest.mark.integration

READY_TIMEOUT = 5.0
START_MAX_SECONDS = 0.5  # start() не ждёт подключения
STOP_MAX_SECONDS = 2.0


def _nats_settings() -> NATSSettings:
    return get_settings().OPENBAO.NATS


def _unique_subject() -> str:
    return f"test.tabular.{uuid.uuid4().hex}"


async def _wait_until(predicate: Any, timeout: float = 5.0) -> None:
    deadline = time.monotonic() + timeout
    while not predicate():
        if time.monotonic() > deadline:
            raise AssertionError("условие не выполнено за отведённое время")
        await asyncio.sleep(0.02)


async def _start(conn: NatsConnection) -> NatsConnection:
    await conn.start()
    if not await conn.wait_ready(timeout=READY_TIMEOUT):
        await conn.stop()
        pytest.fail(f"NATS недоступен ({_nats_settings().NATS_URL}, .env.test; make up)", pytrace=False)
    return conn


async def _plain_client() -> NATS:
    nc = NATS()
    await nc.connect(_nats_settings().NATS_URL, connect_timeout=2, max_reconnect_attempts=1)
    return nc


async def test_subscribe_requests_receives_message_with_msg_id_header():
    """Запрос ядра (сторонний клиент с `Nats-Msg-Id`) доходит до подписки сервиса в queue group с заголовком."""
    subject = _unique_subject()
    received: list[Msg] = []

    async def handler(msg: Msg) -> None:
        received.append(msg)

    async def on_connected(conn: NatsConnection) -> None:
        await conn.subscribe_requests(handler, subject=subject)

    conn = await _start(NatsConnection(_nats_settings(), on_connected=on_connected))
    core = await _plain_client()
    try:
        assert conn.is_connected
        await core.publish(subject, b'{"job_id": "j-1"}', headers={NATS_MSG_ID_HEADER: "j-1:1"})
        await core.flush()
        await _wait_until(lambda: len(received) == 1)
    finally:
        await core.close()
        await conn.stop()

    msg = received[0]
    assert msg.subject == subject
    assert msg.data == b'{"job_id": "j-1"}'
    assert msg.headers == {NATS_MSG_ID_HEADER: "j-1:1"}


async def test_publish_delivers_payload_with_msg_id_header():
    """Ack/результат сервиса доходит до подписчика (как ядро) с заголовком `Nats-Msg-Id`."""
    subject = _unique_subject()
    core = await _plain_client()
    subscription = await core.subscribe(subject)
    conn = await _start(NatsConnection(_nats_settings()))
    try:
        await conn.publish(subject, b'{"status": "processing"}', "ack:j-1:1")
        msg = await subscription.next_msg(timeout=5)
    finally:
        await conn.stop()
        await core.close()

    assert msg.data == b'{"status": "processing"}'
    assert msg.headers == {NATS_MSG_ID_HEADER: "ack:j-1:1"}


async def test_queue_group_delivers_each_request_to_exactly_one_instance():
    subject = _unique_subject()
    received: dict[str, list[str]] = {"a": [], "b": []}

    def make_on_connected(name: str) -> Any:
        async def handler(msg: Msg) -> None:
            assert msg.headers is not None
            received[name].append(msg.headers[NATS_MSG_ID_HEADER])

        async def on_connected(conn: NatsConnection) -> None:
            await conn.subscribe_requests(handler, subject=subject)

        return on_connected

    instance_a = await _start(NatsConnection(_nats_settings(), on_connected=make_on_connected("a")))
    instance_b = await _start(NatsConnection(_nats_settings(), on_connected=make_on_connected("b")))
    total = 20
    try:
        for i in range(total):
            await instance_a.publish(subject, b"{}", f"job-{i}:1")
        await _wait_until(lambda: len(received["a"]) + len(received["b"]) >= total)
        await asyncio.sleep(0.2)  # дубль, если бы он был, успел бы прийти
    finally:
        await instance_a.stop()
        await instance_b.stop()

    assert REQUEST_QUEUE_GROUP == "tabular-parser"
    all_ids = received["a"] + received["b"]
    assert Counter(all_ids) == Counter(f"job-{i}:1" for i in range(total))


async def test_start_against_unreachable_server_does_not_block_and_stops_cleanly():
    """O-3: NATS недоступен при старте — start() не ждёт, публикация после повторов падает, stop() быстрый."""
    conn = NatsConnection(NATSSettings(NATS_URL="nats://127.0.0.1:54999"), publish_retries=2, publish_retry_pause=0.05)

    started = time.monotonic()
    await conn.start()
    assert time.monotonic() - started < START_MAX_SECONDS
    await asyncio.sleep(0.3)
    assert not conn.is_connected
    with pytest.raises(NatsPublishError):
        await conn.publish("test.tabular.unreachable", b"{}", "m:1")

    started = time.monotonic()
    await conn.stop()
    assert time.monotonic() - started < STOP_MAX_SECONDS