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

"""Интеграционный round-trip сервиса против compose-NATS (без JetStream) и compose-MinIO (`.env.test`).

Приложение поднимается в процессе теста через lifespan (`src.app`) с `NATS_ENABLED=True` — тот же код, что в проде:
`_connect_s3`, `NatsConnection` с подпиской в queue group `tabular-parser`, `RecognitionWorker` с парсером по умолчанию
(`get_parser(settings)` → `OsvParser`); исходник — битый zip, поэтому результат — ошибка `UNREADABLE`.
Тест играет роль ядра: кладёт исходник в MinIO `uploads/…`, публикует запрос в `recognition.tabular.request` с
`Nats-Msg-Id = <job_id>:<attempt>` и собирает конверты из `recognition.result`.

Spec tabular-recognition-service: «Ack раньше результата», «Повторный запрос во время обработки» (здесь — после
обработки: битый файл отвергается мгновенно; повтор во время обработки покрыт unit-тестами воркера), ошибка разбора
конвертом (бывш. «Запрос с заглушкой парсера»), «Нет объекта в S3», «Несовместимая версия».

Изоляция: `job_id`/`file_id`/ключ S3 уникальны на прогон (uuid4), исходник удаляется после теста; на шине может
работать другой сервис — сообщения `recognition.result` фильтруются по своему `job_id`. Ожидания — по событиям с
таймаутом; единственная фиксированная пауза — явное окно тишины `QUIET_WINDOW` для отрицательных проверок.
"""

import asyncio
import contextlib
import copy
import importlib
import io
import json
import uuid
from collections.abc import AsyncIterator
from pathlib import Path
from typing import Any

import pytest
from jsonschema import Draft202012Validator, FormatChecker
from miniopy_async import Minio
from nats.aio.client import Client as NATS
from nats.aio.msg import Msg

from src.contract import NATS_MSG_ID_HEADER, RESULT_SUBJECT, TABULAR_REQUEST_SUBJECT, request_msg_id
from src.core.config import get_settings

pytestmark = pytest.mark.integration

CONTRACTS_DIR = Path(__file__).resolve().parents[2] / "contracts"
EXAMPLE_REQUEST = json.loads((CONTRACTS_DIR / "examples" / "request-tabular.json").read_text(encoding="utf-8"))
MESSAGE_VALIDATOR = Draft202012Validator(
    json.loads((CONTRACTS_DIR / "recognition-result-message.schema.json").read_text(encoding="utf-8")),
    format_checker=FormatChecker(),
)

SERVICE_NAME = "tabular-parser"
READY_TIMEOUT = 5.0  # подключение к NATS и подписка приложения
MESSAGE_TIMEOUT = 10.0  # одно сообщение сервиса (с запасом на повторы публикации)
# Окно тишины для отрицательных проверок («второго результата нет», «ack нет»): битый файл отвергается за миллисекунды,
# поэтому лишнее сообщение, если бы оно было, пришло бы гораздо раньше конца окна
QUIET_WINDOW = 1.5
SOURCE_BYTES = b"PK\x03\x04 round-trip: broken zip, the default parser answers UNREADABLE"
XLSX_CONTENT_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"

app_module = importlib.import_module("src.app")  # модуль, а не объект `app`, который реэкспортирует пакет `src`


# --- вспомогательное --------------------------------------------------------------------------------------------------
def _new_request(*, key_name: str | None = None, schema_version: str | None = None) -> dict[str, Any]:
    """Запрос из `contracts/examples/request-tabular.json` с уникальными `job_id`/`file_id`/`s3.key`."""
    request = copy.deepcopy(EXAMPLE_REQUEST)
    request["job_id"] = str(uuid.uuid4())
    request["file_id"] = str(uuid.uuid4())
    name = key_name or request["original_name"]
    request["s3"]["key"] = f"uploads/{request['file_id']}/{request['job_id']}/{name}"
    if schema_version is not None:
        request["schema_version"] = schema_version
    return request


def _describe(msg: Msg) -> str:
    return f"Nats-Msg-Id={(msg.headers or {}).get(NATS_MSG_ID_HEADER)!r}, body={msg.data.decode('utf-8', 'replace')}"


def _envelope(msg: Msg, request: dict[str, Any], msg_id: str) -> dict[str, Any]:
    """Общие проверки конверта: заголовок, схема recognition-result-message, `job_id`/`file_id`, `service.name`."""
    assert msg.headers is not None and msg.headers.get(NATS_MSG_ID_HEADER) == msg_id, _describe(msg)
    body: dict[str, Any] = json.loads(msg.data)
    errors = [f"{list(e.absolute_path)}: {e.message}" for e in MESSAGE_VALIDATOR.iter_errors(body)]
    assert errors == [], f"конверт не проходит recognition-result-message.schema.json: {errors}; {_describe(msg)}"
    assert body["job_id"] == request["job_id"]
    assert body["file_id"] == request["file_id"]
    assert body["service"]["name"] == SERVICE_NAME
    return body


def _assert_error(msg: Msg, request: dict[str, Any], code: str) -> dict[str, Any]:
    body = _envelope(msg, request, f"result:{request['job_id']}:{request['attempt']}")
    assert body["status"] == "error", _describe(msg)
    assert body["error"]["code"] == code, _describe(msg)
    assert body["metrics"]["duration_ms"] >= 0
    return body


def _assert_ack(msg: Msg, request: dict[str, Any]) -> None:
    body = _envelope(msg, request, f"ack:{request['job_id']}:{request['attempt']}")
    assert body["status"] == "processing", _describe(msg)


class Core:
    """Тестовый клиент в роли ядра: публикует запросы и собирает `recognition.result` по своим `job_id`."""

    def __init__(self, client: NATS) -> None:
        self.client = client
        self._inboxes: dict[str, asyncio.Queue[Msg]] = {}

    async def on_result(self, msg: Msg) -> None:
        try:
            job_id = json.loads(msg.data).get("job_id")
        except (ValueError, AttributeError):
            return  # чужое сообщение на общей шине
        inbox = self._inboxes.get(job_id) if isinstance(job_id, str) else None
        if inbox is not None:
            inbox.put_nowait(msg)

    async def publish(self, request: dict[str, Any]) -> None:
        self._inboxes.setdefault(request["job_id"], asyncio.Queue())
        payload = json.dumps(request, ensure_ascii=False).encode("utf-8")
        headers = {NATS_MSG_ID_HEADER: request_msg_id(request["job_id"], request["attempt"])}
        await self.client.publish(TABULAR_REQUEST_SUBJECT, payload, headers=headers)
        await self.client.flush()

    async def next_message(self, request: dict[str, Any]) -> Msg:
        try:
            return await asyncio.wait_for(self._inboxes[request["job_id"]].get(), MESSAGE_TIMEOUT)
        except TimeoutError:
            pytest.fail(
                f"нет сообщения recognition.result для job_id={request['job_id']} за {MESSAGE_TIMEOUT} с "
                "(сервис не ответил; если на шине есть другой инстанс tabular-parser, запрос мог уйти ему)",
                pytrace=False,
            )

    async def assert_quiet(self, request: dict[str, Any]) -> None:
        """Отрицательная проверка: в окне `QUIET_WINDOW` по этому `job_id` больше ничего не приходит."""
        try:
            extra = await asyncio.wait_for(self._inboxes[request["job_id"]].get(), QUIET_WINDOW)
        except TimeoutError:
            return
        pytest.fail(f"лишнее сообщение по job_id={request['job_id']}: {_describe(extra)}", pytrace=False)


# --- фикстуры ---------------------------------------------------------------------------------------------------------
class Uploads:
    """Исходники в MinIO `uploads/…` с уникальными ключами; удаляются после теста."""

    def __init__(self, client: Minio) -> None:
        self.client = client
        self.keys: list[tuple[str, str]] = []

    async def put(self, request: dict[str, Any]) -> None:
        bucket, key = request["s3"]["bucket"], request["s3"]["key"]
        data = io.BytesIO(SOURCE_BYTES)
        await self.client.put_object(bucket, key, data, len(SOURCE_BYTES), content_type=XLSX_CONTENT_TYPE)
        self.keys.append((bucket, key))


@pytest.fixture
async def uploads() -> AsyncIterator[Uploads]:
    minio = get_settings().OPENBAO.MINIO
    client = Minio(
        minio.MINIO_URL,
        access_key=minio.MINIO_ADMIN_LOGIN,
        secret_key=minio.MINIO_ADMIN_PASSWORD,
        secure=minio.MINIO_SECURE,
    )
    store = Uploads(client)
    try:
        yield store
    finally:
        try:
            for bucket, key in store.keys:
                # сбой удаления одного объекта не должен оставлять в MinIO остальные
                with contextlib.suppress(Exception):
                    await client.remove_object(bucket, key)
        finally:
            await client.close_session()


@pytest.fixture
async def service(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[None]:
    """Приложение в процессе теста: lifespan с `NATS_ENABLED=True` (в `.env.test` — false); выход — остановка."""
    monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True)
    application = app_module.app
    tasks: list[asyncio.Task[None]] = []
    async with application.router.lifespan_context(application):
        if not await application.state.nats.wait_ready(READY_TIMEOUT):
            pytest.fail(
                f"NATS недоступен ({app_module.settings.OPENBAO.NATS.NATS_URL}, .env.test; make up)", pytrace=False
            )
        tasks = list(application.state.worker._tasks)
        yield
    # `_tasks` обнуляется в `stop()` безусловно — проверяем сами задачи, снятые до выхода из lifespan
    assert tasks, "воркер не запустил рабочие корутины"
    assert all(t.done() for t in tasks), "рабочие корутины воркера не остановлены"


@pytest.fixture
async def core(service: None) -> AsyncIterator[Core]:
    """Клиент «ядра», подписанный на `recognition.result` до первой публикации запроса."""
    client = NATS()
    await client.connect(get_settings().OPENBAO.NATS.NATS_URL, connect_timeout=2, max_reconnect_attempts=1)
    core = Core(client)
    await client.subscribe(RESULT_SUBJECT, cb=core.on_result)
    await client.flush()
    try:
        yield core
    finally:
        await client.close()


# --- сценарии ---------------------------------------------------------------------------------------------------------
async def test_request_gets_ack_then_parser_error_and_duplicate_gets_only_ack(core: Core, uploads: Uploads) -> None:
    """«Ack раньше результата», ошибка разбора (`UNREADABLE`) конвертом, повтор запроса — только ещё один ack."""
    request = _new_request()
    await uploads.put(request)

    await core.publish(request)
    _assert_ack(await core.next_message(request), request)
    _assert_error(await core.next_message(request), request, "UNREADABLE")

    await core.publish(request)  # та же доставка ещё раз: тот же `Nats-Msg-Id = <job_id>:<attempt>`
    _assert_ack(await core.next_message(request), request)
    await core.assert_quiet(request)  # второго результата нет


async def test_missing_source_object_gives_file_read_error_and_service_keeps_working(
    core: Core, uploads: Uploads
) -> None:
    """«Нет объекта в S3»: ack, затем `FILE_READ_ERROR`; следующий запрос обрабатывается как обычно."""
    missing = _new_request(key_name="missing.xlsx")  # объект не кладётся

    await core.publish(missing)
    _assert_ack(await core.next_message(missing), missing)
    body = _assert_error(await core.next_message(missing), missing, "FILE_READ_ERROR")
    assert missing["s3"]["key"] in body["error"]["message"]

    following = _new_request()
    await uploads.put(following)
    await core.publish(following)
    _assert_ack(await core.next_message(following), following)
    _assert_error(await core.next_message(following), following, "UNREADABLE")


async def test_unsupported_schema_version_gives_single_schema_mismatch_without_ack(
    core: Core, uploads: Uploads
) -> None:
    """«Несовместимая версия»: `schema_version = 2.0` → один конверт `SCHEMA_MISMATCH`, ack не публикуется."""
    request = _new_request(schema_version="2.0")
    await uploads.put(request)

    await core.publish(request)
    # Ack публикуется до результата тем же соединением: если бы он был, он пришёл бы первым
    body = _assert_error(await core.next_message(request), request, "SCHEMA_MISMATCH")
    assert "2.0" in body["error"]["message"]
    await core.assert_quiet(request)