test: интеграционный round-trip сервиса против compose-NATS и MinIO (задача 4.4)

Приложение поднимается в процессе теста через lifespan с NATS_ENABLED=True; тест в роли ядра кладёт исходник в
uploads/, публикует запрос и собирает recognition.result по своему job_id: ack processing, затем error
NOT_IMPLEMENTED с правильными Nats-Msg-Id и валидные по схеме конверты; повтор запроса — только ещё один ack
(второго результата нет в окне тишины QUIET_WINDOW). Плюс сценарии spec: нет объекта в S3 -> FILE_READ_ERROR и
сервис работает дальше; schema_version 2.0 -> один SCHEMA_MISMATCH без ack.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

api/tests/integration/test_roundtrip.py

@@ -0,0 +1,246 @@
+"""Интеграционный round-trip сервиса против compose-NATS (без JetStream) и compose-MinIO (`.env.test`).
+
+Приложение поднимается в процессе теста через lifespan (`src.app`) с `NATS_ENABLED=True` — тот же код, что в проде:
+`_connect_s3`, `NatsConnection` с подпиской в queue group `tabular-parser`, `RecognitionWorker` с `NullParser`.
+Тест играет роль ядра: кладёт исходник в 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 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: content is not read by NullParser"
+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:
+        for bucket, key in store.keys:
+            await client.remove_object(bucket, key)
+        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
+    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
+            )
+        yield
+    assert application.state.worker._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_not_implemented_and_duplicate_gets_only_ack(core: Core, uploads: Uploads) -> None:
+    """«Ack раньше результата», «Запрос с заглушкой парсера», повтор запроса — только ещё один 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, "NOT_IMPLEMENTED")
+
+    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, "NOT_IMPLEMENTED")
+
+
+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)

docs/superpowers/plans/2026-10-03-slice-0-skeleton.md

@@ -55,7 +55,7 @@
 
 ### Task 10: [4.4] `tests/integration/test_roundtrip.py` (маркер `integration`)
 
-- [ ] `tests/integration/test_roundtrip.py` (маркер `integration`): положить файл в MinIO `uploads/…`, опубликовать `request-tabular.json`, собрать сообщения `recognition.result`: первое — `processing`, второе — `error NOT_IMPLEMENTED` с правильными `Nats-Msg-Id`; повторная публикация запроса даёт только ещё один ack; проверка: зелёный против compose
+- [x] `tests/integration/test_roundtrip.py` (маркер `integration`): положить файл в MinIO `uploads/…`, опубликовать `request-tabular.json`, собрать сообщения `recognition.result`: первое — `processing`, второе — `error NOT_IMPLEMENTED` с правильными `Nats-Msg-Id`; повторная публикация запроса даёт только ещё один ack; проверка: зелёный против compose
 
 ### Task 11: [5.1] `make check-full`
 

openspec/changes/slice-0-skeleton/tasks.md

@@ -43,7 +43,7 @@
       `metrics.duration_ms`; запуск/остановка в lifespan при `NATS_ENABLED`; `/healthcheck.nats`; проверка: unit-тесты с
       фейками — сценарии spec (ack раньше результата, повторный запрос, NOT_IMPLEMENTED, FILE_READ_ERROR, SCHEMA_MISMATCH без
       ack, подмена парсера → `done`, повтор публикации)
-- [ ] 4.4 `tests/integration/test_roundtrip.py` (маркер `integration`): положить файл в MinIO `uploads/…`, опубликовать
+- [x] 4.4 `tests/integration/test_roundtrip.py` (маркер `integration`): положить файл в MinIO `uploads/…`, опубликовать
       `request-tabular.json`, собрать сообщения `recognition.result`: первое — `processing`, второе — `error NOT_IMPLEMENTED`
       с правильными `Nats-Msg-Id`; повторная публикация запроса даёт только ещё один ack; проверка: зелёный против compose