arm-scan-gateway/api/tests/integration/test_nats_minio_integration.py

"""Интеграция `src.core.nats` и `src.core.clients` с compose-инфраструктурой (`make up`): NATS без JetStream на
127.0.0.1:4222, MinIO на 127.0.0.1:9000 (бакет `arm-ka`); параметры — `api/.env.test`.

Недоступная инфраструктура — падение с понятным сообщением, не skip. Общий контейнер NATS не перезапускается:
сценарий переподключения покрыт unit-тестом колбэков (tests/unit/test_nats.py). Чужой трафик на тех же subject'ах
отсекается уникальным `job_id` теста; ключи MinIO — под уникальным префиксом `itest/<uuid>/` и удаляются.
"""

import asyncio
import json
import socket
import uuid
from collections.abc import AsyncIterator
from typing import Any
from urllib.parse import urlparse

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

from src.contract import (
    IMAGE_REQUEST_SUBJECT,
    NATS_MSG_ID_HEADER,
    RESULT_SUBJECT,
    ServiceInfo,
    build_ack_message,
)
from src.core.clients import S3ObjectNotFoundError, S3Storage
from src.core.config import Settings, get_settings
from src.core.nats import REQUEST_QUEUE_GROUP, NatsGateway

pytestmark = pytest.mark.integration

CONNECT_WAIT = 5.0  # секунд на подключение к NATS (nats-py ждёт сервер бесконечно — ограничиваем снаружи)
DELIVERY_WAIT = 5.0


def _require_port(host: str, port: int, what: str) -> None:
    try:
        with socket.create_connection((host, port), timeout=2):
            return
    except OSError as exc:
        pytest.fail(f"{what} недоступен на {host}:{port} ({exc!r}): поднимите инфраструктуру `make up`")


@pytest.fixture
def settings() -> Settings:
    settings = get_settings()
    nats_url = urlparse(settings.OPENBAO.NATS.NATS_URL)
    _require_port(nats_url.hostname or "127.0.0.1", nats_url.port or 4222, "NATS")
    host, _, port = settings.OPENBAO.MINIO.MINIO_URL.partition(":")
    _require_port(host, int(port or 9000), "MinIO")
    return settings


def _no_jetstream_client() -> NATS:
    """Настоящий клиент nats-py, у которого вызов `jetstream()` — ошибка теста (D-8: JetStream не используется)."""
    client = NATS()

    def forbidden(**_: Any) -> None:
        raise AssertionError("JetStream не используется (D-8)")

    client.jetstream = forbidden  # type: ignore[method-assign]
    return client


async def _connected_gateway(settings: Settings) -> NatsGateway:
    gateway = NatsGateway(settings, client_factory=_no_jetstream_client, retry_delay=0.1)
    try:
        connected = await asyncio.wait_for(gateway.connect(), CONNECT_WAIT)
    except TimeoutError:
        await gateway.close()
        pytest.fail(f"Нет подключения к NATS {settings.OPENBAO.NATS.NATS_URL} за {CONNECT_WAIT} с: `make up`")
    assert connected is True
    assert gateway.is_connected is True
    return gateway


@pytest.fixture
async def gateways(settings: Settings) -> AsyncIterator[list[NatsGateway]]:
    created: list[NatsGateway] = []
    yield created
    for gateway in created:
        await gateway.close()


async def _wait_until(predicate: Any, timeout: float = DELIVERY_WAIT) -> None:
    loop = asyncio.get_running_loop()
    deadline = loop.time() + timeout
    while not predicate() and loop.time() < deadline:
        await asyncio.sleep(0.05)


# --- NATS ---------------------------------------------------------------------------------------------------------


async def test_subscribe_and_publish_with_msg_id_without_jetstream(settings: Settings, gateways: list[NatsGateway]):
    subscriber = await _connected_gateway(settings)
    publisher = await _connected_gateway(settings)
    gateways += [subscriber, publisher]
    job_id = str(uuid.uuid4())
    received: list[Msg] = []

    async def on_request(msg: Msg) -> None:
        if json.loads(msg.data).get("job_id") == job_id:  # чужие сообщения на общем subject пропускаем
            received.append(msg)

    await subscriber.subscribe_requests(on_request)
    await publisher.publish(IMAGE_REQUEST_SUBJECT, json.dumps({"job_id": job_id}).encode(), f"{job_id}:1")
    await _wait_until(lambda: len(received) >= 1)

    assert len(received) == 1
    [msg] = received
    assert msg.subject == IMAGE_REQUEST_SUBJECT
    assert msg.headers is not None
    assert msg.headers[NATS_MSG_ID_HEADER] == f"{job_id}:1"


async def test_queue_group_delivers_each_message_to_exactly_one_member(settings: Settings, gateways: list[NatsGateway]):
    members = [await _connected_gateway(settings), await _connected_gateway(settings)]
    publisher = await _connected_gateway(settings)
    gateways += [*members, publisher]
    run_id = str(uuid.uuid4())
    deliveries: list[tuple[int, str]] = []

    def collector(member: int) -> Any:
        async def on_request(msg: Msg) -> None:
            body = json.loads(msg.data)
            if body.get("run_id") == run_id:
                assert msg.headers is not None
                deliveries.append((member, msg.headers[NATS_MSG_ID_HEADER]))

        return on_request

    for index, member in enumerate(members):
        await member.subscribe_requests(collector(index), queue=REQUEST_QUEUE_GROUP)

    total = 40
    sent = {f"{run_id}-{n}:1" for n in range(total)}
    for msg_id in sorted(sent):
        await publisher.publish(IMAGE_REQUEST_SUBJECT, json.dumps({"run_id": run_id}).encode(), msg_id)
    await _wait_until(lambda: len(deliveries) >= total)
    await asyncio.sleep(0.3)  # лишние (дублирующие) доставки успели бы прийти

    delivered_ids = [msg_id for _, msg_id in deliveries]
    assert len(deliveries) == total
    assert set(delivered_ids) == sent  # каждое сообщение — ровно одному участнику группы, без дублей
    assert {member for member, _ in deliveries} == {0, 1}  # сообщения распределены между обоими участниками


async def test_publish_result_reaches_result_subject_with_ack_msg_id(settings: Settings, gateways: list[NatsGateway]):
    gateway = await _connected_gateway(settings)
    gateways.append(gateway)
    listener = NATS()
    await asyncio.wait_for(listener.connect(settings.OPENBAO.NATS.NATS_URL), CONNECT_WAIT)
    job_id, file_id = uuid.uuid4(), uuid.uuid4()
    received: list[Msg] = []

    async def on_result(msg: Msg) -> None:
        if json.loads(msg.data).get("job_id") == str(job_id):
            received.append(msg)

    try:
        await listener.subscribe(RESULT_SUBJECT, cb=on_result)
        await listener.flush()
        message = build_ack_message(job_id, file_id, ServiceInfo(name="scan-gateway", version="1.0.0"))
        await gateway.publish_result(message, attempt=1)
        await _wait_until(lambda: len(received) >= 1)
    finally:
        await listener.drain()

    assert len(received) == 1
    assert received[0].headers is not None
    assert received[0].headers[NATS_MSG_ID_HEADER] == f"ack:{job_id}:1"
    assert json.loads(received[0].data) == message.to_wire()


# --- S3 -----------------------------------------------------------------------------------------------------------


@pytest.fixture
async def storage(settings: Settings) -> AsyncIterator[tuple[S3Storage, str, list[str]]]:
    storage = S3Storage.from_settings(settings.OPENBAO.MINIO)
    prefix = f"itest/{uuid.uuid4()}"
    written: list[str] = []
    yield storage, prefix, written
    for key in written:
        await storage.client.remove_object(storage.bucket, key)
    await storage.close()


async def test_s3_put_and_get_roundtrip(storage: tuple[S3Storage, str, list[str]]):
    s3, prefix, written = storage
    json_key, bytes_key = f"{prefix}/raw.json", f"{prefix}/file.pdf"
    written += [json_key, bytes_key]
    payload = {"error": None, "message": "Бухгалтерский баланс", "n": 1}

    await s3.put_json(json_key, payload)
    await s3.put_bytes(bytes_key, b"%PDF-1.7 \x00\xff", content_type="application/pdf")

    raw = await s3.get_bytes(json_key)
    assert "Бухгалтерский баланс".encode() in raw  # UTF-8 без экранирования
    assert json.loads(raw) == payload
    assert await s3.get_bytes(bytes_key, bucket=s3.bucket) == b"%PDF-1.7 \x00\xff"
    stat = await s3.client.stat_object(s3.bucket, json_key)
    assert stat.content_type == "application/json"


async def test_s3_get_missing_key_raises_not_found(storage: tuple[S3Storage, str, list[str]]):
    s3, prefix, _ = storage
    with pytest.raises(S3ObjectNotFoundError) as exc_info:
        await s3.get_bytes(f"{prefix}/missing.pdf")
    assert exc_info.value.key == f"{prefix}/missing.pdf"