5ae8ff8a81
Решение владельца 08.10.2026: recognition.tabular.request → files.tabular_uploaded, recognition.image.request → files.img_uploaded, recognition.result → files.tabular_parsed (общий subject ack и результатов; files.img_parsed не используется). Тела сообщений, Nats-Msg-Id, queue group'ы и schema_version не меняются. Константы api/src/contract/subjects.py, докстринги, тесты, README, CLAUDE.md; копии схем api/contracts/ из docs/contracts/ побайтно; change nats-files-subjects с дельтой спеки. Сквозной change — RSHB/arm-ka-docs#74. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
1008 lines
49 KiB
Python
1008 lines
49 KiB
Python
"""Inbox обмена `recognition_job`: `handle_message()` (spec recognition-job-exchange: «Ack сервиса переводит задание в
|
||
`processing`», «Inbox принимает результат идемпотентно»; design D-6; рулинги R-7, R-12, R-22). Срез 2c: статусы
|
||
`stored_file` по ack / результату (spec primary-files «Жизненный цикл статуса файла», design D-5, рулинги R-4, R-5);
|
||
`ResultMapper` удалён — inbox не читает S3 и не создаёт документы (маппинг — отдельный цикл, D-1).
|
||
|
||
Быстрые тесты (без маркера) — разбор заголовка `Nats-Msg-Id`, слияние метрик, отказ невалидным сообщениям без
|
||
обращения к БД, ошибка БД не выходит наружу, шаблонный `nats_callback` удалён.
|
||
Интеграционные — `handle_message` с фейковым Msg (`SimpleNamespace(data, headers, subject)`) на реальном PostgreSQL
|
||
(БД сессии pytest; данные фиксируются `CommittedJobs` и удаляются в teardown) и полный цикл через реальный NATS
|
||
(уникальный result-subject: общий `files.tabular_parsed` не трогаем).
|
||
"""
|
||
|
||
import asyncio
|
||
import importlib.util
|
||
import inspect
|
||
import json
|
||
import time
|
||
import uuid
|
||
from collections.abc import AsyncIterator, Iterator
|
||
from datetime import UTC, datetime, timedelta
|
||
from functools import partial
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
|
||
import pytest
|
||
import pytest_asyncio
|
||
from jsonschema import Draft202012Validator, FormatChecker
|
||
from loguru import logger
|
||
from nats.aio.client import Client as NATS
|
||
from nats.aio.msg import Msg
|
||
from sqlalchemy import JSON, func, select
|
||
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
|
||
|
||
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 NatsConnection
|
||
from src.models.enums import FileStatus, RecognitionStatus, S3BucketRole
|
||
from src.models.files import RecognitionJob, S3Object, StoredFile
|
||
from src.services import recognition_exchange as rx
|
||
from src.services.recognition_exchange import (
|
||
FINAL_STATUSES,
|
||
expire_deadlines,
|
||
handle_message,
|
||
header_msg_id,
|
||
merge_metrics,
|
||
publish_pending,
|
||
)
|
||
from tests.factories import CommittedJobs
|
||
|
||
API_DIR = Path(__file__).resolve().parents[1]
|
||
MESSAGE_SCHEMA = json.loads((API_DIR / "contracts" / "recognition-result-message.schema.json").read_text("utf-8"))
|
||
MESSAGE_VALIDATOR = Draft202012Validator(MESSAGE_SCHEMA, format_checker=FormatChecker())
|
||
|
||
SERVICE = {"name": "tabular-parser", "version": "0.1.0"}
|
||
BUCKET = "arm-ka"
|
||
READY_TIMEOUT = 5.0
|
||
RACE_JOBS = 10
|
||
LOCK_HOLD_SECONDS = 0.5
|
||
_ERROR = {"code": "UNSUPPORTED_FORMAT", "message": "формат файла не поддерживается"}
|
||
|
||
|
||
# --- инфраструктура ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
def logs() -> Iterator[list[Any]]:
|
||
"""Записи loguru (уровень DEBUG и выше), выпущенные за время теста."""
|
||
records: list[Any] = []
|
||
handler_id = logger.add(lambda message: records.append(message.record), level="DEBUG")
|
||
try:
|
||
yield records
|
||
finally:
|
||
logger.remove(handler_id)
|
||
|
||
|
||
def _logged(records: list[Any], fragment: str, level: str | None = None) -> list[Any]:
|
||
return [r for r in records if fragment in r["message"] and (level is None or r["level"].name == level)]
|
||
|
||
|
||
def _result_key(file_id: uuid.UUID, job_id: uuid.UUID) -> str:
|
||
return f"results/{file_id}/{job_id}/result.json"
|
||
|
||
|
||
def envelope(job_id: uuid.UUID, file_id: uuid.UUID, status: str, **extra: Any) -> dict[str, Any]:
|
||
"""Конверт `recognition-result-message` v1.0; для done/needs_check по умолчанию — `result_s3` и версия схемы."""
|
||
body: dict[str, Any] = {
|
||
"schema_version": "1.0",
|
||
"job_id": str(job_id),
|
||
"file_id": str(file_id),
|
||
"status": status,
|
||
"service": SERVICE,
|
||
}
|
||
if status in ("done", "needs_check"):
|
||
body["result_s3"] = {"bucket": BUCKET, "key": _result_key(file_id, job_id)}
|
||
body["result_schema_version"] = "1.0"
|
||
body.update(extra)
|
||
return body
|
||
|
||
|
||
def fake_msg(body: Any, msg_id: str | None, *, header: str = NATS_MSG_ID_HEADER, validate: bool = True) -> Any:
|
||
"""Msg-подобный объект nats-py: `.data`, `.headers`, `.subject`. Тело по умолчанию проверяется JSON Schema."""
|
||
if validate:
|
||
MESSAGE_VALIDATOR.validate(body)
|
||
data = body if isinstance(body, bytes) else json.dumps(body, ensure_ascii=False).encode("utf-8")
|
||
headers = None if msg_id is None else {header: msg_id}
|
||
return SimpleNamespace(data=data, headers=headers, subject="files.tabular_parsed")
|
||
|
||
|
||
class SlowFileStatus:
|
||
"""Подмена `set_file_status_for_job` в inbox: запоминает вызовы (id задания, статус) и держит паузу `delay`
|
||
ВНУТРИ транзакции сообщения (строка задания остаётся заблокированной), затем вызывает настоящую функцию."""
|
||
|
||
def __init__(self, monkeypatch: pytest.MonkeyPatch, *, delay: float) -> None:
|
||
self.delay = delay
|
||
self.calls: list[tuple[uuid.UUID, FileStatus]] = []
|
||
self._original = rx.set_file_status_for_job
|
||
monkeypatch.setattr(rx, "set_file_status_for_job", self)
|
||
|
||
async def __call__(self, session: AsyncSession, job: RecognitionJob, status: FileStatus, **kwargs: Any) -> bool:
|
||
self.calls.append((job.id, status))
|
||
await asyncio.sleep(self.delay)
|
||
return await self._original(session, job, status, **kwargs)
|
||
|
||
|
||
@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()
|
||
|
||
|
||
async def _row(session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID) -> dict[str, Any]:
|
||
"""Снимок всех колонок строки `recognition_job`."""
|
||
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 _s3_total(session_factory: async_sessionmaker[AsyncSession]) -> int:
|
||
async with session_factory() as session:
|
||
return int((await session.execute(select(func.count()).select_from(S3Object))).scalar_one())
|
||
|
||
|
||
async def _file_row(session_factory: async_sessionmaker[AsyncSession], file_id: uuid.UUID) -> dict[str, Any]:
|
||
"""Снимок всех колонок строки `stored_file`."""
|
||
async with session_factory() as session:
|
||
row = (await session.execute(select(StoredFile.__table__).where(StoredFile.id == file_id))).one()
|
||
return dict(row._mapping)
|
||
|
||
|
||
async def _file_id(jobs: CommittedJobs, job_id: uuid.UUID) -> uuid.UUID:
|
||
return (await jobs.get(job_id)).file_id
|
||
|
||
|
||
async def _handle(session_factory: async_sessionmaker[AsyncSession], msg: Any, **kwargs: Any) -> None:
|
||
await handle_message(msg, session_factory=session_factory, **kwargs)
|
||
|
||
|
||
# --- быстрые тесты: заголовок, метрики, разбор ------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("headers", "expected"),
|
||
[
|
||
({"Nats-Msg-Id": "result:x:1"}, "result:x:1"),
|
||
({"nats-msg-id": "result:x:1"}, "result:x:1"),
|
||
({"NATS-MSG-ID": "ack:x:1", "other": "y"}, "ack:x:1"),
|
||
({"other": "y"}, None),
|
||
({"Nats-Msg-Id": ""}, None),
|
||
({}, None),
|
||
(None, None),
|
||
],
|
||
)
|
||
def test_header_msg_id_is_case_insensitive(headers: dict[str, str] | None, expected: str | None):
|
||
assert header_msg_id(headers) == expected
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("existing", "incoming", "expected"),
|
||
[
|
||
({"republish_count": 2}, {"duration_ms": 5}, {"republish_count": 2, "duration_ms": 5}),
|
||
({"republish_count": 2, "rows": 1}, {"rows": 7}, {"republish_count": 2, "rows": 7}),
|
||
(None, {"duration_ms": 5}, {"duration_ms": 5}), # SQL NULL / jsonb null (старые строки, сырой SQL)
|
||
([None, {"x": 1}], {"duration_ms": 5}, {"duration_ms": 5}), # не объект — как пустой
|
||
({"republish_count": 1}, None, {"republish_count": 1}), # метрик сервиса нет — без изменений
|
||
(None, None, None),
|
||
],
|
||
)
|
||
def test_merge_metrics_keeps_republish_count(existing: Any, incoming: Any, expected: Any):
|
||
assert merge_metrics(existing, incoming) == expected
|
||
|
||
|
||
class _UntouchableFactory:
|
||
"""session_factory, к которому нельзя обращаться: доказывает, что сообщение отвергнуто до БД."""
|
||
|
||
def __call__(self) -> Any:
|
||
raise AssertionError("невалидное сообщение не должно открывать сессию БД")
|
||
|
||
|
||
_JOB = uuid.UUID("11111111-1111-4111-8111-111111111111")
|
||
_FILE = uuid.UUID("22222222-2222-4222-8222-222222222222")
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"data",
|
||
[
|
||
b"{not json",
|
||
b"\xff\xfe",
|
||
b"[1, 2]",
|
||
b'"text"',
|
||
json.dumps({"schema_version": "1.0", "status": "done"}).encode(), # нет job_id
|
||
json.dumps(envelope(_JOB, _FILE, "done", schema_version="2.0")).encode(), # чужой MAJOR
|
||
json.dumps({k: v for k, v in envelope(_JOB, _FILE, "done").items() if k != "result_s3"}).encode(),
|
||
json.dumps(envelope(_JOB, _FILE, "error")).encode(), # error без поля error
|
||
json.dumps(envelope(_JOB, _FILE, "finished")).encode(), # статус вне enum
|
||
json.dumps(envelope(_JOB, _FILE, "done", service={"name": "other", "version": "1"})).encode(),
|
||
json.dumps({**envelope(_JOB, _FILE, "done"), "job_id": "not-a-uuid"}).encode(),
|
||
],
|
||
)
|
||
async def test_invalid_message_is_logged_and_does_not_touch_db(data: bytes, logs: list[Any]):
|
||
msg = SimpleNamespace(data=data, headers={NATS_MSG_ID_HEADER: "result:x:1"}, subject="files.tabular_parsed")
|
||
await handle_message(msg, session_factory=_UntouchableFactory()) # type: ignore[arg-type]
|
||
rejected = _logged(logs, "отклонено", "WARNING")
|
||
assert len(rejected) == 1
|
||
# тело целиком в лог не попадает
|
||
assert "result.json" not in rejected[0]["message"]
|
||
|
||
|
||
async def test_invalid_message_log_carries_job_id_when_readable(logs: list[Any]):
|
||
body = envelope(_JOB, _FILE, "error") # нет обязательного error
|
||
msg = fake_msg(body, "result:x:1", validate=False)
|
||
await handle_message(msg, session_factory=_UntouchableFactory()) # type: ignore[arg-type]
|
||
(record,) = _logged(logs, "отклонено", "WARNING")
|
||
assert str(_JOB) in record["message"]
|
||
|
||
|
||
async def test_invalid_message_job_id_is_escaped_in_log(logs: list[Any]):
|
||
"""job_id невалидного сообщения с переводом строки попадает в лог через repr: подделать строку лога нельзя (M-7)."""
|
||
forged = "not-a-uuid\n2026-10-04 | INFO | подделанная запись\r\x1b[31m"
|
||
body = {**envelope(_JOB, _FILE, "done"), "job_id": forged}
|
||
msg = fake_msg(body, "result:x:1", validate=False)
|
||
await handle_message(msg, session_factory=_UntouchableFactory()) # type: ignore[arg-type]
|
||
(record,) = _logged(logs, "отклонено", "WARNING")
|
||
message = record["message"]
|
||
assert "\n" not in message and "\r" not in message and "\x1b" not in message
|
||
assert repr(forged) in message
|
||
|
||
|
||
async def test_database_error_is_logged_and_not_raised(logs: list[Any]):
|
||
"""Ошибка БД (сервер недоступен) — `logger.exception`, без повторов, наружу не выходит."""
|
||
engine = create_async_engine("postgresql+asyncpg://nobody:nothing@127.0.0.1:1/none")
|
||
try:
|
||
factory = async_sessionmaker(engine, expire_on_commit=False)
|
||
await handle_message(
|
||
fake_msg(envelope(_JOB, _FILE, "processing"), ack_msg_id(_JOB, 1)), session_factory=factory
|
||
)
|
||
finally:
|
||
await engine.dispose()
|
||
(record,) = _logged(logs, "транзакция откачена", "ERROR")
|
||
assert record["exception"] is not None
|
||
|
||
|
||
def test_inbox_has_no_result_mapper():
|
||
"""OQ-67 (Срез 2c, D-1): точки расширения `ResultMapper` / `NoopMapper` нет, `handle_message` без `mapper`."""
|
||
assert not hasattr(rx, "ResultMapper")
|
||
assert not hasattr(rx, "NoopMapper")
|
||
assert "mapper" not in inspect.signature(handle_message).parameters
|
||
|
||
|
||
def test_final_statuses():
|
||
"""R-22 (design #34): финальные для inbox — done, needs_check, error, timeout, dlq."""
|
||
assert {s.value for s in FINAL_STATUSES} == {"done", "needs_check", "error", "timeout", "dlq"}
|
||
|
||
|
||
def test_template_nats_callback_removed():
|
||
"""R-3 (design #16): шаблонный `core/nats_callback.py` удалён и нигде в `src/` не упоминается."""
|
||
assert importlib.util.find_spec("src.core.nats_callback") is None
|
||
offenders = [p for p in (API_DIR / "src").rglob("*.py") if "nats_callback" in p.read_text(encoding="utf-8")]
|
||
assert offenders == []
|
||
|
||
|
||
# --- ack: processing ---------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_ack_moves_published_job_to_processing(session_factory, jobs: CommittedJobs, logs: list[Any]):
|
||
"""Сценарий «Сервис подтвердил»."""
|
||
job_id = await jobs.make_queued_job(status=RecognitionStatus.published, request_msg_id=f"{uuid.uuid4()}:1")
|
||
file_id = await _file_id(jobs, job_id)
|
||
before = await _row(session_factory, job_id)
|
||
moment = datetime(2026, 10, 4, 12, 0, tzinfo=UTC)
|
||
|
||
await _handle(
|
||
session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)), now=lambda: moment
|
||
)
|
||
|
||
after = await _row(session_factory, job_id)
|
||
assert after["status"] is RecognitionStatus.processing
|
||
assert after["started_at"] == moment
|
||
assert (after["service_name"], after["service_version"]) == ("tabular-parser", "0.1.0")
|
||
changed = {k for k in after if after[k] != before[k]}
|
||
assert changed == {"status", "started_at", "service_name", "service_version"}
|
||
assert _logged(logs, "ack", "INFO")
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_repeated_ack_changes_nothing(session_factory, jobs: CommittedJobs):
|
||
job_id = await jobs.make_queued_job(status=RecognitionStatus.published)
|
||
file_id = await _file_id(jobs, job_id)
|
||
ack = fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1))
|
||
await _handle(session_factory, ack)
|
||
first = await _row(session_factory, job_id)
|
||
|
||
await _handle(session_factory, ack, now=lambda: datetime.now(UTC) + timedelta(minutes=1))
|
||
|
||
assert await _row(session_factory, job_id) == first
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize(
|
||
"status",
|
||
[
|
||
RecognitionStatus.queued,
|
||
RecognitionStatus.done,
|
||
RecognitionStatus.needs_check,
|
||
RecognitionStatus.error,
|
||
RecognitionStatus.timeout,
|
||
RecognitionStatus.dlq,
|
||
],
|
||
)
|
||
async def test_ack_for_other_status_is_logged_and_ignored(
|
||
session_factory, jobs: CommittedJobs, logs: list[Any], status: RecognitionStatus
|
||
):
|
||
job_id = await jobs.make_queued_job(status=status)
|
||
file_id = await _file_id(jobs, job_id)
|
||
before = await _row(session_factory, job_id)
|
||
|
||
await _handle(session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)))
|
||
|
||
assert await _row(session_factory, job_id) == before
|
||
(record,) = _logged(logs, "задание не изменено", "WARNING")
|
||
assert status.value in record["message"]
|
||
|
||
|
||
# --- результат ---------------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_done_result_registers_s3_object_and_updates_job(session_factory, jobs: CommittedJobs):
|
||
"""Сценарий «Успешный результат»: s3_object (results), result_msg_id, finished_at, метрики слиты."""
|
||
job_id = await jobs.make_queued_job(status=RecognitionStatus.processing, metrics={"republish_count": 1})
|
||
file_id = await _file_id(jobs, job_id)
|
||
moment = datetime(2026, 10, 4, 12, 30, tzinfo=UTC)
|
||
metrics = {"duration_ms": 1234, "rows": 50, "fields_total": 300, "fields_low_confidence": 4,
|
||
"fields_unrecognized": 2, "warnings": ["лист без шапки"]} # fmt: skip
|
||
body = envelope(job_id, file_id, "done", metrics=metrics, result_schema_version="1.1")
|
||
before_total = await _s3_total(session_factory)
|
||
|
||
await _handle(session_factory, fake_msg(body, result_msg_id(job_id, 1)), now=lambda: moment)
|
||
|
||
job = await _row(session_factory, job_id)
|
||
(s3,) = await _s3_rows(session_factory, _result_key(file_id, job_id))
|
||
assert await _s3_total(session_factory) == before_total + 1
|
||
assert job["status"] is RecognitionStatus.done
|
||
assert job["result_msg_id"] == f"result:{job_id}:1"
|
||
assert job["result_s3_object_id"] == s3["id"]
|
||
assert job["finished_at"] == moment
|
||
assert job["result_schema_version"] == "1.1"
|
||
assert (job["service_name"], job["service_version"]) == ("tabular-parser", "0.1.0")
|
||
assert job["metrics"] == {"republish_count": 1, **metrics}
|
||
assert (job["fields_total"], job["fields_low_confidence"], job["fields_unrecognized"]) == (300, 4, 2)
|
||
assert (job["error_code"], job["error_message"]) == (None, None)
|
||
assert s3["bucket_role"] is S3BucketRole.results
|
||
assert (s3["bucket"], s3["created_by"], s3["created_at"]) == (BUCKET, "tabular-parser", moment)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_needs_check_result(session_factory, jobs: CommittedJobs):
|
||
job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
|
||
file_id = await _file_id(jobs, job_id)
|
||
|
||
await _handle(session_factory, fake_msg(envelope(job_id, file_id, "needs_check"), result_msg_id(job_id, 1)))
|
||
|
||
job = await _row(session_factory, job_id)
|
||
assert job["status"] is RecognitionStatus.needs_check
|
||
assert job["result_s3_object_id"] is not None
|
||
assert job["metrics"] is None # метрик не было ни у задания, ни у сервиса: колонка не трогается
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize(
|
||
("stored", "expected_state"),
|
||
[(None, (True, None)), (JSON.NULL, (False, "null"))],
|
||
ids=["sql_null", "json_null"],
|
||
)
|
||
async def test_result_metrics_merge_over_null_metrics(
|
||
session_factory, jobs: CommittedJobs, stored: Any, expected_state: Any
|
||
):
|
||
"""Слияние метрик в inbox (Python `merge_metrics`) работает и для SQL NULL, и для jsonb null (финальное ревью I-1);
|
||
результат без метрик колонку не трогает: SQL NULL остаётся SQL NULL."""
|
||
with_metrics = await jobs.make_queued_job(status=RecognitionStatus.processing, metrics=stored)
|
||
without_metrics = await jobs.make_queued_job(status=RecognitionStatus.processing, metrics=stored)
|
||
assert await jobs.metrics_sql_state(with_metrics) == expected_state
|
||
|
||
file_id = await _file_id(jobs, with_metrics)
|
||
body = envelope(with_metrics, file_id, "done", metrics={"duration_ms": 10})
|
||
await _handle(session_factory, fake_msg(body, result_msg_id(with_metrics, 1)))
|
||
file_id = await _file_id(jobs, without_metrics)
|
||
await _handle(
|
||
session_factory, fake_msg(envelope(without_metrics, file_id, "done"), result_msg_id(without_metrics, 1))
|
||
)
|
||
|
||
job = await _row(session_factory, with_metrics)
|
||
assert job["status"] is RecognitionStatus.done
|
||
assert job["metrics"] == {"duration_ms": 10}
|
||
assert await jobs.metrics_sql_state(with_metrics) == (False, "object")
|
||
assert (await _row(session_factory, without_metrics))["status"] is RecognitionStatus.done
|
||
assert await jobs.metrics_sql_state(without_metrics) == expected_state
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_redelivery_changes_nothing(session_factory, jobs: CommittedJobs, logs: list[Any]):
|
||
"""Сценарий «Повторная доставка»: ни recognition_job, ни s3_object, ни stored_file не меняются."""
|
||
file_id, job_id = await jobs.make_file_job(
|
||
file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing
|
||
)
|
||
msg = fake_msg(envelope(job_id, file_id, "done", metrics={"rows": 3}), result_msg_id(job_id, 1))
|
||
await _handle(session_factory, msg)
|
||
job_before = await _row(session_factory, job_id)
|
||
file_before = await _file_row(session_factory, file_id)
|
||
s3_before = await _s3_rows(session_factory, _result_key(file_id, job_id))
|
||
total_before = await _s3_total(session_factory)
|
||
|
||
await _handle(session_factory, msg, now=lambda: datetime.now(UTC) + timedelta(hours=1))
|
||
|
||
assert await _row(session_factory, job_id) == job_before
|
||
assert await _s3_rows(session_factory, _result_key(file_id, job_id)) == s3_before
|
||
assert await _s3_total(session_factory) == total_before
|
||
assert await _file_row(session_factory, file_id) == file_before
|
||
assert _logged(logs, "дубликат", "DEBUG")
|
||
assert not _logged(logs, "финальном статусе")
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_service_error_without_result_s3(session_factory, jobs: CommittedJobs):
|
||
"""Сценарий «Ошибка сервиса»: NOT_IMPLEMENTED без result_s3."""
|
||
job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
|
||
file_id = await _file_id(jobs, job_id)
|
||
total_before = await _s3_total(session_factory)
|
||
body = envelope(job_id, file_id, "error", error={"code": "NOT_IMPLEMENTED", "message": "разбор ОСВ — Срез 1"})
|
||
|
||
await _handle(session_factory, fake_msg(body, result_msg_id(job_id, 1)))
|
||
|
||
job = await _row(session_factory, job_id)
|
||
assert job["status"] is RecognitionStatus.error
|
||
assert (job["error_code"], job["error_message"]) == ("NOT_IMPLEMENTED", "разбор ОСВ — Срез 1")
|
||
assert job["result_s3_object_id"] is None
|
||
assert job["result_msg_id"] == f"result:{job_id}:1"
|
||
assert job["finished_at"] is not None
|
||
assert job["result_schema_version"] is None
|
||
assert await _s3_total(session_factory) == total_before
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize(
|
||
"status",
|
||
[
|
||
RecognitionStatus.timeout,
|
||
RecognitionStatus.dlq,
|
||
RecognitionStatus.error,
|
||
RecognitionStatus.done,
|
||
RecognitionStatus.needs_check,
|
||
],
|
||
)
|
||
async def test_result_for_final_status_is_logged_and_ignored(
|
||
session_factory, jobs: CommittedJobs, logs: list[Any], status: RecognitionStatus
|
||
):
|
||
"""Сценарий «Результат пришёл после таймаута» (и прочие финальные статусы, R-22): задание не меняется."""
|
||
other = f"result:{uuid.uuid4()}:9" if status in (RecognitionStatus.done, RecognitionStatus.needs_check) else None
|
||
job_id = await jobs.make_queued_job(status=status, result_msg_id=other)
|
||
file_id = await _file_id(jobs, job_id)
|
||
before = await _row(session_factory, job_id)
|
||
total_before = await _s3_total(session_factory)
|
||
|
||
await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)))
|
||
|
||
assert await _row(session_factory, job_id) == before
|
||
assert await _s3_total(session_factory) == total_before
|
||
(record,) = _logged(logs, "финальном статусе", "WARNING")
|
||
assert status.value in record["message"]
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_unknown_job_is_logged_and_creates_nothing(session_factory, logs: list[Any]):
|
||
job_id, file_id = uuid.uuid4(), uuid.uuid4()
|
||
total_before = await _s3_total(session_factory)
|
||
|
||
await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)))
|
||
|
||
assert await _s3_total(session_factory) == total_before
|
||
assert await _s3_rows(session_factory, _result_key(file_id, job_id)) == []
|
||
(record,) = _logged(logs, "неизвестного задания", "WARNING")
|
||
assert str(job_id) in record["message"]
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_file_id_mismatch_is_logged_and_ignored(session_factory, jobs: CommittedJobs, logs: list[Any]):
|
||
job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
|
||
before = await _row(session_factory, job_id)
|
||
total_before = await _s3_total(session_factory)
|
||
|
||
await _handle(session_factory, fake_msg(envelope(job_id, uuid.uuid4(), "done"), result_msg_id(job_id, 1)))
|
||
|
||
assert await _row(session_factory, job_id) == before
|
||
assert await _s3_total(session_factory) == total_before
|
||
assert _logged(logs, "file_id", "WARNING")
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize("status", [RecognitionStatus.published, RecognitionStatus.queued])
|
||
async def test_result_without_ack_is_accepted(session_factory, jobs: CommittedJobs, status: RecognitionStatus):
|
||
"""R-22: потерянный ack не блокирует результат; `started_at` остаётся пустым (начало обработки неизвестно)."""
|
||
job_id = await jobs.make_queued_job(status=status)
|
||
file_id = await _file_id(jobs, job_id)
|
||
|
||
await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)))
|
||
|
||
job = await _row(session_factory, job_id)
|
||
assert job["status"] is RecognitionStatus.done
|
||
assert job["started_at"] is None
|
||
assert job["service_name"] == "tabular-parser"
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_msg_id_header_case_insensitive(session_factory, jobs: CommittedJobs):
|
||
job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
|
||
file_id = await _file_id(jobs, job_id)
|
||
|
||
await _handle(
|
||
session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1), header="nats-msg-id")
|
||
)
|
||
|
||
assert (await _row(session_factory, job_id))["result_msg_id"] == f"result:{job_id}:1"
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_missing_msg_id_header_uses_derived_id(session_factory, jobs: CommittedJobs, logs: list[Any]):
|
||
"""Нет `Nats-Msg-Id` — предупреждение и производный `result:<job_id>:<attempt>`; повтор — дубликат."""
|
||
job_id = await jobs.make_queued_job(status=RecognitionStatus.processing, attempt=2)
|
||
file_id = await _file_id(jobs, job_id)
|
||
msg = fake_msg(envelope(job_id, file_id, "done"), None)
|
||
|
||
await _handle(session_factory, msg)
|
||
job = await _row(session_factory, job_id)
|
||
await _handle(session_factory, msg)
|
||
|
||
assert job["status"] is RecognitionStatus.done
|
||
assert job["result_msg_id"] == f"result:{job_id}:2"
|
||
assert await _row(session_factory, job_id) == job
|
||
assert len(_logged(logs, "без заголовка Nats-Msg-Id", "WARNING")) == 2 # noqa: PLR2004 - оба сообщения
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_existing_s3_object_is_reused(session_factory, jobs: CommittedJobs):
|
||
"""Ключ результата уже зарегистрирован — строка переиспользуется, дубль (bucket, key) не вставляется."""
|
||
job_id = await jobs.make_queued_job(status=RecognitionStatus.processing)
|
||
file_id = await _file_id(jobs, job_id)
|
||
async with session_factory() as session, session.begin():
|
||
existing = S3Object(
|
||
bucket_role=S3BucketRole.results,
|
||
bucket=BUCKET,
|
||
key=_result_key(file_id, job_id),
|
||
created_by="tabular-parser",
|
||
created_at=datetime.now(UTC),
|
||
)
|
||
session.add(existing)
|
||
await session.flush()
|
||
existing_id = existing.id
|
||
jobs.track_s3_object(existing_id)
|
||
|
||
await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)))
|
||
|
||
assert (await _row(session_factory, job_id))["result_s3_object_id"] == existing_id
|
||
assert len(await _s3_rows(session_factory, _result_key(file_id, job_id))) == 1
|
||
|
||
|
||
# --- статусы stored_file (Срез 2c: spec primary-files «Жизненный цикл статуса файла», D-5, R-4, R-5) -------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_ack_moves_uploaded_file_to_processing(session_factory, jobs: CommittedJobs):
|
||
"""Сценарий «Сервис подтвердил»: файл текущего задания — `PROCESSING`, `updated_at` — момент ack."""
|
||
file_id, job_id = await jobs.make_file_job(job_status=RecognitionStatus.published)
|
||
moment = datetime(2026, 10, 5, 9, 0, tzinfo=UTC)
|
||
|
||
await _handle(
|
||
session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)), now=lambda: moment
|
||
)
|
||
|
||
file = await _file_row(session_factory, file_id)
|
||
assert file["status"] is FileStatus.PROCESSING
|
||
assert file["updated_at"] == moment
|
||
assert file["current_recognition_job_id"] == job_id
|
||
assert (await _row(session_factory, job_id))["status"] is RecognitionStatus.processing
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize(
|
||
"file_status", [FileStatus.ERROR, FileStatus.PROCESSED, FileStatus.SUPERSEDED, FileStatus.PROCESSING]
|
||
)
|
||
async def test_ack_moves_file_only_from_uploaded(session_factory, jobs: CommittedJobs, file_status: FileStatus):
|
||
"""R-5: ack ставит `PROCESSING` только из `UPLOADED` — запоздавший ack не воскрешает `ERROR` и не трогает
|
||
`PROCESSED`/`SUPERSEDED`; задание при этом подтверждается как обычно."""
|
||
file_id, job_id = await jobs.make_file_job(file_status=file_status, job_status=RecognitionStatus.published)
|
||
before = await _file_row(session_factory, file_id)
|
||
|
||
await _handle(session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)))
|
||
|
||
assert await _file_row(session_factory, file_id) == before
|
||
assert (await _row(session_factory, job_id))["status"] is RecognitionStatus.processing
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize("file_status", [FileStatus.PROCESSING, FileStatus.UPLOADED])
|
||
async def test_error_result_moves_file_to_error(session_factory, jobs: CommittedJobs, file_status: FileStatus):
|
||
"""Сценарий «Парсер вернул ошибку»: файл — `ERROR`, причина — в задании (`error_code`)."""
|
||
file_id, job_id = await jobs.make_file_job(file_status=file_status, job_status=RecognitionStatus.processing)
|
||
moment = datetime(2026, 10, 5, 9, 30, tzinfo=UTC)
|
||
|
||
await _handle(
|
||
session_factory,
|
||
fake_msg(envelope(job_id, file_id, "error", error=_ERROR), result_msg_id(job_id, 1)),
|
||
now=lambda: moment,
|
||
)
|
||
|
||
file = await _file_row(session_factory, file_id)
|
||
assert (file["status"], file["updated_at"]) == (FileStatus.ERROR, moment)
|
||
assert (await _row(session_factory, job_id))["error_code"] == "UNSUPPORTED_FORMAT"
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize("status", ["done", "needs_check"])
|
||
async def test_success_result_leaves_processing_file_for_mapping(session_factory, jobs: CommittedJobs, status: str):
|
||
"""Сценарий «Успешный результат»: файл остаётся `PROCESSING` до маппинга; строка файла не меняется вовсе."""
|
||
file_id, job_id = await jobs.make_file_job(
|
||
file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing
|
||
)
|
||
before = await _file_row(session_factory, file_id)
|
||
|
||
await _handle(session_factory, fake_msg(envelope(job_id, file_id, status), result_msg_id(job_id, 1)))
|
||
|
||
assert await _file_row(session_factory, file_id) == before
|
||
assert (await _row(session_factory, job_id))["status"] is RecognitionStatus(status)
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize("status", ["done", "needs_check"])
|
||
@pytest.mark.parametrize("job_status", [RecognitionStatus.published, RecognitionStatus.queued])
|
||
async def test_success_result_without_ack_moves_uploaded_file_to_processing(
|
||
session_factory, jobs: CommittedJobs, status: str, job_status: RecognitionStatus
|
||
):
|
||
"""R-4: ack потерялся (core NATS без JetStream) — результат `done`/`needs_check` переводит файл `UPLOADED →
|
||
PROCESSING`, иначе файл с готовым результатом не попал бы в цикл маппинга."""
|
||
file_id, job_id = await jobs.make_file_job(job_status=job_status)
|
||
moment = datetime(2026, 10, 5, 10, 0, tzinfo=UTC)
|
||
|
||
await _handle(
|
||
session_factory, fake_msg(envelope(job_id, file_id, status), result_msg_id(job_id, 1)), now=lambda: moment
|
||
)
|
||
|
||
file = await _file_row(session_factory, file_id)
|
||
assert (file["status"], file["updated_at"]) == (FileStatus.PROCESSING, moment)
|
||
assert (await _row(session_factory, job_id))["status"] is RecognitionStatus(status)
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize("file_status", [FileStatus.PROCESSED, FileStatus.SUPERSEDED])
|
||
@pytest.mark.parametrize("status", ["done", "error"])
|
||
async def test_result_does_not_touch_processed_or_superseded_file(
|
||
session_factory, jobs: CommittedJobs, file_status: FileStatus, status: str
|
||
):
|
||
"""`PROCESSED` и `SUPERSEDED` — итог маппинга / автозамены: результат обмена их не перезаписывает."""
|
||
file_id, job_id = await jobs.make_file_job(file_status=file_status, job_status=RecognitionStatus.processing)
|
||
before = await _file_row(session_factory, file_id)
|
||
extra = {"error": _ERROR} if status == "error" else {}
|
||
|
||
await _handle(session_factory, fake_msg(envelope(job_id, file_id, status, **extra), result_msg_id(job_id, 1)))
|
||
|
||
assert await _file_row(session_factory, file_id) == before
|
||
assert (await _row(session_factory, job_id))["status"] is RecognitionStatus(status)
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.parametrize(
|
||
("status", "msg_id_fn", "expected_job"),
|
||
[
|
||
("processing", ack_msg_id, RecognitionStatus.processing),
|
||
("done", result_msg_id, RecognitionStatus.done),
|
||
("needs_check", result_msg_id, RecognitionStatus.needs_check),
|
||
("error", result_msg_id, RecognitionStatus.error),
|
||
],
|
||
)
|
||
async def test_message_of_non_current_job_does_not_touch_file( # noqa: PLR0913 - фикстуры и параметры
|
||
session_factory, jobs: CommittedJobs, status: str, msg_id_fn: Any, expected_job: RecognitionStatus
|
||
):
|
||
"""Review Focus 1: у файла два задания, текущее — новое; ack/результат старого задания применяется к старому
|
||
заданию, но файл (статус, `updated_at`, `current_recognition_job_id`) не трогает."""
|
||
file_id, old_job = await jobs.make_file_job(job_status=RecognitionStatus.published)
|
||
new_job = await jobs.add_job(file_id, status=RecognitionStatus.queued, current=True)
|
||
before = await _file_row(session_factory, file_id)
|
||
assert before["current_recognition_job_id"] == new_job
|
||
extra = {"error": _ERROR} if status == "error" else {}
|
||
|
||
await _handle(session_factory, fake_msg(envelope(old_job, file_id, status, **extra), msg_id_fn(old_job, 1)))
|
||
|
||
assert (await _row(session_factory, old_job))["status"] is expected_job
|
||
assert await _file_row(session_factory, file_id) == before
|
||
assert (await _row(session_factory, new_job))["status"] is RecognitionStatus.queued
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_file_status_rolls_back_with_message_transaction(
|
||
session_factory, jobs: CommittedJobs, logs: list[Any], monkeypatch: pytest.MonkeyPatch
|
||
):
|
||
"""Статус файла меняется в транзакции сообщения: ошибка после обновления задания откатывает и задание, и файл."""
|
||
file_id, job_id = await jobs.make_file_job(
|
||
file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing
|
||
)
|
||
job_before = await _row(session_factory, job_id)
|
||
file_before = await _file_row(session_factory, file_id)
|
||
original = rx.set_file_status_for_job
|
||
|
||
async def failing(session: AsyncSession, job: RecognitionJob, status: FileStatus, **kwargs: Any) -> bool:
|
||
await original(session, job, status, **kwargs)
|
||
raise RuntimeError("подставная ошибка после обновления файла")
|
||
|
||
monkeypatch.setattr(rx, "set_file_status_for_job", failing)
|
||
|
||
await _handle(session_factory, fake_msg(envelope(job_id, file_id, "error", error=_ERROR), result_msg_id(job_id, 1)))
|
||
|
||
assert await _row(session_factory, job_id) == job_before
|
||
assert await _file_row(session_factory, file_id) == file_before
|
||
(record,) = _logged(logs, "транзакция откачена", "ERROR")
|
||
assert record["exception"] is not None
|
||
|
||
|
||
# --- гонки --------------------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_concurrent_same_message_applied_once(
|
||
session_factory, jobs: CommittedJobs, logs: list[Any], monkeypatch: pytest.MonkeyPatch
|
||
):
|
||
file_id, job_id = await jobs.make_file_job(job_status=RecognitionStatus.processing)
|
||
msg = fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1))
|
||
slow = SlowFileStatus(monkeypatch, delay=LOCK_HOLD_SECONDS) # первый держит строку, второй ждёт его commit
|
||
|
||
await asyncio.gather(*[_handle(session_factory, msg) for _ in range(2)])
|
||
|
||
assert len(await _s3_rows(session_factory, _result_key(file_id, job_id))) == 1
|
||
assert (await _row(session_factory, job_id))["result_msg_id"] == f"result:{job_id}:1"
|
||
assert slow.calls == [(job_id, FileStatus.PROCESSING)] # применён один раз
|
||
assert (await _file_row(session_factory, file_id))["status"] is FileStatus.PROCESSING
|
||
assert len(_logged(logs, "дубликат", "DEBUG")) == 1
|
||
assert not _logged(logs, "неизвестного задания")
|
||
|
||
|
||
class _SlowNats:
|
||
"""Подставной NATS: публикация «висит» `delay` секунд, пока публикатор держит блокировку строки задания."""
|
||
|
||
def __init__(self, delay: float) -> None:
|
||
self.delay = delay
|
||
self.entered = asyncio.Event()
|
||
|
||
@property
|
||
def is_connected(self) -> bool:
|
||
return True
|
||
|
||
async def publish(self, subject: str, payload: bytes, msg_id: str) -> None:
|
||
self.entered.set()
|
||
await asyncio.sleep(self.delay)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_ack_waits_for_publisher_lock(session_factory, jobs: CommittedJobs):
|
||
"""Ack пришёл, пока публикатор держит строку (до его commit): обработчик ждёт блокировку и применяет ack после."""
|
||
job_id = await jobs.make_queued_job()
|
||
file_id = await _file_id(jobs, job_id)
|
||
nats = _SlowNats(LOCK_HOLD_SECONDS)
|
||
publisher = asyncio.create_task(
|
||
publish_pending(session_factory, nats, ack_timeout=timedelta(seconds=30), max_republish=3, batch_size=1)
|
||
)
|
||
await asyncio.wait_for(nats.entered.wait(), 5)
|
||
started = time.monotonic()
|
||
|
||
await _handle(session_factory, fake_msg(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1)))
|
||
|
||
waited = time.monotonic() - started
|
||
report = await publisher
|
||
job = await _row(session_factory, job_id)
|
||
assert report.published == 1
|
||
assert waited >= LOCK_HOLD_SECONDS / 2 # ждал блокировку, а не увидел queued
|
||
assert job["status"] is RecognitionStatus.processing
|
||
assert job["published_at"] is not None
|
||
assert job["started_at"] is not None
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_result_vs_watchdog_is_consistent(session_factory, jobs: CommittedJobs):
|
||
"""Результат и вахтёр на заданиях с прошедшим сроком одновременно: каждое — либо done (с s3_object, файл
|
||
`PROCESSING`), либо timeout (без результата, файл `ERROR`), но не смесь. Порядок решает блокировка строки задания:
|
||
кто первый, тот и финальный (и задание, и файл меняются в порядке «задание → файл», без взаимоблокировок)."""
|
||
past = datetime.now(UTC) - timedelta(minutes=1)
|
||
pairs = [
|
||
await jobs.make_file_job(
|
||
file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing, reply_deadline=past
|
||
)
|
||
for _ in range(RACE_JOBS)
|
||
]
|
||
job_ids = [job_id for _, job_id in pairs]
|
||
messages = [fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1)) for file_id, job_id in pairs]
|
||
|
||
await asyncio.gather(expire_deadlines(session_factory), *[_handle(session_factory, m) for m in messages])
|
||
|
||
for job_id in job_ids:
|
||
job = await _row(session_factory, job_id)
|
||
s3 = await _s3_rows(session_factory, _result_key(job["file_id"], job_id))
|
||
file = await _file_row(session_factory, job["file_id"])
|
||
if job["status"] is RecognitionStatus.done:
|
||
assert job["error_code"] is None
|
||
assert job["result_msg_id"] == f"result:{job_id}:1"
|
||
assert [r["id"] for r in s3] == [job["result_s3_object_id"]]
|
||
assert file["status"] is FileStatus.PROCESSING
|
||
else:
|
||
assert file["status"] is FileStatus.ERROR
|
||
assert job["status"] is RecognitionStatus.timeout
|
||
assert job["error_code"] == "TIMEOUT"
|
||
assert job["result_msg_id"] is None
|
||
assert job["result_s3_object_id"] is None
|
||
assert s3 == []
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_watchdog_waits_for_result_transaction(
|
||
session_factory, jobs: CommittedJobs, monkeypatch: pytest.MonkeyPatch
|
||
):
|
||
"""Вахтёр запущен, пока обработчик результата держит строку: его UPDATE ждёт commit, перепроверяет статус на новой
|
||
версии строки (`done`) и задание не трогает; файл остаётся `PROCESSING` (не `ERROR`)."""
|
||
past = datetime.now(UTC) - timedelta(minutes=1)
|
||
file_id, job_id = await jobs.make_file_job(
|
||
file_status=FileStatus.PROCESSING, job_status=RecognitionStatus.processing, reply_deadline=past
|
||
)
|
||
slow = SlowFileStatus(monkeypatch, delay=LOCK_HOLD_SECONDS)
|
||
msg = fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1))
|
||
handler = asyncio.create_task(_handle(session_factory, msg))
|
||
|
||
async def locked() -> None: # обработчик внутри транзакции, строка заблокирована
|
||
while not slow.calls:
|
||
await asyncio.sleep(0.01)
|
||
|
||
await asyncio.wait_for(locked(), 5)
|
||
|
||
expired = await expire_deadlines(session_factory)
|
||
await handler
|
||
|
||
assert expired == 0
|
||
job = await _row(session_factory, job_id)
|
||
assert job["status"] is RecognitionStatus.done
|
||
assert job["error_code"] is None
|
||
assert (await _file_row(session_factory, file_id))["status"] is FileStatus.PROCESSING
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_watchdog_after_result_and_result_after_watchdog(session_factory, jobs: CommittedJobs):
|
||
past = datetime.now(UTC) - timedelta(minutes=1)
|
||
first = await jobs.make_queued_job(status=RecognitionStatus.processing, reply_deadline=past)
|
||
second = await jobs.make_queued_job(status=RecognitionStatus.processing, reply_deadline=past)
|
||
|
||
await _handle(
|
||
session_factory, fake_msg(envelope(first, await _file_id(jobs, first), "done"), result_msg_id(first, 1))
|
||
)
|
||
await expire_deadlines(session_factory)
|
||
await _handle(
|
||
session_factory, fake_msg(envelope(second, await _file_id(jobs, second), "done"), result_msg_id(second, 1))
|
||
)
|
||
|
||
assert (await _row(session_factory, first))["status"] is RecognitionStatus.done
|
||
assert (await _row(session_factory, second))["status"] is RecognitionStatus.timeout
|
||
|
||
|
||
# --- e2e: реальный NATS -------------------------------------------------------------------------------------------
|
||
|
||
|
||
async def _wait_until(predicate: Any, timeout: float = 10.0) -> None:
|
||
deadline = time.monotonic() + timeout
|
||
while not await predicate():
|
||
if time.monotonic() > deadline:
|
||
raise AssertionError("условие не выполнено за отведённое время")
|
||
await asyncio.sleep(0.05)
|
||
|
||
|
||
async def _wait_status(
|
||
session_factory: async_sessionmaker[AsyncSession], job_id: uuid.UUID, status: RecognitionStatus
|
||
) -> dict[str, Any]:
|
||
async def reached() -> bool:
|
||
return (await _row(session_factory, job_id))["status"] is status
|
||
|
||
await _wait_until(reached)
|
||
return await _row(session_factory, job_id)
|
||
|
||
|
||
class _FakeService:
|
||
"""«Сервис распознавания» поверх голого nats-py: принимает запросы своего задания, публикует конверты."""
|
||
|
||
def __init__(self, job_id: uuid.UUID, result_subject: str) -> None:
|
||
self.job_id = job_id
|
||
self.result_subject = result_subject
|
||
self.requests: list[Msg] = []
|
||
self._nc = NATS()
|
||
|
||
async def start(self, url: str) -> None:
|
||
await self._nc.connect(url, connect_timeout=2, allow_reconnect=False)
|
||
|
||
async def collect(msg: Msg) -> None:
|
||
if json.loads(msg.data).get("job_id") == str(self.job_id):
|
||
self.requests.append(msg)
|
||
|
||
await self._nc.subscribe("files.tabular_uploaded", cb=collect)
|
||
await self._nc.flush()
|
||
|
||
async def send(self, body: dict[str, Any], msg_id: str) -> None:
|
||
MESSAGE_VALIDATOR.validate(body)
|
||
await self._nc.publish(self.result_subject, json.dumps(body).encode(), headers={NATS_MSG_ID_HEADER: msg_id})
|
||
await self._nc.flush()
|
||
|
||
async def stop(self) -> None:
|
||
await self._nc.close()
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_full_cycle_over_real_nats(session_factory, jobs: CommittedJobs, logs: list[Any]):
|
||
"""queued → published (outbox) → processing (ack) → done + s3_object (результат) → повтор без изменений;
|
||
файл: UPLOADED → PROCESSING (ack) → PROCESSING (результат, ждёт цикл маппинга)."""
|
||
settings = get_settings().OPENBAO.NATS
|
||
result_subject = f"test.result.{uuid.uuid4().hex}"
|
||
handler = partial(handle_message, session_factory=session_factory)
|
||
|
||
async def subscribe(conn: NatsConnection) -> None:
|
||
await conn.subscribe_results(handler)
|
||
|
||
core = NatsConnection(settings, on_connected=subscribe, result_subject=result_subject)
|
||
file_id, job_id = await jobs.make_file_job()
|
||
service = _FakeService(job_id, result_subject)
|
||
await core.start()
|
||
try:
|
||
if not await core.wait_ready(timeout=READY_TIMEOUT):
|
||
pytest.fail(f"NATS недоступен ({settings.NATS_URL}, .env.test)", pytrace=False)
|
||
await service.start(settings.NATS_URL)
|
||
|
||
async def published() -> bool:
|
||
await publish_pending(session_factory, core, ack_timeout=timedelta(seconds=30), max_republish=3)
|
||
return bool(service.requests)
|
||
|
||
await _wait_until(published)
|
||
await _wait_status(session_factory, job_id, RecognitionStatus.published)
|
||
assert service.requests[0].headers == {NATS_MSG_ID_HEADER: f"{job_id}:1"}
|
||
|
||
await service.send(envelope(job_id, file_id, "processing"), ack_msg_id(job_id, 1))
|
||
job = await _wait_status(session_factory, job_id, RecognitionStatus.processing)
|
||
assert job["started_at"] is not None
|
||
assert (await _file_row(session_factory, file_id))["status"] is FileStatus.PROCESSING
|
||
assert (job["service_name"], job["service_version"]) == ("tabular-parser", "0.1.0")
|
||
|
||
result = envelope(job_id, file_id, "done", metrics={"duration_ms": 10})
|
||
await service.send(result, result_msg_id(job_id, 1))
|
||
job = await _wait_status(session_factory, job_id, RecognitionStatus.done)
|
||
(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 job["finished_at"] is not None
|
||
assert s3["bucket_role"] is S3BucketRole.results
|
||
assert (await _file_row(session_factory, file_id))["status"] is FileStatus.PROCESSING
|
||
|
||
await service.send(result, result_msg_id(job_id, 1))
|
||
|
||
async def duplicate_seen() -> bool:
|
||
return bool(_logged(logs, "дубликат", "DEBUG"))
|
||
|
||
await _wait_until(duplicate_seen)
|
||
assert await _row(session_factory, job_id) == job
|
||
assert await _s3_rows(session_factory, _result_key(file_id, job_id)) == [s3]
|
||
finally:
|
||
await service.stop()
|
||
await core.stop()
|