diff --git a/api/tests/integration/test_nats_integration.py b/api/tests/integration/test_nats_integration.py new file mode 100644 index 0000000..2c7e0be --- /dev/null +++ b/api/tests/integration/test_nats_integration.py @@ -0,0 +1,149 @@ +"""Интеграционные тесты `src/core/nats.py` против compose-NATS без JetStream (`.env.test`: nats://localhost:4222; +spec «Соединение с NATS по правилам шаблона, без JetStream»; design D-6). + +`.env.test` содержит `NATS_ENABLED=false` (для приложения) — тесты строят соединение сами. Subject'ы уникальны на +тест (`test.tabular.`): общие `recognition.*` не трогаем, на шине может работать другой сервис. +""" + +import asyncio +import time +import uuid +from collections import Counter +from typing import Any + +import pytest +from nats.aio.client import Client as NATS +from nats.aio.msg import Msg + +from src.contract.subjects import NATS_MSG_ID_HEADER +from src.core.config import NATSSettings, get_settings +from src.core.nats import REQUEST_QUEUE_GROUP, NatsConnection, NatsPublishError + +pytestmark = pytest.mark.integration + +READY_TIMEOUT = 5.0 +START_MAX_SECONDS = 0.5 # start() не ждёт подключения +STOP_MAX_SECONDS = 2.0 + + +def _nats_settings() -> NATSSettings: + return get_settings().OPENBAO.NATS + + +def _unique_subject() -> str: + return f"test.tabular.{uuid.uuid4().hex}" + + +async def _wait_until(predicate: Any, timeout: float = 5.0) -> None: + deadline = time.monotonic() + timeout + while not predicate(): + if time.monotonic() > deadline: + raise AssertionError("условие не выполнено за отведённое время") + await asyncio.sleep(0.02) + + +async def _start(conn: NatsConnection) -> NatsConnection: + await conn.start() + if not await conn.wait_ready(timeout=READY_TIMEOUT): + await conn.stop() + pytest.fail(f"NATS недоступен ({_nats_settings().NATS_URL}, .env.test; make up)", pytrace=False) + return conn + + +async def _plain_client() -> NATS: + nc = NATS() + await nc.connect(_nats_settings().NATS_URL, connect_timeout=2, max_reconnect_attempts=1) + return nc + + +async def test_subscribe_requests_receives_message_with_msg_id_header(): + """Запрос ядра (сторонний клиент с `Nats-Msg-Id`) доходит до подписки сервиса в queue group с заголовком.""" + subject = _unique_subject() + received: list[Msg] = [] + + async def handler(msg: Msg) -> None: + received.append(msg) + + async def on_connected(conn: NatsConnection) -> None: + await conn.subscribe_requests(handler, subject=subject) + + conn = await _start(NatsConnection(_nats_settings(), on_connected=on_connected)) + core = await _plain_client() + try: + assert conn.is_connected + await core.publish(subject, b'{"job_id": "j-1"}', headers={NATS_MSG_ID_HEADER: "j-1:1"}) + await core.flush() + await _wait_until(lambda: len(received) == 1) + finally: + await core.close() + await conn.stop() + + msg = received[0] + assert msg.subject == subject + assert msg.data == b'{"job_id": "j-1"}' + assert msg.headers == {NATS_MSG_ID_HEADER: "j-1:1"} + + +async def test_publish_delivers_payload_with_msg_id_header(): + """Ack/результат сервиса доходит до подписчика (как ядро) с заголовком `Nats-Msg-Id`.""" + subject = _unique_subject() + core = await _plain_client() + subscription = await core.subscribe(subject) + conn = await _start(NatsConnection(_nats_settings())) + try: + await conn.publish(subject, b'{"status": "processing"}', "ack:j-1:1") + msg = await subscription.next_msg(timeout=5) + finally: + await conn.stop() + await core.close() + + assert msg.data == b'{"status": "processing"}' + assert msg.headers == {NATS_MSG_ID_HEADER: "ack:j-1:1"} + + +async def test_queue_group_delivers_each_request_to_exactly_one_instance(): + subject = _unique_subject() + received: dict[str, list[str]] = {"a": [], "b": []} + + def make_on_connected(name: str) -> Any: + async def handler(msg: Msg) -> None: + assert msg.headers is not None + received[name].append(msg.headers[NATS_MSG_ID_HEADER]) + + async def on_connected(conn: NatsConnection) -> None: + await conn.subscribe_requests(handler, subject=subject) + + return on_connected + + instance_a = await _start(NatsConnection(_nats_settings(), on_connected=make_on_connected("a"))) + instance_b = await _start(NatsConnection(_nats_settings(), on_connected=make_on_connected("b"))) + total = 20 + try: + for i in range(total): + await instance_a.publish(subject, b"{}", f"job-{i}:1") + await _wait_until(lambda: len(received["a"]) + len(received["b"]) >= total) + await asyncio.sleep(0.2) # дубль, если бы он был, успел бы прийти + finally: + await instance_a.stop() + await instance_b.stop() + + assert REQUEST_QUEUE_GROUP == "tabular-parser" + all_ids = received["a"] + received["b"] + assert Counter(all_ids) == Counter(f"job-{i}:1" for i in range(total)) + + +async def test_start_against_unreachable_server_does_not_block_and_stops_cleanly(): + """O-3: NATS недоступен при старте — start() не ждёт, публикация после повторов падает, stop() быстрый.""" + conn = NatsConnection(NATSSettings(NATS_URL="nats://127.0.0.1:54999"), publish_retries=2, publish_retry_pause=0.05) + + started = time.monotonic() + await conn.start() + assert time.monotonic() - started < START_MAX_SECONDS + await asyncio.sleep(0.3) + assert not conn.is_connected + with pytest.raises(NatsPublishError): + await conn.publish("test.tabular.unreachable", b"{}", "m:1") + + started = time.monotonic() + await conn.stop() + assert time.monotonic() - started < STOP_MAX_SECONDS diff --git a/api/tests/integration/test_s3_integration.py b/api/tests/integration/test_s3_integration.py new file mode 100644 index 0000000..80e29a1 --- /dev/null +++ b/api/tests/integration/test_s3_integration.py @@ -0,0 +1,39 @@ +"""Интеграционный тест `S3Storage` против compose-MinIO (`.env.test`: localhost:9000, бакет `arm-ka`; design D-4). + +Ключ уникален на прогон (`tests//x.json`) и удаляется после теста. +""" + +import json +import uuid + +import pytest +from miniopy_async import Minio + +from src.core.clients import S3Storage +from src.core.config import get_settings + +pytestmark = pytest.mark.integration + + +async def test_put_json_then_get_bytes_round_trip_in_minio(): + 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, + ) + storage = S3Storage(client=client, bucket_name=minio.MINIO_BUCKET) + key = f"tests/{uuid.uuid4()}/x.json" + obj = {"job_id": "j-1", "сообщение": "разбор XLSX", "amount": "1234.50"} + try: + await storage.put_json(minio.MINIO_BUCKET, key, obj) + + data = await storage.get_bytes(minio.MINIO_BUCKET, key) + stat = await client.stat_object(minio.MINIO_BUCKET, key) + finally: + await client.remove_object(minio.MINIO_BUCKET, key) + await client.close_session() + + assert json.loads(data.decode("utf-8")) == obj + assert stat.content_type == "application/json" 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 d92efee..f158944 100644 --- a/docs/superpowers/plans/2026-10-03-slice-0-skeleton.md +++ b/docs/superpowers/plans/2026-10-03-slice-0-skeleton.md @@ -47,7 +47,7 @@ ### Task 8: [4.2] `src/core/clients.py` -- [ ] `src/core/clients.py`: `S3Storage.get_bytes(bucket, key)`, `put_json(bucket, key, obj)`; `src/core/nats.py`: подключение (с/без nkeys, автопереподключение), `subscribe_requests(queue="tabular-parser", cb)`, `publish(subject, payload, msg_id)` с повторами; проверка: unit-тесты с фейковым соединением; интеграционный тест подключения к compose-NATS без JetStream и получения сообщения с заголовком +- [x] `src/core/clients.py`: `S3Storage.get_bytes(bucket, key)`, `put_json(bucket, key, obj)`; `src/core/nats.py`: подключение (с/без nkeys, автопереподключение), `subscribe_requests(queue="tabular-parser", cb)`, `publish(subject, payload, msg_id)` с повторами; проверка: unit-тесты с фейковым соединением; интеграционный тест подключения к compose-NATS без JetStream и получения сообщения с заголовком ### Task 9: [4.3] `src/core/worker.py` diff --git a/openspec/changes/slice-0-skeleton/tasks.md b/openspec/changes/slice-0-skeleton/tasks.md index 10d821a..47cd053 100644 --- a/openspec/changes/slice-0-skeleton/tasks.md +++ b/openspec/changes/slice-0-skeleton/tasks.md @@ -34,7 +34,7 @@ - [x] 4.1 `src/parser/__init__.py`: протокол `Parser`, `ParseOutcome`, `NullParser` (`NOT_IMPLEMENTED`); `api/rules/README.md`; `tests/unit/test_null_parser.py`; проверка: заглушка возвращает код и сообщение, не трогая входные байты -- [ ] 4.2 `src/core/clients.py`: `S3Storage.get_bytes(bucket, key)`, `put_json(bucket, key, obj)`; `src/core/nats.py`: +- [x] 4.2 `src/core/clients.py`: `S3Storage.get_bytes(bucket, key)`, `put_json(bucket, key, obj)`; `src/core/nats.py`: подключение (с/без nkeys, автопереподключение), `subscribe_requests(queue="tabular-parser", cb)`, `publish(subject, payload, msg_id)` с повторами; проверка: unit-тесты с фейковым соединением; интеграционный тест подключения к compose-NATS без JetStream и получения сообщения с заголовком