256 lines
13 KiB
Python
256 lines
13 KiB
Python
"""Интеграционный 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 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: 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:
|
||
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_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)
|