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/…`, публикует запрос на subject запросов прогона с
`Nats-Msg-Id = <job_id>:<attempt>` и собирает конверты с subject'а ответа прогона.

Spec tabular-recognition-service: «Ack раньше результата», «Повторный запрос во время обработки» (здесь — после
обработки: битый файл отвергается мгновенно; повтор во время обработки покрыт unit-тестами воркера), ошибка разбора
конвертом (бывш. «Запрос с заглушкой парсера»), «Нет объекта в S3», «Несовместимая версия»,
«Реальная ОСВ через воркер» (маркер `fixtures`: реальный XLS из `../fixtures` → `done` и файл результата в MinIO),
«Реальные формы через воркер» (книга НПЗ → `done`, документы BALANCE и INCOME, `result_schema_version 2.0`).

Изоляция: subject'ы прогона уникальны — контейнер стенда сообщения теста не получает (OQ-83);
`job_id`/`file_id`/ключ S3 уникальны на прогон (uuid4), исходник удаляется после теста; на шине может
работать другой сервис — сообщения с subject'а ответа фильтруются по своему `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 result_contract import schema_errors

from src.contract import (
    NATS_MSG_ID_HEADER,
    RecognitionResult,
    request_msg_id,
)
from src.core.config import get_settings

pytestmark = pytest.mark.integration

REAL_OSV_66 = Path("osv") / "rb" / "2024-FY" / "ОСВ_66 за 2024.xls"
REAL_NPZ = (
    Path("statements-xlsx")
    / "npz"
    / "2025-Q3"
    / "Бухгалтерская отчетность за Январь - Сентябрь 2025 г. (НОВОРОССИЙСКИЙ ПРОКАТНЫЙ ЗАВОД).xlsx"
)

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"
XLS_CONTENT_TYPE = "application/vnd.ms-excel"

# Subject'ы прогона уникальны (OQ-83): контейнер tabular-parser стенда слушает умолчания и сообщения теста не получает
RUN = uuid.uuid4().hex[:8]
TEST_SUBJECTS = {
    "NATS_TABULAR_REQUEST_SUBJECT": f"test-{RUN}.files.tabular_uploaded",
    "NATS_IMAGE_REQUEST_SUBJECT": f"test-{RUN}.files.img_uploaded",
    "NATS_RESULT_SUBJECT": f"test-{RUN}.files.tabular_parsed",
}

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:
    """Тестовый клиент в роли ядра: публикует запросы и собирает сообщения с subject'а ответа по своим `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(TEST_SUBJECTS["NATS_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"нет сообщения на subject ответа прогона для 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]] = []
        self.buckets: set[str] = set()

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

    async def _ensure_bucket(self, bucket: str) -> None:
        """`NoSuchBucket` (OQ-80): на чистом MinIO бакета нет — тест создаёт его сам, compose не трогаем."""
        if bucket not in self.buckets:
            if not await self.client.bucket_exists(bucket):
                await self.client.make_bucket(bucket)
            self.buckets.add(bucket)

    async def read(self, bucket: str, key: str) -> bytes:
        response = await self.client.get_object(bucket, key)
        try:
            return await response.read()
        finally:
            response.release()


@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)
    nats_settings = app_module.settings.OPENBAO.NATS
    monkeypatch.setattr(app_module.settings.OPENBAO, "NATS", nats_settings.model_copy(update=TEST_SUBJECTS))
    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]:
    """Клиент «ядра», подписанный на subject ответа прогона до первой публикации запроса."""
    client = NATS()
    await client.connect(get_settings().OPENBAO.NATS.NATS_URL, connect_timeout=2, max_reconnect_attempts=1)
    core = Core(client)
    await client.subscribe(TEST_SUBJECTS["NATS_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)


@pytest.mark.fixtures
async def test_real_osv_through_worker_gives_done_and_valid_result_file(
    core: Core, uploads: Uploads, fixtures_dir: Path
) -> None:
    """«Реальная ОСВ через воркер»: реальный XLS в MinIO → ack → `done`, `result_schema_version 2.0`, файл по схеме."""
    request = _new_request(key_name="osv_66_2024.xls")
    request["file_type"] = "XLS"
    await uploads.put(request, (fixtures_dir / REAL_OSV_66).read_bytes(), XLS_CONTENT_TYPE)

    await core.publish(request)
    _assert_ack(await core.next_message(request), request)
    body = _envelope(await core.next_message(request), request, f"result:{request['job_id']}:{request['attempt']}")

    assert body["status"] == "done", body
    assert body["result_schema_version"] == "2.0"
    assert body["metrics"]["duration_ms"] >= 0
    result_ref = body["result_s3"]
    assert result_ref["bucket"] == request["s3"]["bucket"]
    assert result_ref["key"] == f"results/{request['file_id']}/{request['job_id']}/result.json"
    uploads.keys.append((result_ref["bucket"], result_ref["key"]))  # результат тоже убирается после теста

    stored: dict[str, Any] = json.loads(await uploads.read(result_ref["bucket"], result_ref["key"]))
    assert schema_errors(stored) == []
    assert RecognitionResult.model_validate(stored).to_wire() == stored
    assert stored["status"] == "done"
    assert stored["job_id"] == request["job_id"]
    assert [document["document_type"] for document in stored["documents"]] == ["OSV"]
    assert stored["documents"][0]["header"]["account_code_raw"] == "66"
    await core.assert_quiet(request)


@pytest.mark.fixtures
async def test_real_statements_through_worker_give_done_with_two_documents(
    core: Core, uploads: Uploads, fixtures_dir: Path
) -> None:
    """«Реальные формы через воркер»: книга НПЗ → ack → `done`; жёстких расхождений нет, мягкие — по golden."""
    request = _new_request(key_name="npz_2025_q3.xlsx")
    request["file_type"] = "XLSX"
    await uploads.put(request, (fixtures_dir / REAL_NPZ).read_bytes(), XLSX_CONTENT_TYPE)

    await core.publish(request)
    _assert_ack(await core.next_message(request), request)
    body = _envelope(await core.next_message(request), request, f"result:{request['job_id']}:{request['attempt']}")

    assert body["status"] == "done", body
    assert body["result_schema_version"] == "2.0"
    result_ref = body["result_s3"]
    uploads.keys.append((result_ref["bucket"], result_ref["key"]))

    stored: dict[str, Any] = json.loads(await uploads.read(result_ref["bucket"], result_ref["key"]))
    assert schema_errors(stored) == []
    assert RecognitionResult.model_validate(stored).to_wire() == stored
    assert stored["status"] == "done"
    assert stored["schema_version"] == "2.0" and stored["service"]["statements_version"]
    balance, income = stored["documents"]
    assert [balance["document_type"], income["document_type"]] == ["BALANCE", "INCOME"]
    assert balance["checks"]["relations_failed"] == [] and income["checks"]["relations_failed"] == []
    # мягкие расхождения (OQ-138) видны аналитику; набор — как в golden НПЗ
    assert {item["relation"] for item in balance["checks"]["soft_relations_failed"]} >= {
        "1500 = 1510 + 1520 + 1530 + 1540 + 1550",
        "1410 = 14101 + 14102 + 14103 + 14104 + 14105",
    }
    assert {item["relation"] for item in income["checks"]["soft_relations_failed"]} >= {
        "2400 = 2300 - 2410 + 2420",
        "2410 = 2411 + 2412",
    }
    # лист с единственной строкой 2400 присоединён (O-9)
    assert "2400" in {line["line_code"] for line in income["lines"]}
    await core.assert_quiet(request)