c9d60907d3
I-1: ветка повторной публикации в _process_one записывает request_subject (как первая публикация); тест test_republish_updates_request_subject. M-1: примеры /healthcheck в README (ядро: три subject'а; tabular-parser: два). M-3: docstring'и и логи не выдают умолчание за фактический subject. M-5: CLAUDE.md — subject'ы задаются OPENBAO__NATS__NATS_*_SUBJECT. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
894 lines
38 KiB
Python
894 lines
38 KiB
Python
"""Outbox-публикатор и вахтёр сроков `recognition_job` (spec recognition-job-exchange: «Outbox публикует задания из
|
||
`queued`», «Запрос без ack публикуется повторно, затем — `dlq`», «Таймаут ожидания результата»; design D-6). Срез 2c:
|
||
`timeout` вахтёра и `dlq` публикатора переводят файл текущего задания в `ERROR` (spec primary-files «Жизненный цикл
|
||
статуса файла», design D-5).
|
||
|
||
Интеграционные тесты работают против реального PostgreSQL (БД сессии pytest) и NATS из `.env.test`. Публикатор и
|
||
вахтёр ведут собственные транзакции, поэтому данные фиксируются реально (`CommittedJobs`) и удаляются в teardown.
|
||
Subject'ы запросов общие (`files.*_uploaded`): тестовый подписчик фильтрует сообщения по `job_id` своих заданий.
|
||
Время в тестах повторов — инжектируемые часы `now`, без реальных ожиданий таймаутов.
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import time
|
||
import uuid
|
||
from collections.abc import AsyncIterator, Callable
|
||
from datetime import UTC, datetime, timedelta
|
||
from pathlib import Path
|
||
from typing import Any, get_args
|
||
|
||
import pytest
|
||
import pytest_asyncio
|
||
from jsonschema import Draft202012Validator, FormatChecker
|
||
from nats.aio.client import Client as NATS
|
||
from nats.aio.msg import Msg
|
||
from sqlalchemy import JSON, update
|
||
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
|
||
|
||
from src.contract import NATS_MSG_ID_HEADER, RecognitionRequest
|
||
from src.contract.request import FileType as ContractFileType
|
||
from src.core.config import get_settings
|
||
from src.core.nats import NatsConnection, NatsPublishError
|
||
from src.models.enums import FileStatus, FileType, RecognitionPipeline, RecognitionStatus
|
||
from src.models.files import RecognitionJob
|
||
from src.services import recognition_exchange as rx
|
||
from src.services.recognition_exchange import (
|
||
WATCHDOG_PERIOD_SECONDS,
|
||
ExchangeTasks,
|
||
expire_deadlines,
|
||
publish_pending,
|
||
run_publisher_loop,
|
||
run_watchdog_loop,
|
||
)
|
||
from tests.factories import CommittedJobs
|
||
|
||
CONTRACTS_DIR = Path(__file__).resolve().parents[1] / "contracts"
|
||
REQUEST_SCHEMA = json.loads((CONTRACTS_DIR / "recognition-request.schema.json").read_text(encoding="utf-8"))
|
||
REQUEST_VALIDATOR = Draft202012Validator(REQUEST_SCHEMA, format_checker=FormatChecker())
|
||
REQUEST_SUBJECTS = ("files.tabular_uploaded", "files.img_uploaded")
|
||
|
||
ACK = timedelta(seconds=30)
|
||
MAX_REPUBLISH = 3
|
||
READY_TIMEOUT = 5.0
|
||
CONCURRENT_JOBS = 20
|
||
MIN_LOOP_PASSES = 3 # цикл пережил несколько ошибок прохода подряд
|
||
EXCHANGE_TASKS = 2 # публикатор и вахтёр
|
||
|
||
|
||
# --- инфраструктура ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
class Clock:
|
||
"""Инжектируемые часы: `now` для публикатора и вахтёра, `advance` — сдвиг без реального ожидания."""
|
||
|
||
def __init__(self) -> None:
|
||
self.current = datetime.now(UTC)
|
||
|
||
def __call__(self) -> datetime:
|
||
return self.current
|
||
|
||
def advance(self, delta: timedelta) -> None:
|
||
self.current += delta
|
||
|
||
|
||
class RequestSubscriber:
|
||
"""Тестовый подписчик на subject'ы запросов: «голый» клиент nats-py, сообщения фильтруются по `job_id`."""
|
||
|
||
def __init__(self, subjects: tuple[str, ...] = REQUEST_SUBJECTS) -> None:
|
||
self.subjects = subjects
|
||
self.messages: list[Msg] = []
|
||
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 collect(msg: Msg) -> None:
|
||
self.messages.append(msg)
|
||
|
||
for subject in self.subjects:
|
||
await self._nc.subscribe(subject, cb=collect)
|
||
await self._nc.flush()
|
||
|
||
async def stop(self) -> None:
|
||
await self._nc.close()
|
||
|
||
def for_job(self, job_id: uuid.UUID) -> list[Msg]:
|
||
return [m for m in self.messages if json.loads(m.data).get("job_id") == str(job_id)]
|
||
|
||
async def settle(self) -> None:
|
||
"""Всё, что сервер принял до этого момента, доставлено подписчику (PING/PONG после сообщений) и обработано."""
|
||
await self._nc.flush()
|
||
await asyncio.sleep(0.1)
|
||
|
||
async def wait_for(self, job_id: uuid.UUID, count: int, timeout: float = 10.0) -> list[Msg]:
|
||
deadline = time.monotonic() + timeout
|
||
while len(self.for_job(job_id)) < count:
|
||
if time.monotonic() > deadline:
|
||
raise AssertionError(f"за {timeout} с получено {len(self.for_job(job_id))} из {count} сообщений")
|
||
await asyncio.sleep(0.02)
|
||
return self.for_job(job_id)
|
||
|
||
|
||
class FakeNats:
|
||
"""Подставной NATS: состояние соединения и счётчик публикаций; первые `fail_first` публикаций — ошибка."""
|
||
|
||
def __init__(self, *, connected: bool = True, fail_first: int = 0, delegate: Any = None, delay: float = 0) -> None:
|
||
self.is_connected_value = connected
|
||
self.fail_first = fail_first
|
||
self.delegate = delegate
|
||
self.delay = delay
|
||
self.calls: list[tuple[str, str]] = []
|
||
|
||
@property
|
||
def is_connected(self) -> bool:
|
||
return self.is_connected_value
|
||
|
||
async def publish(self, subject: str, payload: bytes, msg_id: str) -> None:
|
||
self.calls.append((subject, msg_id))
|
||
if len(self.calls) <= self.fail_first:
|
||
raise NatsPublishError(f"подставная ошибка публикации {msg_id}")
|
||
if self.delay:
|
||
await asyncio.sleep(self.delay)
|
||
if self.delegate is not None:
|
||
await self.delegate.publish(subject, payload, msg_id)
|
||
|
||
|
||
@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_asyncio.fixture
|
||
async def nats() -> AsyncIterator[NatsConnection]:
|
||
conn = NatsConnection(get_settings().OPENBAO.NATS)
|
||
await conn.start()
|
||
try:
|
||
if not await conn.wait_ready(timeout=READY_TIMEOUT):
|
||
pytest.fail(f"NATS недоступен ({get_settings().OPENBAO.NATS.NATS_URL}, .env.test)", pytrace=False)
|
||
yield conn
|
||
finally:
|
||
await conn.stop()
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def subscriber() -> AsyncIterator[RequestSubscriber]:
|
||
sub = RequestSubscriber()
|
||
await sub.start()
|
||
try:
|
||
yield sub
|
||
finally:
|
||
await sub.stop()
|
||
|
||
|
||
async def _set_status(
|
||
session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID, status: RecognitionStatus
|
||
) -> None:
|
||
async with session_factory() as session, session.begin():
|
||
await session.execute(update(RecognitionJob).where(RecognitionJob.id == job_id).values(status=status))
|
||
|
||
|
||
def _publish(
|
||
session_factory: async_sessionmaker[AsyncSession], nats: Any, now: Callable[[], datetime] | None = None
|
||
) -> Any:
|
||
return publish_pending(session_factory, nats, ack_timeout=ACK, max_republish=MAX_REPUBLISH, now=now)
|
||
|
||
|
||
# --- юнит: соответствие enum, константы ------------------------------------------------------------------------
|
||
|
||
|
||
def test_orm_file_type_matches_contract_enum():
|
||
"""`stored_file.file_type` (DBML) и `file_type` схемы запроса — одно множество значений."""
|
||
assert {m.value for m in FileType} == set(get_args(ContractFileType))
|
||
assert set(REQUEST_SCHEMA["properties"]["file_type"]["enum"]) == {m.value for m in FileType}
|
||
|
||
|
||
def test_orm_pipeline_matches_request_subjects():
|
||
assert {m.value for m in RecognitionPipeline} == {"tabular", "image"}
|
||
|
||
|
||
def test_watchdog_period_is_30_seconds():
|
||
assert WATCHDOG_PERIOD_SECONDS == 30 # noqa: PLR2004 - значение из spec
|
||
|
||
|
||
# --- публикация -------------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize(
|
||
("pipeline", "file_type", "subject"),
|
||
[
|
||
(RecognitionPipeline.tabular, FileType.XLSX, "files.tabular_uploaded"),
|
||
(RecognitionPipeline.image, FileType.PDF, "files.img_uploaded"),
|
||
],
|
||
)
|
||
async def test_publish_request_to_service( # noqa: PLR0913,PLR0917 - фикстуры и параметры
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
jobs: CommittedJobs,
|
||
nats: NatsConnection,
|
||
subscriber: RequestSubscriber,
|
||
pipeline: RecognitionPipeline,
|
||
file_type: FileType,
|
||
subject: str,
|
||
):
|
||
"""Сценарий «Публикация запроса парсеру»: сообщение по схеме, `attempt = 1`, строка `published`."""
|
||
job_id = await jobs.make_queued_job(pipeline=pipeline, file_type=file_type, original_name="ОСВ 60.xlsx")
|
||
before = datetime.now(UTC)
|
||
|
||
report = await _publish(session_factory, nats)
|
||
|
||
assert report.published == 1
|
||
(msg,) = await subscriber.wait_for(job_id, 1)
|
||
body = json.loads(msg.data)
|
||
errors = [e.message for e in REQUEST_VALIDATOR.iter_errors(body)]
|
||
assert errors == []
|
||
request = RecognitionRequest.model_validate_json(msg.data)
|
||
job = await jobs.get(job_id)
|
||
assert msg.subject == subject
|
||
assert msg.headers == {NATS_MSG_ID_HEADER: f"{job_id}:1"}
|
||
assert request.job_id == job_id
|
||
assert request.file_id == job.file_id
|
||
assert request.pipeline == pipeline.value
|
||
assert request.file_type == file_type.value
|
||
assert request.original_name == "ОСВ 60.xlsx"
|
||
assert request.s3.bucket == "arm-ka"
|
||
assert request.s3.key == await jobs.s3_key(job_id)
|
||
assert request.attempt == 1
|
||
assert request.schema_version == "1.0"
|
||
# reply_deadline не был задан: вычислен при первой публикации (парсер — 5 мин, OCR — 20 мин) и сохранён в задание
|
||
assert job.reply_deadline is not None
|
||
assert request.reply_deadline == job.reply_deadline
|
||
expected = timedelta(minutes=20 if pipeline is RecognitionPipeline.image else 5)
|
||
assert expected - timedelta(minutes=1) < job.reply_deadline - before < expected + timedelta(minutes=1)
|
||
|
||
assert job.status is RecognitionStatus.published
|
||
assert job.request_msg_id == f"{job_id}:1"
|
||
assert job.request_subject == subject
|
||
assert job.published_at is not None and job.published_at >= before
|
||
assert job.requested_at is not None
|
||
assert job.attempt == 1
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_publish_uses_configured_request_subject(
|
||
session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, nats: NatsConnection
|
||
) -> None:
|
||
"""Review Focus 4: subject — из настроек окружения, `request_subject` задания — фактический subject публикации."""
|
||
run = uuid.uuid4().hex[:8]
|
||
subjects = {"tabular": f"test-{run}.files.tabular_uploaded", "image": f"test-{run}.files.img_uploaded"}
|
||
custom = RequestSubscriber(subjects=tuple(subjects.values()))
|
||
await custom.start()
|
||
try:
|
||
job_id = await jobs.make_queued_job(
|
||
pipeline=RecognitionPipeline.tabular, file_type=FileType.XLSX, original_name="ОСВ 60.xlsx"
|
||
)
|
||
await publish_pending(
|
||
session_factory, nats, ack_timeout=ACK, max_republish=MAX_REPUBLISH, request_subjects=subjects
|
||
)
|
||
(msg,) = await custom.wait_for(job_id, 1)
|
||
finally:
|
||
await custom.stop()
|
||
assert msg.subject == subjects["tabular"]
|
||
assert (await jobs.get(job_id)).request_subject == subjects["tabular"]
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_republish_updates_request_subject(
|
||
session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, nats: NatsConnection
|
||
) -> None:
|
||
"""I-1: повторная публикация уходит на актуальный subject и обновляет `request_subject` задания."""
|
||
run = uuid.uuid4().hex[:8]
|
||
subjects = {"tabular": f"test-{run}.files.tabular_uploaded", "image": f"test-{run}.files.img_uploaded"}
|
||
clock = Clock()
|
||
job_id = await jobs.make_queued_job(
|
||
pipeline=RecognitionPipeline.tabular,
|
||
file_type=FileType.XLSX,
|
||
original_name="ОСВ 61.xlsx",
|
||
reply_deadline=clock() + timedelta(hours=1),
|
||
)
|
||
assert (await _publish(session_factory, nats, clock)).published == 1
|
||
old_subject = (await jobs.get(job_id)).request_subject
|
||
assert old_subject != subjects["tabular"]
|
||
|
||
custom = RequestSubscriber(subjects=tuple(subjects.values()))
|
||
await custom.start()
|
||
try:
|
||
clock.advance(ACK + timedelta(seconds=1))
|
||
report = await publish_pending(
|
||
session_factory,
|
||
nats,
|
||
ack_timeout=ACK,
|
||
max_republish=MAX_REPUBLISH,
|
||
now=clock,
|
||
request_subjects=subjects,
|
||
)
|
||
assert report.republished == 1
|
||
(msg,) = await custom.wait_for(job_id, 1)
|
||
finally:
|
||
await custom.stop()
|
||
assert msg.subject == subjects["tabular"]
|
||
assert (await jobs.get(job_id)).request_subject == subjects["tabular"]
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_publish_keeps_existing_reply_deadline(
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
jobs: CommittedJobs,
|
||
nats: NatsConnection,
|
||
subscriber: RequestSubscriber,
|
||
):
|
||
deadline = datetime.now(UTC).replace(microsecond=0) + timedelta(hours=2)
|
||
job_id = await jobs.make_queued_job(reply_deadline=deadline)
|
||
|
||
await _publish(session_factory, nats)
|
||
|
||
(msg,) = await subscriber.wait_for(job_id, 1)
|
||
assert RecognitionRequest.model_validate_json(msg.data).reply_deadline == deadline
|
||
assert (await jobs.get(job_id)).reply_deadline == deadline
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_first_publication_msg_id_follows_attempt(
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
jobs: CommittedJobs,
|
||
nats: NatsConnection,
|
||
subscriber: RequestSubscriber,
|
||
):
|
||
"""`queued` с `attempt = 2` и устаревшим `request_msg_id` (будущий replay): id и тело — по текущему attempt."""
|
||
job_id = await jobs.make_queued_job(attempt=2, request_msg_id=f"stale-{uuid.uuid4()}")
|
||
|
||
await _publish(session_factory, nats)
|
||
|
||
(msg,) = await subscriber.wait_for(job_id, 1)
|
||
assert msg.headers == {NATS_MSG_ID_HEADER: f"{job_id}:2"}
|
||
assert RecognitionRequest.model_validate_json(msg.data).attempt == 2 # noqa: PLR2004 - attempt задания
|
||
job = await jobs.get(job_id)
|
||
assert (job.request_msg_id, job.attempt) == (f"{job_id}:2", 2)
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize("current", [True, False], ids=["current", "non_current"])
|
||
async def test_invalid_request_marks_job_error_and_pass_continues(
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
jobs: CommittedJobs,
|
||
nats: NatsConnection,
|
||
subscriber: RequestSubscriber,
|
||
current: bool,
|
||
):
|
||
"""Запрос не собирается по схеме (нет `original_name`) — задание `error/INVALID_REQUEST`, проход идёт дальше; файл
|
||
текущего задания — `ERROR` (финальная ошибка задания, spec primary-files), файл чужого задания не меняется."""
|
||
clock = Clock()
|
||
file_id, broken = await jobs.make_file_job(original_name=None)
|
||
# новое задание — `processing`, чтобы публикатор его не трогал (в проходе только старое задание файла)
|
||
new_job = None if current else await jobs.add_job(file_id, status=RecognitionStatus.processing, current=True)
|
||
good = await jobs.make_queued_job()
|
||
|
||
report = await _publish(session_factory, nats, clock)
|
||
|
||
assert (report.invalid, report.published) == (1, 1)
|
||
await subscriber.wait_for(good, 1)
|
||
await subscriber.settle()
|
||
assert subscriber.for_job(broken) == []
|
||
job = await jobs.get(broken)
|
||
assert job.status is RecognitionStatus.error
|
||
assert job.error_code == "INVALID_REQUEST"
|
||
assert job.error_message and "original_name" in job.error_message
|
||
assert job.finished_at is not None
|
||
assert job.request_msg_id is None
|
||
file = await jobs.file(file_id)
|
||
if current:
|
||
assert (file.status, file.updated_at) == (FileStatus.ERROR, clock())
|
||
else:
|
||
assert (file.status, file.current_recognition_job_id, file.updated_at) == (FileStatus.UPLOADED, new_job, None)
|
||
|
||
|
||
# --- повторы без ack и dlq ---------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_republish_without_ack_then_dlq(
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
jobs: CommittedJobs,
|
||
nats: NatsConnection,
|
||
subscriber: RequestSubscriber,
|
||
):
|
||
"""Сценарий «Сервис недоступен»: 1 + 3 публикации с одним `Nats-Msg-Id`, затем `dlq / NO_ACK`."""
|
||
clock = Clock()
|
||
job_id = await jobs.make_queued_job(reply_deadline=clock() + timedelta(hours=1), metrics={"note": "сохранить"})
|
||
|
||
assert (await _publish(session_factory, nats, clock)).published == 1
|
||
first = await jobs.get(job_id)
|
||
# раньше ACK_TIMEOUT повтора нет
|
||
clock.advance(ACK - timedelta(seconds=1))
|
||
assert (await _publish(session_factory, nats, clock)).republished == 0
|
||
|
||
for n in range(1, MAX_REPUBLISH + 1):
|
||
clock.advance(timedelta(seconds=2)) # с последней публикации прошло больше ACK_TIMEOUT
|
||
report = await _publish(session_factory, nats, clock)
|
||
assert report.republished == 1
|
||
job = await jobs.get(job_id)
|
||
assert job.metrics == {"note": "сохранить", "republish_count": n}
|
||
assert job.published_at == clock()
|
||
assert job.status is RecognitionStatus.published
|
||
clock.advance(ACK - timedelta(seconds=1))
|
||
|
||
messages = await subscriber.wait_for(job_id, 1 + MAX_REPUBLISH)
|
||
assert {m.headers[NATS_MSG_ID_HEADER] for m in messages} == {f"{job_id}:1"}
|
||
assert len({m.data for m in messages}) == 1 # то же тело: тот же attempt и reply_deadline
|
||
|
||
clock.advance(timedelta(seconds=2))
|
||
report = await _publish(session_factory, nats, clock)
|
||
assert (report.dlq, report.republished, report.published) == (1, 0, 0)
|
||
job = await jobs.get(job_id)
|
||
assert job.status is RecognitionStatus.dlq
|
||
assert job.error_code == "NO_ACK"
|
||
assert job.error_message
|
||
assert job.finished_at == clock()
|
||
assert job.request_msg_id == first.request_msg_id == f"{job_id}:1"
|
||
assert job.attempt == 1
|
||
assert job.metrics == {"note": "сохранить", "republish_count": MAX_REPUBLISH}
|
||
|
||
clock.advance(ACK * 2)
|
||
assert (await _publish(session_factory, nats, clock)).total == 0
|
||
await subscriber.settle()
|
||
assert len(subscriber.for_job(job_id)) == 1 + MAX_REPUBLISH
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize(
|
||
("stored", "expected_state"),
|
||
[
|
||
(None, (True, None)), # Python None → SQL NULL (JSONB(none_as_null=True), финальное ревью I-1)
|
||
(JSON.NULL, (False, "null")), # jsonb null: строки, записанные до исправления или сырым SQL
|
||
],
|
||
ids=["sql_null", "json_null"],
|
||
)
|
||
async def test_republish_counter_on_null_metrics(
|
||
session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, stored: Any, expected_state: Any
|
||
):
|
||
"""Обход для не-объекта в `metrics` (`jsonb_typeof` CASE) работает и для SQL NULL, и для jsonb null."""
|
||
clock = Clock()
|
||
job_id = await jobs.make_queued_job(reply_deadline=clock() + timedelta(hours=1), metrics=stored)
|
||
assert await jobs.metrics_sql_state(job_id) == expected_state
|
||
fake = FakeNats()
|
||
|
||
assert (await _publish(session_factory, fake, clock)).published == 1
|
||
assert await jobs.metrics_sql_state(job_id) == expected_state # первая публикация metrics не трогает
|
||
clock.advance(ACK + timedelta(seconds=1))
|
||
assert (await _publish(session_factory, fake, clock)).republished == 1
|
||
|
||
assert (await jobs.get(job_id)).metrics == {"republish_count": 1}
|
||
assert await jobs.metrics_sql_state(job_id) == (False, "object")
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_ack_after_first_republish_stops_republishing(
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
jobs: CommittedJobs,
|
||
nats: NatsConnection,
|
||
subscriber: RequestSubscriber,
|
||
):
|
||
"""Сценарий «Сервис появился после первого повтора»: после ack (`processing`) повторов нет."""
|
||
clock = Clock()
|
||
job_id = await jobs.make_queued_job(reply_deadline=clock() + timedelta(hours=1))
|
||
await _publish(session_factory, nats, clock)
|
||
clock.advance(ACK + timedelta(seconds=1))
|
||
assert (await _publish(session_factory, nats, clock)).republished == 1
|
||
await subscriber.wait_for(job_id, 2)
|
||
|
||
await _set_status(session_factory, job_id, RecognitionStatus.processing) # имитация ack
|
||
for _ in range(MAX_REPUBLISH + 1):
|
||
clock.advance(ACK + timedelta(seconds=1))
|
||
assert (await _publish(session_factory, nats, clock)).total == 0
|
||
|
||
await subscriber.settle()
|
||
assert len(subscriber.for_job(job_id)) == 2 # noqa: PLR2004 - первая публикация и один повтор
|
||
job = await jobs.get(job_id)
|
||
assert job.status is RecognitionStatus.processing
|
||
assert job.metrics == {"republish_count": 1}
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_publisher_skips_published_job_with_passed_deadline(
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
jobs: CommittedJobs,
|
||
nats: NatsConnection,
|
||
subscriber: RequestSubscriber,
|
||
):
|
||
"""`published` с прошедшим `reply_deadline` не републикуется и не уходит в dlq — его забирает вахтёр."""
|
||
now = datetime.now(UTC)
|
||
job_id = await jobs.make_queued_job(
|
||
status=RecognitionStatus.published,
|
||
published_at=now - timedelta(hours=1),
|
||
reply_deadline=now - timedelta(minutes=1),
|
||
request_msg_id=f"{uuid.uuid4()}:1",
|
||
metrics={"republish_count": MAX_REPUBLISH},
|
||
)
|
||
|
||
assert (await _publish(session_factory, nats)).total == 0
|
||
await subscriber.settle()
|
||
assert subscriber.for_job(job_id) == []
|
||
assert (await jobs.get(job_id)).status is RecognitionStatus.published
|
||
|
||
assert await expire_deadlines(session_factory) == 1
|
||
job = await jobs.get(job_id)
|
||
assert (job.status, job.error_code) == (RecognitionStatus.timeout, "TIMEOUT")
|
||
|
||
|
||
# --- NATS недоступен ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_pass_skipped_when_nats_disconnected(
|
||
session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs
|
||
):
|
||
job_id = await jobs.make_queued_job()
|
||
fake = FakeNats(connected=False)
|
||
|
||
report = await _publish(session_factory, fake)
|
||
|
||
assert report.skipped
|
||
assert fake.calls == []
|
||
assert (await jobs.get(job_id)).status is RecognitionStatus.queued
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_publish_error_keeps_queued_and_stops_pass(
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
jobs: CommittedJobs,
|
||
nats: NatsConnection,
|
||
subscriber: RequestSubscriber,
|
||
):
|
||
"""Сценарий «NATS недоступен»: задание остаётся `queued`, проход прерван; следующий проход публикует."""
|
||
first = await jobs.make_queued_job()
|
||
second = await jobs.make_queued_job()
|
||
fake = FakeNats(fail_first=1)
|
||
|
||
report = await _publish(session_factory, fake)
|
||
|
||
assert (report.failed, report.published) == (1, 0)
|
||
assert len(fake.calls) == 1 # второе задание в этом проходе не пытались публиковать
|
||
for job_id in (first, second):
|
||
job = await jobs.get(job_id)
|
||
assert job.status is RecognitionStatus.queued
|
||
assert job.request_msg_id is None
|
||
assert job.published_at is None
|
||
assert job.reply_deadline is None # откат: срок не сохранён
|
||
|
||
report = await _publish(session_factory, nats)
|
||
|
||
assert report.published == 2 # noqa: PLR2004 - оба задания
|
||
for job_id in (first, second):
|
||
await subscriber.wait_for(job_id, 1)
|
||
assert (await jobs.get(job_id)).status is RecognitionStatus.published
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_republish_error_keeps_published_state(
|
||
session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, nats: NatsConnection
|
||
):
|
||
clock = Clock()
|
||
job_id = await jobs.make_queued_job(reply_deadline=clock() + timedelta(hours=1))
|
||
await _publish(session_factory, nats, clock)
|
||
published_at = (await jobs.get(job_id)).published_at
|
||
clock.advance(ACK + timedelta(seconds=1))
|
||
|
||
report = await _publish(session_factory, FakeNats(fail_first=1), clock)
|
||
|
||
assert report.failed == 1
|
||
job = await jobs.get(job_id)
|
||
assert job.status is RecognitionStatus.published
|
||
assert job.published_at == published_at
|
||
assert job.metrics is None
|
||
|
||
|
||
# --- конкурентность -----------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_concurrent_publishers_publish_each_job_once(
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
jobs: CommittedJobs,
|
||
nats: NatsConnection,
|
||
subscriber: RequestSubscriber,
|
||
):
|
||
"""Два одновременных прохода (SKIP LOCKED): каждое задание опубликовано ровно один раз."""
|
||
job_ids = [await jobs.make_queued_job() for _ in range(CONCURRENT_JOBS)]
|
||
# задержка внутри публикации держит блокировку строки, пока второй проход выбирает задания
|
||
first, second = FakeNats(delegate=nats, delay=0.02), FakeNats(delegate=nats, delay=0.02)
|
||
|
||
reports = await asyncio.gather(_publish(session_factory, first), _publish(session_factory, second))
|
||
|
||
assert sum(r.published for r in reports) == CONCURRENT_JOBS
|
||
assert all(r.published > 0 for r in reports) # оба прохода реально работали параллельно
|
||
published_ids = [msg_id for fake in (first, second) for _, msg_id in fake.calls]
|
||
assert sorted(published_ids) == sorted(f"{job_id}:1" for job_id in job_ids)
|
||
await subscriber.settle()
|
||
for job_id in job_ids:
|
||
assert len(await subscriber.wait_for(job_id, 1)) == 1
|
||
assert (await jobs.get(job_id)).status is RecognitionStatus.published
|
||
|
||
|
||
# --- вахтёр ----------------------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize(
|
||
("status", "deadline_offset", "expected"),
|
||
[
|
||
(RecognitionStatus.processing, timedelta(minutes=-1), RecognitionStatus.timeout),
|
||
(RecognitionStatus.published, timedelta(minutes=-1), RecognitionStatus.timeout),
|
||
(RecognitionStatus.processing, timedelta(minutes=5), RecognitionStatus.processing),
|
||
(RecognitionStatus.published, timedelta(minutes=5), RecognitionStatus.published),
|
||
(RecognitionStatus.processing, None, RecognitionStatus.processing),
|
||
(RecognitionStatus.queued, timedelta(minutes=-1), RecognitionStatus.queued),
|
||
(RecognitionStatus.done, timedelta(minutes=-1), RecognitionStatus.done),
|
||
(RecognitionStatus.dlq, timedelta(minutes=-1), RecognitionStatus.dlq),
|
||
],
|
||
)
|
||
async def test_expire_deadlines(
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
jobs: CommittedJobs,
|
||
status: RecognitionStatus,
|
||
deadline_offset: timedelta | None,
|
||
expected: RecognitionStatus,
|
||
):
|
||
"""Сценарий «Сервис взял, но не ответил»: `processing`/`published` с прошедшим сроком → `timeout`."""
|
||
clock = Clock()
|
||
deadline = None if deadline_offset is None else clock() + deadline_offset
|
||
job_id = await jobs.make_queued_job(status=status, reply_deadline=deadline)
|
||
|
||
expired = await expire_deadlines(session_factory, now=clock)
|
||
|
||
job = await jobs.get(job_id)
|
||
assert job.status is expected
|
||
if expected is RecognitionStatus.timeout:
|
||
assert expired == 1
|
||
assert job.error_code == "TIMEOUT"
|
||
assert job.error_message
|
||
assert job.finished_at == clock()
|
||
assert job.attempt == 1
|
||
else:
|
||
assert expired == 0
|
||
assert job.error_code is None
|
||
assert job.finished_at is None
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize("file_status", [FileStatus.PROCESSING, FileStatus.UPLOADED])
|
||
@pytest.mark.parametrize("job_status", [RecognitionStatus.processing, RecognitionStatus.published])
|
||
async def test_expire_deadlines_moves_current_file_to_error(
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
jobs: CommittedJobs,
|
||
file_status: FileStatus,
|
||
job_status: RecognitionStatus,
|
||
):
|
||
"""Сценарий «Сервис не ответил» (primary-files): задание — `timeout`, файл — `ERROR` в той же транзакции."""
|
||
clock = Clock()
|
||
file_id, job_id = await jobs.make_file_job(
|
||
file_status=file_status, job_status=job_status, reply_deadline=clock() - timedelta(minutes=1)
|
||
)
|
||
|
||
assert await expire_deadlines(session_factory, now=clock) == 1
|
||
|
||
job, file = await jobs.get(job_id), await jobs.file(file_id)
|
||
assert (job.status, job.error_code) == (RecognitionStatus.timeout, "TIMEOUT")
|
||
assert (file.status, file.updated_at) == (FileStatus.ERROR, clock())
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize("file_status", [FileStatus.PROCESSED, FileStatus.SUPERSEDED])
|
||
async def test_expire_deadlines_keeps_processed_and_superseded_file(
|
||
session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, file_status: FileStatus
|
||
):
|
||
clock = Clock()
|
||
file_id, job_id = await jobs.make_file_job(
|
||
file_status=file_status, job_status=RecognitionStatus.processing, reply_deadline=clock() - timedelta(minutes=1)
|
||
)
|
||
|
||
assert await expire_deadlines(session_factory, now=clock) == 1
|
||
|
||
assert (await jobs.get(job_id)).status is RecognitionStatus.timeout
|
||
file = await jobs.file(file_id)
|
||
assert (file.status, file.updated_at) == (file_status, None)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_expire_deadlines_of_non_current_job_keeps_file(
|
||
session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs
|
||
):
|
||
"""Review Focus 1: таймаут старого задания файла, у которого уже новое текущее задание, файл не трогает."""
|
||
clock = Clock()
|
||
file_id, old_job = await jobs.make_file_job(
|
||
file_status=FileStatus.UPLOADED,
|
||
job_status=RecognitionStatus.processing,
|
||
reply_deadline=clock() - timedelta(minutes=1),
|
||
)
|
||
new_job = await jobs.add_job(file_id, status=RecognitionStatus.queued, current=True)
|
||
|
||
assert await expire_deadlines(session_factory, now=clock) == 1
|
||
|
||
assert (await jobs.get(old_job)).status is RecognitionStatus.timeout
|
||
file = await jobs.file(file_id)
|
||
assert (file.status, file.current_recognition_job_id, file.updated_at) == (FileStatus.UPLOADED, new_job, None)
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize("current", [True, False], ids=["current", "non_current"])
|
||
async def test_dlq_moves_current_file_to_error(
|
||
session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs, current: bool
|
||
):
|
||
"""Публикатор переводит задание в `dlq` — файл текущего задания `ERROR`; файл чужого задания не меняется."""
|
||
clock = Clock()
|
||
file_id, job_id = await jobs.make_file_job(
|
||
job_status=RecognitionStatus.published,
|
||
published_at=clock() - ACK * 2,
|
||
request_msg_id="dlq-test:1",
|
||
reply_deadline=clock() + timedelta(hours=1),
|
||
metrics={"republish_count": MAX_REPUBLISH},
|
||
)
|
||
# новое задание — `processing`, чтобы публикатор его не трогал (в проходе только задание старого)
|
||
new_job = None if current else await jobs.add_job(file_id, status=RecognitionStatus.processing, current=True)
|
||
|
||
report = await _publish(session_factory, FakeNats(), clock)
|
||
|
||
assert report.dlq == 1
|
||
assert (await jobs.get(job_id)).status is RecognitionStatus.dlq
|
||
file = await jobs.file(file_id)
|
||
if current:
|
||
assert (file.status, file.updated_at) == (FileStatus.ERROR, clock())
|
||
else:
|
||
assert (file.status, file.current_recognition_job_id, file.updated_at) == (FileStatus.UPLOADED, new_job, None)
|
||
|
||
|
||
# --- фоновые циклы -----------------------------------------------------------------------------------------------
|
||
|
||
|
||
async def _wait_until(predicate: Callable[[], bool], timeout: float = 2.0) -> None:
|
||
deadline = time.monotonic() + timeout
|
||
while not predicate():
|
||
if time.monotonic() > deadline:
|
||
raise AssertionError("условие не выполнено за отведённое время")
|
||
await asyncio.sleep(0.01)
|
||
|
||
|
||
async def _cancel(task: asyncio.Task[Any]) -> None:
|
||
task.cancel()
|
||
with pytest.raises(asyncio.CancelledError):
|
||
await task
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_publisher_loop_publishes_and_stops_on_cancel(
|
||
session_factory: async_sessionmaker[AsyncSession],
|
||
jobs: CommittedJobs,
|
||
nats: NatsConnection,
|
||
subscriber: RequestSubscriber,
|
||
):
|
||
job_id = await jobs.make_queued_job()
|
||
task = asyncio.create_task(
|
||
run_publisher_loop(session_factory, nats, poll_seconds=0.1, ack_timeout=ACK, max_republish=MAX_REPUBLISH)
|
||
)
|
||
try:
|
||
await subscriber.wait_for(job_id, 1, timeout=2.0)
|
||
finally:
|
||
await _cancel(task)
|
||
assert (await jobs.get(job_id)).status is RecognitionStatus.published
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_watchdog_loop_expires_and_stops_on_cancel(
|
||
session_factory: async_sessionmaker[AsyncSession], jobs: CommittedJobs
|
||
):
|
||
job_id = await jobs.make_queued_job(
|
||
status=RecognitionStatus.processing, reply_deadline=datetime.now(UTC) - timedelta(seconds=1)
|
||
)
|
||
task = asyncio.create_task(run_watchdog_loop(session_factory, period_seconds=0.1))
|
||
try:
|
||
deadline = time.monotonic() + 2.0
|
||
while (await jobs.get(job_id)).status is not RecognitionStatus.timeout:
|
||
assert time.monotonic() < deadline, "вахтёр не перевёл задание в timeout за 2 с"
|
||
await asyncio.sleep(0.05)
|
||
finally:
|
||
await _cancel(task)
|
||
|
||
|
||
class RecordingSleep:
|
||
"""Подмена `sleep` циклов: запоминает паузы и только уступает управление."""
|
||
|
||
def __init__(self) -> None:
|
||
self.delays: list[float] = []
|
||
|
||
async def __call__(self, delay: float) -> None:
|
||
self.delays.append(delay)
|
||
await asyncio.sleep(0)
|
||
|
||
|
||
async def test_publisher_loop_survives_pass_errors(monkeypatch: pytest.MonkeyPatch):
|
||
calls: list[int] = []
|
||
|
||
async def failing_pass(*args: Any, **kwargs: Any) -> Any:
|
||
calls.append(1)
|
||
raise RuntimeError("БД недоступна")
|
||
|
||
monkeypatch.setattr(rx, "publish_pending", failing_pass)
|
||
sleep = RecordingSleep()
|
||
task = asyncio.create_task(
|
||
run_publisher_loop(
|
||
object(), # type: ignore[arg-type] # проход подменён: фабрика сессий не используется
|
||
FakeNats(),
|
||
poll_seconds=5,
|
||
ack_timeout=ACK,
|
||
max_republish=MAX_REPUBLISH,
|
||
sleep=sleep,
|
||
)
|
||
)
|
||
try:
|
||
await _wait_until(lambda: len(calls) >= MIN_LOOP_PASSES)
|
||
finally:
|
||
await _cancel(task)
|
||
assert set(sleep.delays) == {5}
|
||
|
||
|
||
async def test_watchdog_loop_survives_pass_errors(monkeypatch: pytest.MonkeyPatch):
|
||
calls: list[int] = []
|
||
|
||
async def failing_pass(*args: Any, **kwargs: Any) -> int:
|
||
calls.append(1)
|
||
raise RuntimeError("БД недоступна")
|
||
|
||
monkeypatch.setattr(rx, "expire_deadlines", failing_pass)
|
||
sleep = RecordingSleep()
|
||
task = asyncio.create_task(run_watchdog_loop(object(), sleep=sleep)) # type: ignore[arg-type]
|
||
try:
|
||
await _wait_until(lambda: len(calls) >= MIN_LOOP_PASSES)
|
||
finally:
|
||
await _cancel(task)
|
||
assert set(sleep.delays) == {WATCHDOG_PERIOD_SECONDS}
|
||
|
||
|
||
async def test_exchange_tasks_start_two_loops_and_stop(monkeypatch: pytest.MonkeyPatch):
|
||
started: dict[str, dict[str, Any]] = {}
|
||
|
||
async def fake_publisher(session_factory: Any, nats: Any, **kwargs: Any) -> None:
|
||
started["publisher"] = kwargs
|
||
await asyncio.Event().wait()
|
||
|
||
async def fake_watchdog(session_factory: Any, **kwargs: Any) -> None:
|
||
started["watchdog"] = kwargs
|
||
await asyncio.Event().wait()
|
||
|
||
monkeypatch.setattr(rx, "run_publisher_loop", fake_publisher)
|
||
monkeypatch.setattr(rx, "run_watchdog_loop", fake_watchdog)
|
||
settings = get_settings().model_copy(
|
||
update={"OUTBOX_POLL_SECONDS": 7, "ACK_TIMEOUT_SECONDS": 11, "OUTBOX_MAX_REPUBLISH": 2}
|
||
)
|
||
|
||
exchange = ExchangeTasks.start(object(), FakeNats(), settings) # type: ignore[arg-type]
|
||
await _wait_until(lambda: len(started) == EXCHANGE_TASKS)
|
||
|
||
assert len(exchange.tasks) == EXCHANGE_TASKS
|
||
assert started["publisher"] == {
|
||
"poll_seconds": 7,
|
||
"ack_timeout": timedelta(seconds=11),
|
||
"max_republish": 2,
|
||
"request_subjects": {"tabular": "files.tabular_uploaded", "image": "files.img_uploaded"},
|
||
}
|
||
assert started["watchdog"] == {"period_seconds": WATCHDOG_PERIOD_SECONDS}
|
||
await exchange.stop()
|
||
assert all(task.done() for task in exchange.tasks)
|
||
await exchange.stop() # повторная остановка безопасна
|