test: обмен в реальном lifespan и два инстанса ядра на NATS (I-7)
tests/test_exchange_lifespan_nats.py (integration, NATS и PostgreSQL из .env.test, уникальный subject результатов, queue group arm-core): - настоящий lifespan при NATS_ENABLED=true: /healthcheck -> nats=ok, подписка видна в мониторинге NATS (/connz) с qgroup arm-core, публикатор lifespan переводит задание в published, inbox принимает ack и результат; остановка гасит задачи обмена до nats.stop(), закрывает пул БД и не оставляет висящих задач; - два NatsConnection с настоящим handle_message: результат обрабатывает ровно один инстанс, задание и s3_object меняются один раз; повторная доставка того же сообщения ничего не меняет. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
cce92212d2
commit
75045af723
339
api/tests/test_exchange_lifespan_nats.py
Normal file
339
api/tests/test_exchange_lifespan_nats.py
Normal file
@ -0,0 +1,339 @@
|
||||
"""Обмен `recognition_job` в реальном lifespan и два инстанса ядра (финальное ревью I-7; spec recognition-job-exchange,
|
||||
«Соединение с NATS по шаблону, без JetStream», «Inbox принимает результат идемпотентно»; design D-6).
|
||||
|
||||
Интеграционные тесты против NATS и PostgreSQL из `.env.test` (БД сессии pytest; задания фиксируются `CommittedJobs`
|
||||
и удаляются в teardown). Subject результатов — уникальный на тест (`NatsConnection(result_subject=...)`): общий
|
||||
`recognition.result` не трогаем, queue group — боевая `arm-core`. S3 не нужен (`_connect_s3` → None), SMB выключен
|
||||
(`.env.test`), пул движка БД приложения подменён (lifespan закрывает его последним шагом). Ожидания — с условием и
|
||||
ограничением по времени, без «голых» пауз.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import importlib
|
||||
import json
|
||||
import time
|
||||
import uuid
|
||||
from collections.abc import AsyncIterator, Awaitable, Callable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from functools import partial
|
||||
from typing import Any
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from nats.aio.client import Client as NATS
|
||||
from nats.aio.msg import Msg
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
|
||||
|
||||
from src.contract import NATS_MSG_ID_HEADER, ack_msg_id, result_msg_id
|
||||
from src.core.config import get_settings
|
||||
from src.core.nats import RESULT_QUEUE_GROUP, NatsConnection
|
||||
from src.models.enums import RecognitionStatus, S3BucketRole
|
||||
from src.models.files import RecognitionJob, S3Object
|
||||
from src.services.recognition_exchange import NoopMapper, handle_message
|
||||
from tests.factories import CommittedJobs
|
||||
from tests.test_recognition_inbox import MESSAGE_VALIDATOR, envelope
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
# src/__init__.py реэкспортирует объект app под именем модуля, поэтому модуль берём через importlib
|
||||
app_module = importlib.import_module("src.app")
|
||||
|
||||
READY_TIMEOUT = 5.0
|
||||
WAIT_TIMEOUT = 10.0
|
||||
NATS_MONITOR_PORT = 8222 # мониторинг NATS из docs/dev/docker-compose.yml
|
||||
EXCHANGE_TASKS = 2 # публикатор и вахтёр
|
||||
|
||||
|
||||
# --- инфраструктура ---------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _wait_until(predicate: Callable[[], Awaitable[bool]], what: str, timeout: float = WAIT_TIMEOUT) -> None:
|
||||
deadline = time.monotonic() + timeout
|
||||
while not await predicate():
|
||||
if time.monotonic() > deadline:
|
||||
raise AssertionError(f"за {timeout} с не выполнено: {what}")
|
||||
await asyncio.sleep(0.05)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def session_factory(db_engine: AsyncEngine) -> async_sessionmaker[AsyncSession]:
|
||||
return async_sessionmaker(db_engine, expire_on_commit=False)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def jobs(session_factory: async_sessionmaker[AsyncSession]) -> AsyncIterator[CommittedJobs]:
|
||||
committed = CommittedJobs(session_factory)
|
||||
try:
|
||||
yield committed
|
||||
finally:
|
||||
await committed.cleanup()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def result_subject() -> str:
|
||||
return f"test.result.{uuid.uuid4().hex}"
|
||||
|
||||
|
||||
class _Service:
|
||||
"""«Сервис распознавания» на голом nats-py: публикует конверты `recognition-result-message` в subject ядра."""
|
||||
|
||||
def __init__(self, result_subject: str) -> None:
|
||||
self.result_subject = result_subject
|
||||
self.nc = NATS()
|
||||
|
||||
async def start(self) -> None:
|
||||
url = get_settings().OPENBAO.NATS.NATS_URL
|
||||
try:
|
||||
await self.nc.connect(url, connect_timeout=2, allow_reconnect=False)
|
||||
except Exception as exc:
|
||||
pytest.fail(f"NATS недоступен ({url}, .env.test): {type(exc).__name__}", pytrace=False)
|
||||
|
||||
async def send(self, body: dict[str, Any], msg_id: str) -> None:
|
||||
MESSAGE_VALIDATOR.validate(body)
|
||||
payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
|
||||
await self.nc.publish(self.result_subject, payload, headers={NATS_MSG_ID_HEADER: msg_id})
|
||||
await self.nc.flush()
|
||||
|
||||
async def stop(self) -> None:
|
||||
await self.nc.close()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def service(result_subject: str) -> AsyncIterator[_Service]:
|
||||
svc = _Service(result_subject)
|
||||
await svc.start()
|
||||
try:
|
||||
yield svc
|
||||
finally:
|
||||
await svc.stop()
|
||||
|
||||
|
||||
async def _row(session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID) -> dict[str, Any]:
|
||||
async with session_factory() as session:
|
||||
row = (await session.execute(select(RecognitionJob.__table__).where(RecognitionJob.id == job_id))).one()
|
||||
return dict(row._mapping)
|
||||
|
||||
|
||||
async def _s3_rows(session_factory: async_sessionmaker[AsyncSession], key: str) -> list[dict[str, Any]]:
|
||||
async with session_factory() as session:
|
||||
rows = await session.execute(select(S3Object.__table__).where(S3Object.key == key))
|
||||
return [dict(r._mapping) for r in rows]
|
||||
|
||||
|
||||
async def _status_is(
|
||||
session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID, status: RecognitionStatus
|
||||
) -> bool:
|
||||
return (await _row(session_factory, job_id))["status"] is status
|
||||
|
||||
|
||||
def _result_key(file_id: uuid.UUID, job_id: uuid.UUID) -> str:
|
||||
return f"results/{file_id}/{job_id}/result.json"
|
||||
|
||||
|
||||
async def _queue_subscriptions(subject: str) -> list[dict[str, Any]] | None:
|
||||
"""Подписки на `subject` по мониторингу NATS (`/connz?subs=detail`); `None` — мониторинг недоступен."""
|
||||
host = urlsplit(get_settings().OPENBAO.NATS.NATS_URL).hostname or "localhost"
|
||||
url = f"http://{host}:{NATS_MONITOR_PORT}/connz"
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=2.0) as client:
|
||||
response = await client.get(url, params={"subs": "detail", "limit": 1024})
|
||||
except httpx.HTTPError:
|
||||
return None
|
||||
if response.status_code != httpx.codes.OK:
|
||||
return None
|
||||
return [
|
||||
sub
|
||||
for conn in response.json().get("connections", [])
|
||||
for sub in conn.get("subscriptions_list_detail") or []
|
||||
if sub.get("subject") == subject
|
||||
]
|
||||
|
||||
|
||||
class _EngineStub:
|
||||
"""Подмена движка БД приложения: lifespan вызывает `dispose()` последним шагом остановки."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.disposed = 0
|
||||
|
||||
async def dispose(self) -> None:
|
||||
self.disposed += 1
|
||||
|
||||
|
||||
# --- (1) реальный lifespan ----------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_real_lifespan_runs_exchange_over_real_nats( # noqa: PLR0915 - один сквозной сценарий жизненного цикла
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
session_factory: async_sessionmaker[AsyncSession],
|
||||
jobs: CommittedJobs,
|
||||
service: _Service,
|
||||
result_subject: str,
|
||||
):
|
||||
"""NATS_ENABLED=true: `/healthcheck` → `nats=ok`, подписка в queue group `arm-core` активна, публикатор и inbox
|
||||
работают с БД; остановка гасит задачи обмена ДО NATS и не оставляет висящих задач."""
|
||||
|
||||
async def no_s3() -> None:
|
||||
return None
|
||||
|
||||
engine = _EngineStub()
|
||||
monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True)
|
||||
monkeypatch.setattr(app_module.settings, "OUTBOX_POLL_SECONDS", 1) # публикатор не ждёт 5 с между проходами
|
||||
monkeypatch.setattr(app_module.settings.OPENBAO.SMB, "SMB_HOST", "")
|
||||
monkeypatch.setattr(app_module, "_connect_s3", no_s3)
|
||||
monkeypatch.setattr(app_module, "async_session_maker", session_factory)
|
||||
monkeypatch.setattr(app_module, "engine", engine)
|
||||
# Настоящий `_create_nats` и подписка `_subscribe_results`; отличается только subject результатов
|
||||
monkeypatch.setattr(app_module, "NatsConnection", partial(NatsConnection, result_subject=result_subject))
|
||||
|
||||
job_id = await jobs.make_queued_job(reply_deadline=datetime.now(UTC) + timedelta(hours=1))
|
||||
file_id = (await jobs.get(job_id)).file_id
|
||||
tasks_before = asyncio.all_tasks()
|
||||
exchange_done_at_nats_stop: list[bool] = []
|
||||
|
||||
async with app_module.lifespan(app_module.app):
|
||||
app = app_module.app
|
||||
nats: NatsConnection = app.state.nats
|
||||
exchange = app.state.exchange
|
||||
assert isinstance(nats, NatsConnection)
|
||||
assert nats.result_subject == result_subject
|
||||
assert nats.queue == RESULT_QUEUE_GROUP == "arm-core"
|
||||
assert exchange is not None
|
||||
assert len(exchange.tasks) == EXCHANGE_TASKS
|
||||
|
||||
if not await nats.wait_ready(timeout=READY_TIMEOUT): # подключение и подписка выполнены
|
||||
pytest.fail(f"NATS недоступен ({get_settings().OPENBAO.NATS.NATS_URL}, .env.test)", pytrace=False)
|
||||
transport = httpx.ASGITransport(app=app)
|
||||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||||
health = (await client.get("/healthcheck")).json()
|
||||
assert health["nats"] == "ok"
|
||||
|
||||
subscriptions = await _queue_subscriptions(result_subject)
|
||||
if subscriptions is not None: # мониторинг NATS доступен: подписка ядра видна серверу в queue group
|
||||
assert [sub.get("qgroup") for sub in subscriptions] == ["arm-core"]
|
||||
assert not any(task.done() for task in exchange.tasks)
|
||||
|
||||
# публикатор lifespan: queued → published
|
||||
await _wait_until(
|
||||
partial(_status_is, session_factory, job_id, RecognitionStatus.published), "публикация задания"
|
||||
)
|
||||
# inbox lifespan (подписка на result_subject): ack → processing, результат → done
|
||||
await service.send(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1))
|
||||
await _wait_until(partial(_status_is, session_factory, job_id, RecognitionStatus.processing), "ack")
|
||||
await service.send(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1))
|
||||
await _wait_until(partial(_status_is, session_factory, job_id, RecognitionStatus.done), "результат")
|
||||
assert not any(task.done() for task in exchange.tasks)
|
||||
|
||||
original_stop = nats.stop
|
||||
|
||||
async def recording_stop() -> None:
|
||||
exchange_done_at_nats_stop.append(all(task.done() for task in exchange.tasks))
|
||||
await original_stop()
|
||||
|
||||
monkeypatch.setattr(nats, "stop", recording_stop)
|
||||
|
||||
assert exchange_done_at_nats_stop == [True] # задачи обмена остановлены до NATS
|
||||
assert all(task.done() for task in exchange.tasks)
|
||||
assert not nats.is_connected
|
||||
assert engine.disposed == 1
|
||||
|
||||
current = asyncio.current_task()
|
||||
|
||||
async def no_leftover_tasks() -> bool:
|
||||
return not {t for t in asyncio.all_tasks() - tasks_before if t is not current and not t.done()}
|
||||
|
||||
await _wait_until(no_leftover_tasks, "все задачи lifespan завершены", timeout=READY_TIMEOUT)
|
||||
job = await _row(session_factory, job_id)
|
||||
(s3,) = await _s3_rows(session_factory, _result_key(file_id, job_id))
|
||||
assert (job["result_s3_object_id"], job["result_msg_id"]) == (s3["id"], f"result:{job_id}:1")
|
||||
assert s3["bucket_role"] is S3BucketRole.results
|
||||
|
||||
|
||||
# --- (2) два инстанса ядра в одной queue group ----------------------------------------------------------------------
|
||||
|
||||
|
||||
class _CountingHandler:
|
||||
"""Обёртка над настоящим `handle_message`: считает вызовы инстанса и незавершённые обработки."""
|
||||
|
||||
def __init__(self, handler: Callable[[Msg], Awaitable[None]]) -> None:
|
||||
self._handler = handler
|
||||
self.calls = 0
|
||||
self.in_flight = 0
|
||||
|
||||
async def __call__(self, msg: Msg) -> None:
|
||||
self.calls += 1
|
||||
self.in_flight += 1
|
||||
try:
|
||||
await self._handler(msg)
|
||||
finally:
|
||||
self.in_flight -= 1
|
||||
|
||||
|
||||
async def test_two_core_instances_handle_result_once(
|
||||
session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, service: _Service, result_subject: str
|
||||
):
|
||||
"""Два `NatsConnection` в queue group `arm-core` на одном subject: результат обрабатывает ровно один инстанс,
|
||||
задание меняется один раз; повторная доставка того же сообщения ничего не меняет."""
|
||||
settings = get_settings().OPENBAO.NATS
|
||||
handlers = [
|
||||
_CountingHandler(partial(handle_message, session_factory=session_factory, mapper=NoopMapper()))
|
||||
for _ in range(2)
|
||||
]
|
||||
|
||||
def subscriber(handler: _CountingHandler) -> Callable[[NatsConnection], Awaitable[None]]:
|
||||
async def subscribe(conn: NatsConnection) -> None:
|
||||
await conn.subscribe_results(handler)
|
||||
|
||||
return subscribe
|
||||
|
||||
cores = [NatsConnection(settings, on_connected=subscriber(h), result_subject=result_subject) for h in handlers]
|
||||
job_id = await jobs.make_queued_job(
|
||||
status=RecognitionStatus.published,
|
||||
published_at=datetime.now(UTC),
|
||||
reply_deadline=datetime.now(UTC) + timedelta(hours=1),
|
||||
)
|
||||
file_id = (await jobs.get(job_id)).file_id
|
||||
key = _result_key(file_id, job_id)
|
||||
result = envelope(job_id, file_id, "done", metrics={"duration_ms": 7})
|
||||
|
||||
async def settled() -> bool:
|
||||
"""Сервер доставил всё отправленное обоим инстансам (PING/PONG), и их обработчики завершились."""
|
||||
for core in cores:
|
||||
await core._client.flush(timeout=2)
|
||||
return all(h.in_flight == 0 for h in handlers)
|
||||
|
||||
for core in cores:
|
||||
await core.start()
|
||||
try:
|
||||
for core in cores:
|
||||
if not await core.wait_ready(timeout=READY_TIMEOUT):
|
||||
pytest.fail(f"NATS недоступен ({settings.NATS_URL}, .env.test)", pytrace=False)
|
||||
subscriptions = await _queue_subscriptions(result_subject)
|
||||
if subscriptions is not None:
|
||||
assert [sub.get("qgroup") for sub in subscriptions] == ["arm-core", "arm-core"]
|
||||
|
||||
await service.send(result, result_msg_id(job_id, 1))
|
||||
await _wait_until(partial(_status_is, session_factory, job_id, RecognitionStatus.done), "результат")
|
||||
await _wait_until(settled, "обработка доставленных сообщений")
|
||||
|
||||
assert sorted(h.calls for h in handlers) == [0, 1] # ровно один инстанс получил сообщение
|
||||
job = await _row(session_factory, job_id)
|
||||
s3_rows = await _s3_rows(session_factory, key)
|
||||
assert len(s3_rows) == 1
|
||||
assert (job["result_msg_id"], job["result_s3_object_id"]) == (f"result:{job_id}:1", s3_rows[0]["id"])
|
||||
assert job["metrics"] == {"duration_ms": 7}
|
||||
|
||||
# повторная доставка того же сообщения (at-least-once): снова один инстанс, без изменений
|
||||
await service.send(result, result_msg_id(job_id, 1))
|
||||
|
||||
async def redelivered() -> bool:
|
||||
return sum(h.calls for h in handlers) == 2 and await settled() # noqa: PLR2004 - две доставки
|
||||
|
||||
await _wait_until(redelivered, "повторная доставка обработана")
|
||||
assert await _row(session_factory, job_id) == job
|
||||
assert await _s3_rows(session_factory, key) == s3_rows
|
||||
finally:
|
||||
for core in cores:
|
||||
await core.stop()
|
||||
Loading…
x
Reference in New Issue
Block a user