diff --git a/api/tests/integration/test_roundtrip.py b/api/tests/integration/test_roundtrip.py new file mode 100644 index 0000000..5ab7241 --- /dev/null +++ b/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 = :` и собирает конверты из `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 = :` + _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) diff --git a/docs/superpowers/plans/2026-10-03-slice-0-skeleton.md b/docs/superpowers/plans/2026-10-03-slice-0-skeleton.md index 3bd1982..055a594 100644 --- a/docs/superpowers/plans/2026-10-03-slice-0-skeleton.md +++ b/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` diff --git a/openspec/changes/slice-0-skeleton/tasks.md b/openspec/changes/slice-0-skeleton/tasks.md index 0c7fba1..077f5e1 100644 --- a/openspec/changes/slice-0-skeleton/tasks.md +++ b/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