Срез 0: скелет scan-gateway — без фронта и БД, горячие папки local/smb, воркер NATS без JetStream, эмулятор ИС Сканирование #3
@ -25,7 +25,7 @@ explicit_package_bases = true
|
||||
# Строгий режим для собственного кода сервиса (`src`). `strict = true` в overrides mypy (2.x) применяется глобально,
|
||||
# а не к модулю, поэтому флаги семейства strict перечислены явно.
|
||||
[[tool.mypy.overrides]]
|
||||
module = ["src.contract.*", "src.core.clients", "src.core.nats", "src.gateway.*", "src.normalizer.*", "src.scan_emulator.*"]
|
||||
module = ["src.contract.*", "src.core.clients", "src.core.health", "src.core.nats", "src.core.worker", "src.gateway.*", "src.normalizer.*", "src.scan_emulator.*"]
|
||||
disallow_any_generics = true
|
||||
disallow_subclassing_any = true
|
||||
disallow_untyped_calls = true
|
||||
|
||||
619
api/src/core/worker.py
Normal file
619
api/src/core/worker.py
Normal file
@ -0,0 +1,619 @@
|
||||
"""Воркер scan-gateway: запрос NATS → ack → S3 → горячая папка → ответ ИС → S3 `raw/` → нормализатор → конверт.
|
||||
|
||||
Spec image-recognition-gateway; design D-2, D-3, D-4, D-7; решения R-14..R-20.
|
||||
|
||||
- `on_request(msg)` — колбэк подписки NATS. Только разбирает и проверяет тело, регистрирует `job_id` и запускает
|
||||
задачу задания; ничего не ждёт (nats-py вызывает колбэки одной подписки последовательно, T11.4).
|
||||
Тело не по схеме / MAJOR ≠ 1 / `pipeline` ≠ `image` → `SCHEMA_MISMATCH` без ack; конверт публикуется, только если
|
||||
из тела извлекаются `job_id` и `file_id` (R-16), иначе запрос записывается в лог и отбрасывается.
|
||||
- Задача задания: (1) ack `processing` — до чтения S3 и до ожидания слота; (2) слот семафора `SCAN_MAX_INFLIGHT`;
|
||||
(3) `handle(request)` — обработка и публикация итогового конверта (R-20).
|
||||
- Повторный запрос с `job_id`, который в работе или в LRU обработанных (`DONE_CACHE_SIZE`), получает повторный ack с
|
||||
`attempt` повторного запроса; файл в ИС второй раз не отправляется (D-2, R-17).
|
||||
- Ошибки — результат, не молчание (таблица D-7): каждое задание заканчивается опубликованным конвертом; исключение
|
||||
задания наружу не выходит (кроме отмены при остановке); неудачная публикация — лог error (ядро закроет задание по
|
||||
`reply_deadline`).
|
||||
- Методы горячей папки синхронные (R-9) — вызываются через `asyncio.to_thread`; нормализатор — тоже в потоке.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import re
|
||||
from collections import OrderedDict
|
||||
from collections.abc import Awaitable, Callable, Coroutine
|
||||
from dataclasses import dataclass
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from typing import Any, Protocol
|
||||
|
||||
from loguru import logger
|
||||
from pydantic import ValidationError
|
||||
|
||||
from src.contract import (
|
||||
CONTRACT_VERSION,
|
||||
ErrorCode,
|
||||
Metrics,
|
||||
RecognitionRequest,
|
||||
RecognitionResult,
|
||||
RecognitionResultMessage,
|
||||
ResultStatus,
|
||||
S3Ref,
|
||||
ServiceInfo,
|
||||
build_ack_message,
|
||||
build_done_message,
|
||||
build_error_message,
|
||||
is_supported_major,
|
||||
)
|
||||
from src.core.clients import JSON_CONTENT_TYPE, OCTET_STREAM
|
||||
from src.core.config import Settings
|
||||
from src.core.nats import PublishFailed
|
||||
from src.gateway.hotfolder import HotFolder, inbox_name
|
||||
from src.normalizer import Normalizer
|
||||
|
||||
__all__ = [
|
||||
"MAX_ERROR_MESSAGE_CHARS",
|
||||
"IncomingMessage",
|
||||
"ObjectStorage",
|
||||
"ResultPublisher",
|
||||
"Worker",
|
||||
"raw_key",
|
||||
"result_key",
|
||||
"utc_now",
|
||||
]
|
||||
|
||||
# Предел текста `error.message` в конверте: конверт остаётся далеко меньше лимита NATS (1 МБ) даже при 4 байтах на
|
||||
# символ (T10.1). Длинное сообщение обрезается с многоточием.
|
||||
MAX_ERROR_MESSAGE_CHARS = 4000
|
||||
# Предел фрагментов `repr()` (исключения, недоверенные значения) внутри сообщений и логов
|
||||
MAX_FRAGMENT_CHARS = 300
|
||||
_ELLIPSIS = "…"
|
||||
_MAX_ATTEMPT = 2**31 - 1 # attempt из тела, отвергнутого схемой, берётся только из разумного диапазона
|
||||
_UUID_RE = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}", re.ASCII)
|
||||
_IMAGE_PIPELINE = "image"
|
||||
|
||||
SleepFn = Callable[[float], Awaitable[None]]
|
||||
ClockFn = Callable[[], datetime]
|
||||
|
||||
|
||||
class IncomingMessage(Protocol):
|
||||
"""Сообщение NATS в колбэке подписки: воркеру нужно только тело (`nats.aio.msg.Msg` подходит)."""
|
||||
|
||||
data: bytes
|
||||
|
||||
|
||||
class ObjectStorage(Protocol):
|
||||
"""Часть `S3Storage`, нужная воркеру; ошибки — классы `src.core.clients`."""
|
||||
|
||||
bucket: str
|
||||
|
||||
async def get_bytes(self, key: str, *, bucket: str | None = None) -> bytes: ...
|
||||
|
||||
async def put_bytes(self, key: str, data: bytes, content_type: str = OCTET_STREAM) -> None: ...
|
||||
|
||||
async def put_json(self, key: str, obj: Any) -> None: ...
|
||||
|
||||
|
||||
class ResultPublisher(Protocol):
|
||||
"""Часть `NatsGateway`: публикация конверта в `recognition.result` с `Nats-Msg-Id` и повторами."""
|
||||
|
||||
async def publish_result(self, message: RecognitionResultMessage, attempt: int) -> None: ...
|
||||
|
||||
|
||||
def utc_now() -> datetime:
|
||||
return datetime.now(UTC)
|
||||
|
||||
|
||||
def raw_key(job_id: object) -> str:
|
||||
"""Ключ сырого ответа ИС в S3 (D-4)."""
|
||||
return f"raw/{job_id}.json"
|
||||
|
||||
|
||||
def result_key(file_id: object, job_id: object) -> str:
|
||||
"""Ключ файла результата в S3 (контракт, «Схема ключей S3»)."""
|
||||
return f"results/{file_id}/{job_id}/result.json"
|
||||
|
||||
|
||||
def clean_text(text: str, limit: int = MAX_ERROR_MESSAGE_CHARS) -> str:
|
||||
"""Текст, безопасный для конверта: одиночные суррогаты (из `\\udXXX` в JSON) заменены на `?` — иначе публикация
|
||||
падает на кодировании в UTF-8 (T10.2); длина ограничена `limit` с многоточием в конце (T10.1)."""
|
||||
safe = text.encode("utf-8", "replace").decode("utf-8")
|
||||
if len(safe) > limit:
|
||||
safe = safe[: limit - len(_ELLIPSIS)] + _ELLIPSIS
|
||||
return safe
|
||||
|
||||
|
||||
def _fragment(value: object) -> str:
|
||||
"""Короткое безопасное представление недоверенного значения для сообщения или лога."""
|
||||
try:
|
||||
text = repr(value)
|
||||
except Exception: # repr не должен ронять обработку ошибки
|
||||
text = f"<{type(value).__name__}>"
|
||||
return clean_text(text, MAX_FRAGMENT_CHARS)
|
||||
|
||||
|
||||
def _describe(exc: BaseException) -> str:
|
||||
return f"{type(exc).__name__}: {clean_text(str(exc), MAX_FRAGMENT_CHARS)}"
|
||||
|
||||
|
||||
def _parse_json(data: bytes) -> tuple[object, str | None]:
|
||||
"""(разобранный JSON, None) или (None, описание ошибки разбора)."""
|
||||
try:
|
||||
return json.loads(data), None
|
||||
except (ValueError, RecursionError) as exc: # UnicodeDecodeError — подкласс ValueError
|
||||
return None, _describe(exc)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _Rejection:
|
||||
"""Тело запроса, не прошедшее проверку: причина и (если тело — JSON-объект) исходный словарь."""
|
||||
|
||||
reason: str
|
||||
raw: dict[str, Any] | None
|
||||
|
||||
|
||||
@dataclass
|
||||
class _JobContext:
|
||||
"""Состояние задания, нужное для конверта при непредвиденной ошибке: сырой ответ уже в S3."""
|
||||
|
||||
raw_s3_key: str | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _WaitResult:
|
||||
answer: bytes | None
|
||||
polls: int
|
||||
failures: int
|
||||
last_error: BaseException | None
|
||||
|
||||
|
||||
class Worker:
|
||||
"""Обработка запросов `recognition.image.request` (см. докстринг модуля).
|
||||
|
||||
Зависимости внедряются: `storage` (S3), `publisher` (NATS), `hotfolder`, `normalizer`; `clock` — часы UTC для
|
||||
дедлайна (R-20), `sleep` — пауза между опросами (подмены для тестов).
|
||||
Запросы принимаются между `start()` и `stop()`.
|
||||
"""
|
||||
|
||||
def __init__( # noqa: PLR0913 - зависимости воркера; всё, кроме настроек, — только именованные
|
||||
self,
|
||||
settings: Settings,
|
||||
*,
|
||||
storage: ObjectStorage,
|
||||
publisher: ResultPublisher,
|
||||
hotfolder: HotFolder,
|
||||
normalizer: Normalizer,
|
||||
clock: ClockFn = utc_now,
|
||||
sleep: SleepFn = asyncio.sleep,
|
||||
) -> None:
|
||||
self._settings = settings
|
||||
self._storage = storage
|
||||
self._publisher = publisher
|
||||
self._hotfolder = hotfolder
|
||||
self._normalizer = normalizer
|
||||
self._clock = clock
|
||||
self._sleep = sleep
|
||||
self.service = ServiceInfo.model_validate({"name": settings.SERVICE_NAME, "version": settings.VERSION})
|
||||
self._semaphore = asyncio.Semaphore(settings.SCAN_MAX_INFLIGHT)
|
||||
self._active: set[str] = set() # job_id в работе (ack, ожидание слота, обработка) — не вытесняются
|
||||
self._done: OrderedDict[str, None] = OrderedDict() # LRU обработанных job_id
|
||||
self._tasks: set[asyncio.Task[None]] = set()
|
||||
self._accepting = False
|
||||
|
||||
# --- жизненный цикл ---------------------------------------------------------------------------------------
|
||||
@property
|
||||
def active_jobs(self) -> int:
|
||||
"""Задания в работе (включая ожидающие слот)."""
|
||||
return len(self._active)
|
||||
|
||||
def start(self) -> None:
|
||||
"""Начать приём запросов (колбэк `on_request` подписывается после этого)."""
|
||||
self._accepting = True
|
||||
logger.info(
|
||||
"Воркер scan-gateway запущен. max_inflight={max_inflight}, poll={poll}, timeout={timeout}",
|
||||
max_inflight=self._settings.SCAN_MAX_INFLIGHT,
|
||||
poll=self._settings.SCAN_POLL_SECONDS,
|
||||
timeout=self._settings.SCAN_TIMEOUT_SECONDS,
|
||||
)
|
||||
|
||||
async def stop(self) -> None:
|
||||
"""Прекратить приём, отменить задания в работе и дождаться их завершения. Повторный вызов безопасен.
|
||||
|
||||
Отменённые задания конверта не публикуют: ядро закроет их по `reply_deadline` (D-2).
|
||||
"""
|
||||
self._accepting = False
|
||||
tasks = list(self._tasks)
|
||||
for task in tasks:
|
||||
task.cancel()
|
||||
if tasks:
|
||||
await asyncio.gather(*tasks, return_exceptions=True)
|
||||
logger.info("Воркер остановлен, отменено заданий: {count}", count=len(tasks))
|
||||
|
||||
async def wait_idle(self) -> None:
|
||||
"""Дождаться завершения всех запущенных задач (ack, задания, публикации отказов) — для тестов и остановки."""
|
||||
while self._tasks:
|
||||
await asyncio.gather(*list(self._tasks), return_exceptions=True)
|
||||
|
||||
# --- приём запроса ----------------------------------------------------------------------------------------
|
||||
async def on_request(self, msg: IncomingMessage) -> None:
|
||||
"""Колбэк подписки: проверка тела, дедупликация по `job_id`, запуск задачи. Не ждёт ни сети, ни ИС."""
|
||||
if not self._accepting:
|
||||
logger.warning("Воркер не принимает запросы (остановлен): сообщение отброшено, ядро повторит запрос")
|
||||
return
|
||||
parsed = self._parse_request(msg.data)
|
||||
if isinstance(parsed, _Rejection):
|
||||
self._reject(parsed)
|
||||
return
|
||||
request = parsed
|
||||
job_id = str(request.job_id)
|
||||
if self._seen(job_id):
|
||||
logger.info(
|
||||
"Повторный запрос: повторный ack без второй отправки в ИС. job_id={job_id}, attempt={attempt}",
|
||||
job_id=job_id,
|
||||
attempt=request.attempt,
|
||||
)
|
||||
self._spawn(self._publish(self._ack(request), request.attempt), f"ack-repeat:{job_id}")
|
||||
return
|
||||
self._active.add(job_id)
|
||||
logger.info(
|
||||
"Запрос принят. job_id={job_id}, file_id={file_id}, attempt={attempt}, file_type={file_type}",
|
||||
job_id=job_id,
|
||||
file_id=request.file_id,
|
||||
attempt=request.attempt,
|
||||
file_type=request.file_type,
|
||||
)
|
||||
self._spawn(self._run_job(request), f"job:{job_id}")
|
||||
|
||||
def _parse_request(self, data: bytes) -> RecognitionRequest | _Rejection:
|
||||
raw, error = _parse_json(data)
|
||||
if error is not None:
|
||||
return _Rejection(f"тело запроса не разбирается как JSON ({error})", None)
|
||||
if not isinstance(raw, dict):
|
||||
return _Rejection(f"тело запроса — не JSON-объект ({type(raw).__name__})", None)
|
||||
version = raw.get("schema_version")
|
||||
if not is_supported_major(version):
|
||||
return _Rejection(f"неподдерживаемая версия контракта {_fragment(version)}: ожидается MAJOR 1", raw)
|
||||
pipeline = raw.get("pipeline")
|
||||
if pipeline != _IMAGE_PIPELINE:
|
||||
return _Rejection(f"pipeline {_fragment(pipeline)} не обслуживается scan-gateway (ожидается image)", raw)
|
||||
try:
|
||||
return RecognitionRequest.model_validate(raw)
|
||||
except ValidationError as exc:
|
||||
details = "; ".join(
|
||||
f"{'.'.join(str(part) for part in item['loc']) or '<тело>'}: {item['msg']}"
|
||||
for item in exc.errors(include_url=False, include_input=False)[:5]
|
||||
)
|
||||
return _Rejection(f"запрос не соответствует схеме recognition-request: {details}", raw)
|
||||
|
||||
def _reject(self, rejection: _Rejection) -> None:
|
||||
"""`SCHEMA_MISMATCH` без ack (R-16): конверт — только если в теле есть корректные `job_id` и `file_id`."""
|
||||
raw = rejection.raw or {}
|
||||
job_id, file_id = raw.get("job_id"), raw.get("file_id")
|
||||
if not (_is_uuid(job_id) and _is_uuid(file_id)):
|
||||
logger.warning(
|
||||
"SCHEMA_MISMATCH: запрос отброшен без ответа (нет корректных job_id/file_id). reason={reason}",
|
||||
reason=clean_text(rejection.reason, MAX_FRAGMENT_CHARS),
|
||||
)
|
||||
return
|
||||
attempt = raw.get("attempt")
|
||||
if not (type(attempt) is int and 1 <= attempt <= _MAX_ATTEMPT):
|
||||
attempt = 1
|
||||
logger.warning(
|
||||
"SCHEMA_MISMATCH: запрос отвергнут, ack не публикуется. job_id={job_id}, reason={reason}",
|
||||
job_id=job_id,
|
||||
reason=clean_text(rejection.reason, MAX_FRAGMENT_CHARS),
|
||||
)
|
||||
message = build_error_message(
|
||||
str(job_id), str(file_id), self.service, ErrorCode.SCHEMA_MISMATCH, clean_text(rejection.reason)
|
||||
)
|
||||
self._spawn(self._publish(message, attempt), f"schema-mismatch:{job_id}")
|
||||
|
||||
# --- дедупликация (D-2, R-17) -----------------------------------------------------------------------------
|
||||
def _seen(self, job_id: str) -> bool:
|
||||
if job_id in self._active:
|
||||
return True
|
||||
if job_id in self._done:
|
||||
self._done.move_to_end(job_id)
|
||||
return True
|
||||
return False
|
||||
|
||||
def _finish(self, job_id: str) -> None:
|
||||
self._active.discard(job_id)
|
||||
self._done[job_id] = None
|
||||
self._done.move_to_end(job_id)
|
||||
while len(self._done) > self._settings.DONE_CACHE_SIZE:
|
||||
self._done.popitem(last=False)
|
||||
|
||||
# --- задачи -----------------------------------------------------------------------------------------------
|
||||
def _spawn(self, coro: Coroutine[Any, Any, None], name: str) -> None:
|
||||
task = asyncio.create_task(coro, name=name)
|
||||
self._tasks.add(task)
|
||||
task.add_done_callback(self._tasks.discard)
|
||||
|
||||
async def _run_job(self, request: RecognitionRequest) -> None:
|
||||
job_id = str(request.job_id)
|
||||
try:
|
||||
await self._publish(self._ack(request), request.attempt) # до чтения S3 и до ожидания слота
|
||||
async with self._semaphore:
|
||||
await self.handle(request)
|
||||
except asyncio.CancelledError:
|
||||
logger.warning("Задание отменено при остановке воркера. job_id={job_id}", job_id=job_id)
|
||||
raise
|
||||
except Exception: # handle не бросает; страховка, чтобы исключение не осталось в задаче
|
||||
logger.exception("Непредвиденная ошибка задания. job_id={job_id}", job_id=job_id)
|
||||
finally:
|
||||
self._finish(job_id)
|
||||
|
||||
def _ack(self, request: RecognitionRequest) -> RecognitionResultMessage:
|
||||
return build_ack_message(request.job_id, request.file_id, self.service)
|
||||
|
||||
async def _publish(self, message: RecognitionResultMessage, attempt: int) -> None:
|
||||
"""Публикация с повторами `NatsGateway`; неудача — лог error, исключение не выходит (D-7)."""
|
||||
try:
|
||||
await self._publisher.publish_result(message, attempt)
|
||||
except PublishFailed as exc:
|
||||
logger.error(
|
||||
"Конверт не опубликован, ядро закроет задание по reply_deadline. job_id={job_id}, status={status}, "
|
||||
"msg_id={msg_id}, attempts={attempts}, cause={cause}",
|
||||
job_id=message.job_id,
|
||||
status=message.status.value,
|
||||
msg_id=exc.msg_id,
|
||||
attempts=exc.attempts,
|
||||
cause=_fragment(exc.__cause__),
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Конверт не опубликован: непредвиденная ошибка. job_id={job_id}, status={status}, error={error}",
|
||||
job_id=message.job_id,
|
||||
status=message.status.value,
|
||||
error=_describe(exc),
|
||||
)
|
||||
|
||||
# --- обработка задания ------------------------------------------------------------------------------------
|
||||
async def handle(self, request: RecognitionRequest) -> RecognitionResultMessage:
|
||||
"""Обработать задание и опубликовать итоговый конверт; вернуть его.
|
||||
|
||||
Не бросает исключений (кроме `asyncio.CancelledError`): любой сбой становится конвертом `error` по D-7.
|
||||
Ack и семафор — забота вызывающего (`on_request`); здесь — S3 → горячая папка → ответ → `raw/` →
|
||||
нормализатор → `results/` → конверт.
|
||||
"""
|
||||
context = _JobContext()
|
||||
try:
|
||||
message = await self._process(request, context)
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.exception("Непредвиденная ошибка обработки. job_id={job_id}", job_id=request.job_id)
|
||||
message = self._error(
|
||||
request,
|
||||
ErrorCode.UNKNOWN_ERROR,
|
||||
f"непредвиденная ошибка scan-gateway: {_describe(exc)}",
|
||||
raw_s3_key=context.raw_s3_key,
|
||||
)
|
||||
logger.info(
|
||||
"Задание обработано. job_id={job_id}, status={status}, error_code={code}",
|
||||
job_id=request.job_id,
|
||||
status=message.status.value,
|
||||
code=message.error.code.value if message.error else None,
|
||||
)
|
||||
await self._publish(message, request.attempt)
|
||||
return message
|
||||
|
||||
def _error(
|
||||
self,
|
||||
request: RecognitionRequest,
|
||||
code: ErrorCode,
|
||||
text: str,
|
||||
*,
|
||||
raw_s3_key: str | None = None,
|
||||
metrics: Metrics | None = None,
|
||||
) -> RecognitionResultMessage:
|
||||
return build_error_message(
|
||||
request.job_id,
|
||||
request.file_id,
|
||||
self.service,
|
||||
code,
|
||||
clean_text(text),
|
||||
raw_s3_key=raw_s3_key,
|
||||
metrics=metrics,
|
||||
)
|
||||
|
||||
async def _process(self, request: RecognitionRequest, context: _JobContext) -> RecognitionResultMessage:
|
||||
"""Шаги задания; каждая неудача — конверт `error` по строке таблицы D-7."""
|
||||
sent = await self._send(request)
|
||||
if isinstance(sent, RecognitionResultMessage):
|
||||
return sent
|
||||
name, started = sent
|
||||
job_id = str(request.job_id)
|
||||
# 3. Опрос исходящего каталога до min(reply_deadline, старт + SCAN_TIMEOUT_SECONDS) (D-3, R-20)
|
||||
deadline = min(request.reply_deadline, started + timedelta(seconds=self._settings.SCAN_TIMEOUT_SECONDS))
|
||||
waited = await self._wait_answer(job_id, deadline)
|
||||
if waited.answer is None:
|
||||
if waited.failures == waited.polls and waited.last_error is not None:
|
||||
return self._error(
|
||||
request,
|
||||
ErrorCode.RECOGNITION_FAILED,
|
||||
f"горячая папка недоступна: исходящий каталог не прочитан ни разу за {waited.polls} опросов "
|
||||
f"({_describe(waited.last_error)})",
|
||||
)
|
||||
logger.warning(
|
||||
"ИС Сканирование не ответила до дедлайна. job_id={job_id}, deadline={deadline}, polls={polls}",
|
||||
job_id=job_id,
|
||||
deadline=deadline.isoformat(),
|
||||
polls=waited.polls,
|
||||
)
|
||||
return self._error(
|
||||
request,
|
||||
ErrorCode.TIMEOUT,
|
||||
f"ИС Сканирование не ответила до {deadline.isoformat()} (опросов: {waited.polls}); "
|
||||
f"входящий файл {name} остаётся в каталоге, повторной отправки нет",
|
||||
)
|
||||
return await self._accept_answer(request, waited.answer, context)
|
||||
|
||||
async def _send(self, request: RecognitionRequest) -> tuple[str, datetime] | RecognitionResultMessage: # noqa: PLR0911 - каждая ветка — строка таблицы D-7
|
||||
"""Срок, исходник из S3, запись во входящий каталог: (имя файла, момент записи) или конверт `error`."""
|
||||
job_id = str(request.job_id)
|
||||
if request.reply_deadline <= self._clock():
|
||||
# задание дождалось слота после срока ответа: ядро уже закрывает его, файл в ИС не отправляется
|
||||
return self._error(
|
||||
request,
|
||||
ErrorCode.TIMEOUT,
|
||||
f"срок ответа {request.reply_deadline.isoformat()} истёк до отправки файла в ИС Сканирование",
|
||||
)
|
||||
|
||||
# 1. Исходник из S3 (один бакет сервиса, R-19)
|
||||
source = request.s3
|
||||
if source.bucket != self._storage.bucket:
|
||||
return self._error(
|
||||
request,
|
||||
ErrorCode.FILE_READ_ERROR,
|
||||
f"бакет {_fragment(source.bucket)} не обслуживается: scan-gateway работает с одним бакетом "
|
||||
f"{_fragment(self._storage.bucket)}",
|
||||
)
|
||||
try:
|
||||
data = await self._storage.get_bytes(source.key, bucket=source.bucket)
|
||||
except Exception as exc:
|
||||
logger.warning(
|
||||
"Исходник не прочитан из S3. job_id={job_id}, error={error}", job_id=job_id, error=_describe(exc)
|
||||
)
|
||||
return self._error(
|
||||
request,
|
||||
ErrorCode.FILE_READ_ERROR,
|
||||
f"исходный файл {source.bucket}/{source.key} не прочитан из S3: {_describe(exc)}",
|
||||
)
|
||||
|
||||
# 2. Во входящий каталог как <job_id>_<original_name> (R-5, R-15)
|
||||
try:
|
||||
name = inbox_name(job_id, request.original_name)
|
||||
except ValueError:
|
||||
return self._error(
|
||||
request,
|
||||
ErrorCode.RECOGNITION_FAILED,
|
||||
f"недопустимое имя файла {_fragment(request.original_name)}: файл не передан в ИС Сканирование",
|
||||
)
|
||||
try:
|
||||
await asyncio.to_thread(self._hotfolder.put, name, data)
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Горячая папка недоступна: файл не записан. job_id={job_id}, error={error}",
|
||||
job_id=job_id,
|
||||
error=_describe(exc),
|
||||
)
|
||||
return self._error(
|
||||
request,
|
||||
ErrorCode.RECOGNITION_FAILED,
|
||||
f"горячая папка недоступна: файл не записан во входящий каталог ({_describe(exc)})",
|
||||
)
|
||||
return name, self._clock()
|
||||
|
||||
async def _accept_answer( # noqa: PLR0911 - каждая ветка — строка таблицы D-7
|
||||
self, request: RecognitionRequest, answer: bytes, context: _JobContext
|
||||
) -> RecognitionResultMessage:
|
||||
"""Ответ ИС: сырой в S3, нормализатор, файл результата; конверт итога."""
|
||||
job_id = str(request.job_id)
|
||||
|
||||
# 4. Сырой ответ в S3 как есть — до нормализации (D-4, R-18)
|
||||
parsed, parse_error = _parse_json(answer)
|
||||
key = raw_key(job_id)
|
||||
try:
|
||||
await self._storage.put_bytes(key, answer, JSON_CONTENT_TYPE if parse_error is None else OCTET_STREAM)
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Сырой ответ ИС не сохранён. job_id={job_id}, error={error}", job_id=job_id, error=_describe(exc)
|
||||
)
|
||||
return self._error(
|
||||
request, ErrorCode.UNKNOWN_ERROR, f"сырой ответ ИС не сохранён в S3 ({key}): {_describe(exc)}"
|
||||
)
|
||||
context.raw_s3_key = key
|
||||
if parse_error is not None:
|
||||
return self._error(
|
||||
request,
|
||||
ErrorCode.INVALID_FORMAT,
|
||||
f"ответ ИС Сканирование не разбирается как JSON: {parse_error}",
|
||||
raw_s3_key=key,
|
||||
)
|
||||
|
||||
# 5. Нормализатор (D-5); его исключение — UNKNOWN_ERROR с сохранённым сырым ответом
|
||||
try:
|
||||
outcome = await asyncio.to_thread(self._normalizer.normalize, parsed)
|
||||
except Exception as exc:
|
||||
logger.exception("Нормализатор ответа ИС упал. job_id={job_id}", job_id=job_id)
|
||||
return self._error(
|
||||
request, ErrorCode.UNKNOWN_ERROR, f"нормализатор ответа ИС упал: {_describe(exc)}", raw_s3_key=key
|
||||
)
|
||||
if outcome.status is ResultStatus.ERROR:
|
||||
error = outcome.error
|
||||
if error is None: # исключено инвариантом NormalizeOutcome
|
||||
raise ValueError("NormalizeOutcome со status=error без error")
|
||||
return self._error(request, error.code, error.message, raw_s3_key=key, metrics=outcome.metrics)
|
||||
|
||||
# 6. Успех: файл результата в S3, затем конверт с result_s3
|
||||
result = RecognitionResult.model_validate(
|
||||
{
|
||||
"schema_version": CONTRACT_VERSION,
|
||||
"job_id": request.job_id,
|
||||
"file_id": request.file_id,
|
||||
"service": self.service,
|
||||
"status": outcome.status,
|
||||
"raw_s3_key": key,
|
||||
"documents": list(outcome.documents),
|
||||
**({"metrics": outcome.metrics} if outcome.metrics is not None else {}),
|
||||
}
|
||||
)
|
||||
target = result_key(request.file_id, request.job_id)
|
||||
try:
|
||||
await self._storage.put_json(target, result.to_wire())
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Файл результата не записан. job_id={job_id}, error={error}", job_id=job_id, error=_describe(exc)
|
||||
)
|
||||
return self._error(
|
||||
request,
|
||||
ErrorCode.UNKNOWN_ERROR,
|
||||
f"файл результата не записан в S3 ({target}): {_describe(exc)}",
|
||||
raw_s3_key=key,
|
||||
)
|
||||
return build_done_message(
|
||||
request.job_id,
|
||||
request.file_id,
|
||||
self.service,
|
||||
S3Ref(bucket=self._storage.bucket, key=target),
|
||||
needs_check=outcome.status is ResultStatus.NEEDS_CHECK,
|
||||
raw_s3_key=key,
|
||||
metrics=outcome.metrics,
|
||||
)
|
||||
|
||||
async def _wait_answer(self, job_id: str, deadline: datetime) -> _WaitResult:
|
||||
"""Опрос `poll(job_id)` каждые `SCAN_POLL_SECONDS` до дедлайна (последний опрос — в момент дедлайна).
|
||||
|
||||
Исключения `poll` (SMB и прочие) — warning, опрос продолжается (R-14). Ответ не JSON может быть недописан
|
||||
(ИС пишет на шару без атомарного переименования): он перечитывается через период опроса и принимается, когда
|
||||
стал JSON или перестал меняться; на дедлайне берётся последний прочитанный.
|
||||
"""
|
||||
polls = failures = 0
|
||||
last_error: BaseException | None = None
|
||||
pending: bytes | None = None
|
||||
while True:
|
||||
polls += 1
|
||||
answer: bytes | None
|
||||
try:
|
||||
answer = await asyncio.to_thread(self._hotfolder.poll, job_id)
|
||||
except Exception as exc:
|
||||
failures += 1
|
||||
last_error = exc
|
||||
answer = None
|
||||
logger.warning(
|
||||
"Опрос исходящего каталога не удался, опрос продолжается. job_id={job_id}, error={error}",
|
||||
job_id=job_id,
|
||||
error=_describe(exc),
|
||||
)
|
||||
if answer is not None:
|
||||
if answer == pending or _parse_json(answer)[1] is None:
|
||||
return _WaitResult(answer, polls, failures, last_error)
|
||||
pending = answer
|
||||
now = self._clock()
|
||||
if now >= deadline:
|
||||
return _WaitResult(pending, polls, failures, last_error)
|
||||
await self._sleep(min(self._settings.SCAN_POLL_SECONDS, (deadline - now).total_seconds()))
|
||||
|
||||
|
||||
def _is_uuid(value: object) -> bool:
|
||||
return isinstance(value, str) and _UUID_RE.fullmatch(value) is not None
|
||||
@ -2,7 +2,11 @@
|
||||
каталога api/), поэтому `import src` (он создаёт приложение) работает без своей настройки окружения
|
||||
и без инфраструктуры."""
|
||||
|
||||
import sys
|
||||
from collections.abc import Iterator
|
||||
|
||||
import pytest
|
||||
from loguru import logger
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
@ -34,3 +38,17 @@ def minimal_env_file(tmp_path, monkeypatch):
|
||||
)
|
||||
monkeypatch.setenv("ENV_FILE", str(env_file))
|
||||
return env_file
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def log_lines() -> Iterator[list[str]]:
|
||||
"""Перехват loguru: строки `LEVEL|сообщение` вместо stderr (pytest запускается с --capture=no, вывод чистый).
|
||||
|
||||
Обработчик stderr по умолчанию восстанавливается после теста.
|
||||
"""
|
||||
lines: list[str] = []
|
||||
logger.remove()
|
||||
logger.add(lambda message: lines.append(str(message)), level="DEBUG", format="{level}|{message}")
|
||||
yield lines
|
||||
logger.remove()
|
||||
logger.add(sys.stderr)
|
||||
|
||||
899
api/tests/unit/test_worker.py
Normal file
899
api/tests/unit/test_worker.py
Normal file
@ -0,0 +1,899 @@
|
||||
"""Воркер шлюза (`src.core.worker`, задача 5.3): сценарии spec image-recognition-gateway, D-2, D-3, D-7, R-14..R-20.
|
||||
|
||||
Без Docker и сети: S3 — словарь в памяти (ошибки — классы `src.core.clients`), NATS — настоящий `NatsGateway` поверх
|
||||
фейкового клиента nats-py (тело, `Nats-Msg-Id`, повторы и `PublishFailed` — как в проде), горячая папка — настоящий
|
||||
`LocalHotFolder` на `tmp_path` под шпионом, ИС Сканирование — `process_once` эмулятора (задержка 0) в фоновой задаче.
|
||||
Общий журнал событий (`events`) упорядочивает публикации, чтения S3 и запись во входящий каталог.
|
||||
Каждый опубликованный конверт проверяется JSON Schema `recognition-result-message.schema.json`.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import shutil
|
||||
import uuid
|
||||
from collections.abc import AsyncIterator, Awaitable, Callable
|
||||
from contextlib import asynccontextmanager, suppress
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from jsonschema import Draft202012Validator, FormatChecker
|
||||
from nats import errors as nats_errors
|
||||
|
||||
from src.contract import (
|
||||
NATS_MSG_ID_HEADER,
|
||||
RESULT_SUBJECT,
|
||||
ErrorCode,
|
||||
MessageError,
|
||||
ResultStatus,
|
||||
StatementDocument,
|
||||
)
|
||||
from src.core.clients import JSON_CONTENT_TYPE, OCTET_STREAM, S3ObjectNotFoundError, S3WriteError
|
||||
from src.core.config import Settings, get_settings
|
||||
from src.core.nats import NatsGateway
|
||||
from src.core.worker import MAX_ERROR_MESSAGE_CHARS, Worker
|
||||
from src.gateway.hotfolder import LocalHotFolder
|
||||
from src.normalizer import NOT_IMPLEMENTED_MESSAGE, NormalizeOutcome, NullNormalizer
|
||||
from src.scan_emulator import process_once
|
||||
|
||||
API_DIR = Path(__file__).resolve().parents[2]
|
||||
CONTRACTS_DIR = API_DIR / "contracts"
|
||||
# `<корень scan-gateway>/../fixtures/ocr` (workspace и worktree `.wt/<срез>/`), как в тестах эмулятора
|
||||
REAL_FIXTURES = API_DIR.parent.parent / "fixtures" / "ocr"
|
||||
BUCKET = "arm-ka"
|
||||
PDF_BYTES = b"%PDF-1.4\n% scan-gateway test\n\x00\x01\x02binary"
|
||||
|
||||
MESSAGE_VALIDATOR = Draft202012Validator(
|
||||
json.loads((CONTRACTS_DIR / "recognition-result-message.schema.json").read_text(encoding="utf-8")),
|
||||
format_checker=FormatChecker(),
|
||||
)
|
||||
RESULT_VALIDATOR = Draft202012Validator(
|
||||
json.loads((CONTRACTS_DIR / "recognition-result.schema.json").read_text(encoding="utf-8")),
|
||||
format_checker=FormatChecker(),
|
||||
)
|
||||
STATEMENT_EXAMPLE = json.loads((CONTRACTS_DIR / "examples" / "result-statement-minimal.json").read_text("utf-8"))
|
||||
|
||||
# Синтетический успешный ответ ИС (R-10): форма `fixtures/ocr/aep_5.json`, `error = null`, непустой `data_ocr`
|
||||
SUCCESS_ANSWER: dict[str, Any] = {
|
||||
"file_uid": "ok_doc",
|
||||
"error": None,
|
||||
"data_ocr": {
|
||||
"label": {"form_name": "Бухгалтерский баланс", "confidence": 98.0},
|
||||
"organization": {"form_name": "ООО Ромашка", "confidence": 97.5},
|
||||
"inn": {"form_name": "7700000000", "confidence": 99.0},
|
||||
"units": {"form_name": "тыс. руб.", "confidence": 95.0},
|
||||
"date": {"form_name": "30.09.2025", "confidence": 96.0},
|
||||
"table_structure": [{"line_code": "1600", "line_name": "БАЛАНС", "values": ["5678", "-12.3"]}],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
# Логи воркера перехватываются (`log_lines` из conftest): вывод pytest чистый
|
||||
pytestmark = pytest.mark.usefixtures("log_lines")
|
||||
|
||||
|
||||
# --- фейки --------------------------------------------------------------------------------------------------------
|
||||
class FakeStorage:
|
||||
"""S3 в памяти с интерфейсом `S3Storage`: get_bytes / put_bytes / put_json и ошибками `src.core.clients`."""
|
||||
|
||||
def __init__(self, events: list[tuple[str, str]], bucket: str = BUCKET) -> None:
|
||||
self.events = events
|
||||
self.bucket = bucket
|
||||
self.objects: dict[tuple[str, str], bytes] = {}
|
||||
self.content_types: dict[tuple[str, str], str] = {}
|
||||
self.get_calls: list[tuple[str, str]] = []
|
||||
self.fail_put_prefixes: tuple[str, ...] = ()
|
||||
|
||||
async def get_bytes(self, key: str, *, bucket: str | None = None) -> bytes:
|
||||
bucket_name = bucket or self.bucket
|
||||
self.get_calls.append((bucket_name, key))
|
||||
self.events.append(("s3_get", key))
|
||||
try:
|
||||
return self.objects[(bucket_name, key)]
|
||||
except KeyError:
|
||||
raise S3ObjectNotFoundError(
|
||||
f"S3: объект {bucket_name}/{key} не прочитан (NoSuchKey)", bucket=bucket_name, key=key
|
||||
) from None
|
||||
|
||||
async def put_bytes(self, key: str, data: bytes, content_type: str = OCTET_STREAM) -> None:
|
||||
if self.fail_put_prefixes and key.startswith(self.fail_put_prefixes):
|
||||
raise S3WriteError(
|
||||
f"S3: объект {self.bucket}/{key} не записан (ConnectionError)", bucket=self.bucket, key=key
|
||||
)
|
||||
self.objects[(self.bucket, key)] = data
|
||||
self.content_types[(self.bucket, key)] = content_type
|
||||
self.events.append(("s3_put", key))
|
||||
|
||||
async def put_json(self, key: str, obj: Any) -> None:
|
||||
await self.put_bytes(key, json.dumps(obj, ensure_ascii=False).encode("utf-8"), JSON_CONTENT_TYPE)
|
||||
|
||||
|
||||
class FakeNatsClient:
|
||||
"""Подмена `nats.aio.client.Client` для настоящего `NatsGateway`: записывает публикации в общий журнал."""
|
||||
|
||||
def __init__(self, events: list[tuple[str, str]]) -> None:
|
||||
self.events = events
|
||||
self.is_connected = False
|
||||
self.is_closed = False
|
||||
self.published: list[tuple[str, bytes, dict[str, str]]] = []
|
||||
self.fail = False
|
||||
self.block: asyncio.Event | None = None
|
||||
|
||||
async def connect(self, *args: Any, **kwargs: Any) -> None:
|
||||
self.is_connected = True
|
||||
|
||||
async def publish(self, subject: str, payload: bytes = b"", headers: dict[str, str] | None = None) -> None:
|
||||
if self.block is not None:
|
||||
await self.block.wait()
|
||||
if self.fail:
|
||||
raise nats_errors.ConnectionClosedError
|
||||
assert headers is not None
|
||||
self.published.append((subject, payload, headers))
|
||||
self.events.append(("publish", headers[NATS_MSG_ID_HEADER]))
|
||||
|
||||
async def flush(self, timeout: float = 10) -> None:
|
||||
return None
|
||||
|
||||
async def drain(self) -> None:
|
||||
self.is_connected = False
|
||||
self.is_closed = True
|
||||
|
||||
async def close(self) -> None:
|
||||
self.is_connected = False
|
||||
self.is_closed = True
|
||||
|
||||
|
||||
class SpyHotFolder:
|
||||
"""`LocalHotFolder` под шпионом: журнал записей во входящий каталог, сценарии сбоев `poll`."""
|
||||
|
||||
def __init__(self, inner: LocalHotFolder, events: list[tuple[str, str]]) -> None:
|
||||
self.inner = inner
|
||||
self.events = events
|
||||
self.puts: list[str] = []
|
||||
self.polls = 0
|
||||
self.poll_errors: list[BaseException] = []
|
||||
self.poll_always: BaseException | None = None
|
||||
self.poll_script: list[bytes | None] = []
|
||||
|
||||
def put(self, name: str, data: bytes) -> None:
|
||||
self.inner.put(name, data)
|
||||
self.puts.append(name)
|
||||
self.events.append(("put", name))
|
||||
|
||||
def poll(self, job_id: str) -> bytes | None:
|
||||
self.polls += 1
|
||||
if self.poll_always is not None:
|
||||
raise self.poll_always
|
||||
if self.poll_errors:
|
||||
raise self.poll_errors.pop(0)
|
||||
if self.poll_script:
|
||||
return self.poll_script.pop(0)
|
||||
return self.inner.poll(job_id)
|
||||
|
||||
def healthy(self) -> bool:
|
||||
return self.inner.healthy()
|
||||
|
||||
|
||||
class RaisingNormalizer:
|
||||
def normalize(self, raw: object) -> NormalizeOutcome:
|
||||
raise RuntimeError("нормализатор сломан")
|
||||
|
||||
|
||||
class StubNormalizer:
|
||||
"""Возвращает заданный исход (done / needs_check / error) — в Срезе 0 успешный путь достижим только так."""
|
||||
|
||||
def __init__(self, outcome: NormalizeOutcome) -> None:
|
||||
self.outcome = outcome
|
||||
self.calls: list[object] = []
|
||||
|
||||
def normalize(self, raw: object) -> NormalizeOutcome:
|
||||
self.calls.append(raw)
|
||||
return self.outcome
|
||||
|
||||
|
||||
# --- окружение теста ----------------------------------------------------------------------------------------------
|
||||
@dataclass
|
||||
class Env:
|
||||
settings: Settings
|
||||
root: Path
|
||||
fixtures: Path
|
||||
events: list[tuple[str, str]]
|
||||
storage: FakeStorage
|
||||
client: FakeNatsClient
|
||||
gateway: NatsGateway
|
||||
hotfolder: SpyHotFolder
|
||||
worker: Worker
|
||||
local: LocalHotFolder = field(init=False)
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
self.local = self.hotfolder.inner
|
||||
|
||||
# запрос -------------------------------------------------------------------------------------------------------
|
||||
def request( # noqa: PLR0913 - параметры запроса для сценариев
|
||||
self,
|
||||
*,
|
||||
original_name: str = "ok_doc.pdf",
|
||||
deadline_s: float = 60,
|
||||
attempt: int = 1,
|
||||
bucket: str = BUCKET,
|
||||
data: bytes | None = PDF_BYTES,
|
||||
job_id: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
job = job_id or str(uuid.uuid4())
|
||||
file_id = str(uuid.uuid4())
|
||||
key = f"uploads/{file_id}/{job}/source.bin"
|
||||
if data is not None:
|
||||
self.storage.objects[(bucket, key)] = data
|
||||
deadline = datetime.now(UTC) + timedelta(seconds=deadline_s)
|
||||
return {
|
||||
"schema_version": "1.0",
|
||||
"job_id": job,
|
||||
"file_id": file_id,
|
||||
"pipeline": "image",
|
||||
"file_type": "PDF",
|
||||
"original_name": original_name,
|
||||
"s3": {"bucket": bucket, "key": key},
|
||||
"reply_deadline": deadline.isoformat().replace("+00:00", "Z"),
|
||||
"attempt": attempt,
|
||||
}
|
||||
|
||||
async def deliver(self, body: dict[str, Any] | bytes) -> None:
|
||||
data = body if isinstance(body, bytes) else json.dumps(body, ensure_ascii=False).encode("utf-8")
|
||||
await self.worker.on_request(FakeMsg(data))
|
||||
|
||||
async def run(self, body: dict[str, Any], timeout: float = 5) -> dict[str, Any]:
|
||||
"""Доставить запрос, дождаться окончания всех заданий, вернуть единственный итоговый конверт задания."""
|
||||
await self.deliver(body)
|
||||
await asyncio.wait_for(self.worker.wait_idle(), timeout)
|
||||
results = self.results(body["job_id"])
|
||||
assert len(results) == 1, results
|
||||
return results[0][1]
|
||||
|
||||
# публикации ---------------------------------------------------------------------------------------------------
|
||||
def published(self) -> list[tuple[str, dict[str, Any]]]:
|
||||
"""(Nats-Msg-Id, тело) всех публикаций; каждое тело проверено схемой конверта."""
|
||||
out = []
|
||||
for subject, payload, headers in self.client.published:
|
||||
assert subject == RESULT_SUBJECT
|
||||
body = json.loads(payload.decode("utf-8"))
|
||||
MESSAGE_VALIDATOR.validate(body)
|
||||
assert body["service"]["name"] == "scan-gateway"
|
||||
assert body["schema_version"] == "1.0"
|
||||
out.append((headers[NATS_MSG_ID_HEADER], body))
|
||||
return out
|
||||
|
||||
def acks(self, job_id: str) -> list[str]:
|
||||
return [mid for mid, body in self.published() if body["job_id"] == job_id and body["status"] == "processing"]
|
||||
|
||||
def results(self, job_id: str) -> list[tuple[str, dict[str, Any]]]:
|
||||
return [
|
||||
(mid, body) for mid, body in self.published() if body["job_id"] == job_id and body["status"] != "processing"
|
||||
]
|
||||
|
||||
def inbox(self) -> list[str]:
|
||||
if not self.local.input_dir.is_dir():
|
||||
return []
|
||||
return sorted(p.name for p in self.local.input_dir.iterdir() if not p.name.startswith("."))
|
||||
|
||||
def answer(self, job_id: str, content: bytes) -> None:
|
||||
path = self.local.answer_path(job_id)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(content)
|
||||
|
||||
|
||||
class FakeMsg:
|
||||
def __init__(self, data: bytes) -> None:
|
||||
self.data = data
|
||||
self.subject = "recognition.image.request"
|
||||
self.headers: dict[str, str] | None = None
|
||||
|
||||
|
||||
EnvFactory = Callable[..., Awaitable[Env]]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def make_env(tmp_path: Path) -> AsyncIterator[EnvFactory]:
|
||||
created: list[Env] = []
|
||||
|
||||
async def build(
|
||||
*,
|
||||
normalizer: Any = None,
|
||||
root_exists: bool = True,
|
||||
clock: Callable[[], datetime] | None = None,
|
||||
**overrides: Any,
|
||||
) -> Env:
|
||||
root = tmp_path / f"hotfolder-{len(created)}"
|
||||
if root_exists:
|
||||
root.mkdir()
|
||||
fixtures = tmp_path / "fixtures"
|
||||
fixtures.mkdir(exist_ok=True)
|
||||
(fixtures / "ok_doc.json").write_text(json.dumps(SUCCESS_ANSWER, ensure_ascii=False), encoding="utf-8")
|
||||
settings = get_settings().model_copy(
|
||||
update={
|
||||
"SCAN_HOTFOLDER_PATH": str(root),
|
||||
"SCAN_POLL_SECONDS": 0.01,
|
||||
"SCAN_TIMEOUT_SECONDS": 5,
|
||||
"PUBLISH_RETRIES": 1,
|
||||
**overrides,
|
||||
}
|
||||
)
|
||||
events: list[tuple[str, str]] = []
|
||||
storage = FakeStorage(events)
|
||||
client = FakeNatsClient(events)
|
||||
gateway = NatsGateway(settings, client_factory=lambda: client, retry_delay=0)
|
||||
await gateway.connect()
|
||||
hotfolder = SpyHotFolder(LocalHotFolder(root, settings.SCAN_INPUT_DIR, settings.SCAN_OUTPUT_DIR), events)
|
||||
kwargs: dict[str, Any] = {}
|
||||
if clock is not None:
|
||||
kwargs["clock"] = clock
|
||||
worker = Worker(
|
||||
settings,
|
||||
storage=storage,
|
||||
publisher=gateway,
|
||||
hotfolder=hotfolder,
|
||||
normalizer=normalizer or NullNormalizer(),
|
||||
**kwargs,
|
||||
)
|
||||
worker.start()
|
||||
env = Env(settings, root, fixtures, events, storage, client, gateway, hotfolder, worker)
|
||||
created.append(env)
|
||||
return env
|
||||
|
||||
yield build
|
||||
for env in created:
|
||||
await env.worker.stop()
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def emulator(env: Env, fixtures: Path | None = None) -> AsyncIterator[None]:
|
||||
"""ИС Сканирование: `process_once` (задержка 0) каждые 5 мс, пока открыт контекст."""
|
||||
|
||||
async def loop() -> None:
|
||||
while True:
|
||||
process_once(env.root, fixtures or env.fixtures, 0)
|
||||
await asyncio.sleep(0.005)
|
||||
|
||||
task = asyncio.create_task(loop())
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
task.cancel()
|
||||
with suppress(asyncio.CancelledError):
|
||||
await task
|
||||
|
||||
|
||||
async def eventually(predicate: Callable[[], bool], timeout: float = 3.0) -> None:
|
||||
deadline = asyncio.get_running_loop().time() + timeout
|
||||
while not predicate():
|
||||
if asyncio.get_running_loop().time() > deadline:
|
||||
raise AssertionError("условие не выполнилось вовремя")
|
||||
await asyncio.sleep(0.005)
|
||||
|
||||
|
||||
def assert_error(body: dict[str, Any], code: ErrorCode, *, raw: bool) -> None:
|
||||
assert body["status"] == "error"
|
||||
assert body["error"]["code"] == code.value
|
||||
assert body["error"]["message"]
|
||||
if raw:
|
||||
assert body["raw_s3_key"] == f"raw/{body['job_id']}.json"
|
||||
else:
|
||||
assert "raw_s3_key" not in body
|
||||
assert "result_s3" not in body
|
||||
|
||||
|
||||
# --- подтверждение запроса до обработки ---------------------------------------------------------------------------
|
||||
async def test_ack_is_published_before_s3_read_and_before_file_appears(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
request = env.request()
|
||||
async with emulator(env):
|
||||
await env.run(request)
|
||||
job_id = request["job_id"]
|
||||
name = f"{job_id}_ok_doc.pdf"
|
||||
kinds = [event for event in env.events if event[1] in (f"ack:{job_id}:1", request["s3"]["key"], name)]
|
||||
assert kinds == [("publish", f"ack:{job_id}:1"), ("s3_get", request["s3"]["key"]), ("put", name)]
|
||||
|
||||
|
||||
async def test_file_in_inbox_is_byte_for_byte_s3_object(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
request = env.request(original_name="aep_5.pdf")
|
||||
await env.deliver(request)
|
||||
path = env.local.input_dir / f"{request['job_id']}_aep_5.pdf"
|
||||
await eventually(path.exists)
|
||||
assert path.read_bytes() == PDF_BYTES
|
||||
await env.worker.stop()
|
||||
|
||||
|
||||
async def test_repeated_request_gets_second_ack_and_one_file(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
request = env.request(original_name="aep_5.pdf")
|
||||
await env.deliver(request)
|
||||
job_id = request["job_id"]
|
||||
await eventually(lambda: len(env.hotfolder.puts) == 1)
|
||||
await env.deliver({**request, "attempt": 2}) # ядро не дождалось ack и повторило запрос
|
||||
await eventually(lambda: len(env.acks(job_id)) == 2)
|
||||
await asyncio.sleep(0.05)
|
||||
assert env.acks(job_id) == [f"ack:{job_id}:1", f"ack:{job_id}:2"]
|
||||
assert env.inbox() == [f"{job_id}_aep_5.pdf"]
|
||||
assert env.hotfolder.puts == [f"{job_id}_aep_5.pdf"]
|
||||
assert len(env.storage.get_calls) == 1
|
||||
|
||||
|
||||
async def test_repeated_request_after_result_gets_ack_without_reprocessing(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
request = env.request()
|
||||
async with emulator(env):
|
||||
await env.run(request)
|
||||
await env.deliver(request)
|
||||
await asyncio.wait_for(env.worker.wait_idle(), 5)
|
||||
job_id = request["job_id"]
|
||||
assert env.acks(job_id) == [f"ack:{job_id}:1", f"ack:{job_id}:1"]
|
||||
assert len(env.results(job_id)) == 1
|
||||
assert env.hotfolder.puts == [f"{job_id}_ok_doc.pdf"]
|
||||
|
||||
|
||||
async def test_done_cache_is_bounded_lru(make_env: EnvFactory) -> None:
|
||||
env = await make_env(DONE_CACHE_SIZE=2)
|
||||
requests = [env.request() for _ in range(3)]
|
||||
async with emulator(env):
|
||||
for request in requests:
|
||||
await env.run(request)
|
||||
# первый job_id вытеснен из LRU размера 2 — его повтор обрабатывается заново (в проде — 1000 заданий назад)
|
||||
await env.deliver(requests[0])
|
||||
await asyncio.wait_for(env.worker.wait_idle(), 5)
|
||||
# третий ещё в кэше — повтор только с ack
|
||||
await env.deliver(requests[2])
|
||||
await asyncio.wait_for(env.worker.wait_idle(), 5)
|
||||
assert len(env.results(requests[0]["job_id"])) == 2
|
||||
assert len(env.results(requests[2]["job_id"])) == 1
|
||||
assert len(env.acks(requests[2]["job_id"])) == 2
|
||||
|
||||
|
||||
async def test_on_request_returns_without_waiting_for_publication(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
env.client.block = asyncio.Event()
|
||||
request = env.request()
|
||||
await asyncio.wait_for(env.deliver(request), 0.2) # публикация ack висит, а колбэк уже вернулся (T11.4)
|
||||
assert env.client.published == []
|
||||
env.client.block.set()
|
||||
async with emulator(env):
|
||||
await asyncio.wait_for(env.worker.wait_idle(), 5)
|
||||
assert len(env.results(request["job_id"])) == 1
|
||||
|
||||
|
||||
# --- результат по контракту ---------------------------------------------------------------------------------------
|
||||
async def test_success_answer_gives_not_implemented_with_raw_key(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
request = env.request(attempt=3)
|
||||
async with emulator(env):
|
||||
body = await env.run(request)
|
||||
job_id = request["job_id"]
|
||||
assert_error(body, ErrorCode.NOT_IMPLEMENTED, raw=True)
|
||||
assert body["error"]["message"] == NOT_IMPLEMENTED_MESSAGE
|
||||
assert body["file_id"] == request["file_id"]
|
||||
assert [mid for mid, _ in env.published()] == [f"ack:{job_id}:3", f"result:{job_id}:3"]
|
||||
raw = env.storage.objects[(BUCKET, f"raw/{job_id}.json")]
|
||||
assert raw == env.local.answer_path(job_id).read_bytes() # как есть, байт в байт
|
||||
assert json.loads(raw) == {**SUCCESS_ANSWER, "file_uid": job_id}
|
||||
assert env.storage.content_types[(BUCKET, f"raw/{job_id}.json")] == JSON_CONTENT_TYPE
|
||||
# сырой ответ сохранён раньше публикации результата
|
||||
assert env.events.index(("s3_put", f"raw/{job_id}.json")) < env.events.index(("publish", f"result:{job_id}:3"))
|
||||
assert env.inbox() == [f"{job_id}_ok_doc.pdf.done"]
|
||||
|
||||
|
||||
@pytest.mark.fixtures
|
||||
async def test_real_fixture_error_is_passed_through(make_env: EnvFactory, tmp_path: Path) -> None:
|
||||
if not REAL_FIXTURES.is_dir():
|
||||
pytest.skip("fixtures недоступен: положите репозиторий fixtures рядом с workspace")
|
||||
real = tmp_path / "real-fixtures"
|
||||
real.mkdir()
|
||||
for source in REAL_FIXTURES.glob("*.json"):
|
||||
shutil.copy(source, real / source.name)
|
||||
expected = json.loads((real / "aep_5.json").read_text(encoding="utf-8"))["error"]
|
||||
env = await make_env()
|
||||
request = env.request(original_name="aep_5.PDF")
|
||||
async with emulator(env, real):
|
||||
body = await env.run(request)
|
||||
assert_error(body, ErrorCode.INVALID_DOCUMENT, raw=True)
|
||||
assert body["error"]["message"] == expected["message"]
|
||||
assert (BUCKET, f"raw/{request['job_id']}.json") in env.storage.objects
|
||||
|
||||
|
||||
async def test_unknown_name_gets_emulator_invalid_document(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
request = env.request(original_name="unknown.pdf")
|
||||
async with emulator(env):
|
||||
body = await env.run(request)
|
||||
assert_error(body, ErrorCode.INVALID_DOCUMENT, raw=True)
|
||||
assert "unknown.pdf" in body["error"]["message"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("status", [ResultStatus.DONE, ResultStatus.NEEDS_CHECK])
|
||||
async def test_successful_outcome_writes_result_file(make_env: EnvFactory, status: ResultStatus) -> None:
|
||||
document = StatementDocument.model_validate(STATEMENT_EXAMPLE["documents"][0])
|
||||
normalizer = StubNormalizer(NormalizeOutcome(status=status, documents=[document]))
|
||||
env = await make_env(normalizer=normalizer)
|
||||
request = env.request()
|
||||
async with emulator(env):
|
||||
body = await env.run(request)
|
||||
job_id, file_id = request["job_id"], request["file_id"]
|
||||
key = f"results/{file_id}/{job_id}/result.json"
|
||||
assert body["status"] == status.value
|
||||
assert body["result_s3"] == {"bucket": BUCKET, "key": key}
|
||||
assert body["result_schema_version"] == "1.0"
|
||||
assert body["raw_s3_key"] == f"raw/{job_id}.json"
|
||||
assert "error" not in body
|
||||
result = json.loads(env.storage.objects[(BUCKET, key)].decode("utf-8"))
|
||||
RESULT_VALIDATOR.validate(result)
|
||||
assert result["status"] == status.value
|
||||
assert result["job_id"] == job_id
|
||||
assert result["file_id"] == file_id
|
||||
assert result["service"]["name"] == "scan-gateway"
|
||||
assert result["raw_s3_key"] == f"raw/{job_id}.json"
|
||||
assert result["documents"] == STATEMENT_EXAMPLE["documents"]
|
||||
assert env.storage.content_types[(BUCKET, key)] == JSON_CONTENT_TYPE
|
||||
assert normalizer.calls == [{**SUCCESS_ANSWER, "file_uid": job_id}]
|
||||
assert env.events.index(("s3_put", key)) < env.events.index(("publish", f"result:{job_id}:1"))
|
||||
|
||||
|
||||
async def test_result_file_write_failure_is_unknown_error_with_raw_key(make_env: EnvFactory) -> None:
|
||||
document = StatementDocument.model_validate(STATEMENT_EXAMPLE["documents"][0])
|
||||
env = await make_env(normalizer=StubNormalizer(NormalizeOutcome(status=ResultStatus.DONE, documents=[document])))
|
||||
env.storage.fail_put_prefixes = ("results/",)
|
||||
request = env.request()
|
||||
async with emulator(env):
|
||||
body = await env.run(request)
|
||||
assert_error(body, ErrorCode.UNKNOWN_ERROR, raw=True)
|
||||
assert "результат" in body["error"]["message"]
|
||||
|
||||
|
||||
# --- сырой ответ и нормализатор -----------------------------------------------------------------------------------
|
||||
async def test_normalizer_exception_is_unknown_error_with_raw_saved(make_env: EnvFactory) -> None:
|
||||
env = await make_env(normalizer=RaisingNormalizer())
|
||||
request = env.request()
|
||||
async with emulator(env):
|
||||
body = await env.run(request)
|
||||
assert_error(body, ErrorCode.UNKNOWN_ERROR, raw=True)
|
||||
assert "нормализатор" in body["error"]["message"]
|
||||
assert (BUCKET, f"raw/{request['job_id']}.json") in env.storage.objects
|
||||
|
||||
|
||||
async def test_raw_write_failure_is_unknown_error_without_raw_key(make_env: EnvFactory) -> None:
|
||||
normalizer = StubNormalizer(
|
||||
NormalizeOutcome(status=ResultStatus.ERROR, error=MessageError(code="UNREADABLE", message="x"))
|
||||
)
|
||||
env = await make_env(normalizer=normalizer)
|
||||
env.storage.fail_put_prefixes = ("raw/",)
|
||||
request = env.request()
|
||||
async with emulator(env):
|
||||
body = await env.run(request)
|
||||
assert_error(body, ErrorCode.UNKNOWN_ERROR, raw=False)
|
||||
assert "сырой ответ" in body["error"]["message"]
|
||||
assert normalizer.calls == [] # без сохранённого сырого ответа нормализатор не вызывается (D-4)
|
||||
|
||||
|
||||
async def test_answer_not_json_is_invalid_format_and_raw_stored_as_is(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
request = env.request()
|
||||
job_id = request["job_id"]
|
||||
garbage = b"\xff\xfe<html>not json"
|
||||
await env.deliver(request)
|
||||
await eventually(lambda: len(env.hotfolder.puts) == 1)
|
||||
env.answer(job_id, garbage)
|
||||
await asyncio.wait_for(env.worker.wait_idle(), 5)
|
||||
[(_, body)] = env.results(job_id)
|
||||
assert_error(body, ErrorCode.INVALID_FORMAT, raw=True)
|
||||
assert env.storage.objects[(BUCKET, f"raw/{job_id}.json")] == garbage
|
||||
assert env.storage.content_types[(BUCKET, f"raw/{job_id}.json")] == OCTET_STREAM
|
||||
|
||||
|
||||
async def test_partially_written_answer_is_reread(make_env: EnvFactory) -> None:
|
||||
"""ИС пишет ответ не атомарно (SMB): недописанный JSON перечитывается через период опроса."""
|
||||
env = await make_env()
|
||||
full = json.dumps({**SUCCESS_ANSWER, "file_uid": "x"}).encode()
|
||||
env.hotfolder.poll_script = [None, full[:10], full]
|
||||
body = await env.run(env.request())
|
||||
assert_error(body, ErrorCode.NOT_IMPLEMENTED, raw=True)
|
||||
assert env.storage.objects[(BUCKET, f"raw/{body['job_id']}.json")] == full
|
||||
|
||||
|
||||
# --- таймаут ------------------------------------------------------------------------------------------------------
|
||||
async def test_timeout_by_scan_timeout_keeps_file_and_never_resends(make_env: EnvFactory) -> None:
|
||||
env = await make_env(SCAN_TIMEOUT_SECONDS=0.2)
|
||||
request = env.request(original_name="aep_5.pdf", deadline_s=60)
|
||||
started = asyncio.get_running_loop().time()
|
||||
body = await env.run(request)
|
||||
elapsed = asyncio.get_running_loop().time() - started
|
||||
assert_error(body, ErrorCode.TIMEOUT, raw=False)
|
||||
assert 0.15 < elapsed < 2
|
||||
assert env.inbox() == [f"{request['job_id']}_aep_5.pdf"] # входящий файл остаётся
|
||||
assert env.hotfolder.puts == [f"{request['job_id']}_aep_5.pdf"] # повторной отправки нет
|
||||
assert env.hotfolder.polls >= 5
|
||||
|
||||
|
||||
async def test_timeout_by_reply_deadline(make_env: EnvFactory) -> None:
|
||||
env = await make_env(SCAN_TIMEOUT_SECONDS=300)
|
||||
request = env.request(deadline_s=0.2)
|
||||
started = asyncio.get_running_loop().time()
|
||||
body = await env.run(request)
|
||||
assert_error(body, ErrorCode.TIMEOUT, raw=False)
|
||||
assert asyncio.get_running_loop().time() - started < 2
|
||||
assert len(env.hotfolder.puts) == 1
|
||||
|
||||
|
||||
async def test_deadline_passed_before_sending_gives_timeout_without_file(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
request = env.request(deadline_s=-1)
|
||||
body = await env.run(request)
|
||||
assert_error(body, ErrorCode.TIMEOUT, raw=False)
|
||||
assert env.hotfolder.puts == []
|
||||
assert env.storage.get_calls == []
|
||||
assert env.acks(request["job_id"]) == [f"ack:{request['job_id']}:1"]
|
||||
|
||||
|
||||
async def test_deadline_uses_injected_clock(make_env: EnvFactory) -> None:
|
||||
"""Дедлайн считается по инжектируемым часам (R-20): часы «убежали» на час — таймаут на первом же опросе."""
|
||||
shift = {"value": timedelta(0)}
|
||||
|
||||
def clock() -> datetime:
|
||||
return datetime.now(UTC) + shift["value"]
|
||||
|
||||
env = await make_env(clock=clock, SCAN_TIMEOUT_SECONDS=60)
|
||||
request = env.request(deadline_s=120)
|
||||
await env.deliver(request)
|
||||
await eventually(lambda: len(env.hotfolder.puts) == 1)
|
||||
shift["value"] = timedelta(hours=1)
|
||||
await asyncio.wait_for(env.worker.wait_idle(), 2)
|
||||
[(_, body)] = env.results(request["job_id"])
|
||||
assert_error(body, ErrorCode.TIMEOUT, raw=False)
|
||||
|
||||
|
||||
# --- ошибки транспорта --------------------------------------------------------------------------------------------
|
||||
async def test_missing_hotfolder_root_is_recognition_failed(make_env: EnvFactory) -> None:
|
||||
env = await make_env(root_exists=False)
|
||||
request = env.request()
|
||||
body = await env.run(request)
|
||||
assert_error(body, ErrorCode.RECOGNITION_FAILED, raw=False)
|
||||
assert "горячая папка недоступна" in body["error"]["message"]
|
||||
assert not env.root.exists() # корень не создаётся (R-5)
|
||||
assert env.hotfolder.healthy() is False
|
||||
|
||||
|
||||
async def test_unusable_original_name_is_recognition_failed(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
body = await env.run(env.request(original_name=".."))
|
||||
assert_error(body, ErrorCode.RECOGNITION_FAILED, raw=False)
|
||||
assert "недопустимое имя файла" in body["error"]["message"]
|
||||
assert env.hotfolder.puts == []
|
||||
|
||||
|
||||
async def test_transient_poll_errors_do_not_fail_the_job(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
env.hotfolder.poll_errors = [OSError(5, "STATUS_SHARING_VIOLATION"), ConnectionResetError("smb")]
|
||||
async with emulator(env):
|
||||
body = await env.run(env.request())
|
||||
assert_error(body, ErrorCode.NOT_IMPLEMENTED, raw=True)
|
||||
|
||||
|
||||
async def test_poll_failing_until_deadline_is_recognition_failed(make_env: EnvFactory) -> None:
|
||||
env = await make_env(SCAN_TIMEOUT_SECONDS=0.1)
|
||||
env.hotfolder.poll_always = OSError(13, "STATUS_ACCESS_DENIED")
|
||||
body = await env.run(env.request())
|
||||
assert_error(body, ErrorCode.RECOGNITION_FAILED, raw=False)
|
||||
assert "горячая папка недоступна" in body["error"]["message"]
|
||||
|
||||
|
||||
async def test_poll_failing_sometimes_then_silence_is_timeout(make_env: EnvFactory) -> None:
|
||||
env = await make_env(SCAN_TIMEOUT_SECONDS=0.1)
|
||||
env.hotfolder.poll_errors = [OSError(13, "STATUS_ACCESS_DENIED")]
|
||||
body = await env.run(env.request())
|
||||
assert_error(body, ErrorCode.TIMEOUT, raw=False)
|
||||
|
||||
|
||||
async def test_missing_s3_object_is_file_read_error(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
request = env.request(data=None)
|
||||
body = await env.run(request)
|
||||
assert_error(body, ErrorCode.FILE_READ_ERROR, raw=False)
|
||||
assert request["s3"]["key"] in body["error"]["message"]
|
||||
assert env.hotfolder.puts == []
|
||||
assert env.acks(request["job_id"]) == [f"ack:{request['job_id']}:1"]
|
||||
|
||||
|
||||
async def test_foreign_bucket_is_file_read_error_without_reading(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
request = env.request(bucket="other-bucket")
|
||||
body = await env.run(request)
|
||||
assert_error(body, ErrorCode.FILE_READ_ERROR, raw=False)
|
||||
assert "other-bucket" in body["error"]["message"]
|
||||
assert env.storage.get_calls == []
|
||||
assert env.acks(request["job_id"]) == [f"ack:{request['job_id']}:1"]
|
||||
|
||||
|
||||
async def test_exception_outside_steps_is_unknown_error(make_env: EnvFactory) -> None:
|
||||
class BadOutcomeNormalizer:
|
||||
def normalize(self, raw: object) -> NormalizeOutcome:
|
||||
return "не исход" # type: ignore[return-value]
|
||||
|
||||
env = await make_env(normalizer=BadOutcomeNormalizer())
|
||||
async with emulator(env):
|
||||
body = await env.run(env.request())
|
||||
assert_error(body, ErrorCode.UNKNOWN_ERROR, raw=True)
|
||||
|
||||
|
||||
# --- SCHEMA_MISMATCH без ack (R-16) -------------------------------------------------------------------------------
|
||||
def _mutations() -> list[tuple[str, Callable[[dict[str, Any]], dict[str, Any]]]]:
|
||||
return [
|
||||
("major_2", lambda r: {**r, "schema_version": "2.0"}),
|
||||
("pipeline_tabular", lambda r: {**r, "pipeline": "tabular", "file_type": "XLSX"}),
|
||||
("no_s3", lambda r: {k: v for k, v in r.items() if k != "s3"}),
|
||||
("bad_file_type", lambda r: {**r, "file_type": "DOCX"}),
|
||||
("naive_deadline", lambda r: {**r, "reply_deadline": "2026-10-03T12:45:00"}),
|
||||
("attempt_zero", lambda r: {**r, "attempt": 0}),
|
||||
("empty_name", lambda r: {**r, "original_name": ""}),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("case", "mutate"), _mutations(), ids=[case for case, _ in _mutations()])
|
||||
async def test_invalid_request_is_schema_mismatch_without_ack(
|
||||
make_env: EnvFactory, case: str, mutate: Callable[[dict[str, Any]], dict[str, Any]]
|
||||
) -> None:
|
||||
env = await make_env()
|
||||
request = env.request()
|
||||
body = mutate(request)
|
||||
await env.deliver(body)
|
||||
await asyncio.wait_for(env.worker.wait_idle(), 5)
|
||||
job_id = request["job_id"]
|
||||
published = env.published()
|
||||
assert env.acks(job_id) == []
|
||||
attempt = body["attempt"] if body["attempt"] >= 1 else 1
|
||||
assert [mid for mid, _ in published] == [f"result:{job_id}:{attempt}"]
|
||||
assert_error(published[0][1], ErrorCode.SCHEMA_MISMATCH, raw=False)
|
||||
assert published[0][1]["file_id"] == request["file_id"]
|
||||
assert env.storage.get_calls == []
|
||||
assert env.hotfolder.puts == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"data",
|
||||
[
|
||||
b"not json at all",
|
||||
b"\xff\xfe\x00broken utf-8",
|
||||
b"[1, 2, 3]",
|
||||
b'"string"',
|
||||
b"[" * 100_000, # глубокая вложенность: RecursionError парсера не роняет воркер
|
||||
json.dumps({"schema_version": "1.0", "job_id": "not-a-uuid", "file_id": str(uuid.uuid4())}).encode(),
|
||||
json.dumps({"schema_version": "1.0", "job_id": str(uuid.uuid4())}).encode(),
|
||||
json.dumps({"job_id": 5, "file_id": None}).encode(),
|
||||
],
|
||||
ids=["text", "bad_utf8", "array", "string", "deep", "bad_job_id", "no_file_id", "wrong_types"],
|
||||
)
|
||||
async def test_unparsable_request_without_ids_is_dropped(
|
||||
make_env: EnvFactory, data: bytes, log_lines: list[str]
|
||||
) -> None:
|
||||
env = await make_env()
|
||||
await env.deliver(data)
|
||||
await asyncio.wait_for(env.worker.wait_idle(), 5)
|
||||
assert env.client.published == []
|
||||
assert env.hotfolder.puts == []
|
||||
assert any("SCHEMA_MISMATCH" in line for line in log_lines)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("attempt", "expected"), [(3, 3), (True, 1), ("2", 1), (-1, 1), (2**40, 1)])
|
||||
async def test_schema_mismatch_attempt_from_body(make_env: EnvFactory, attempt: Any, expected: int) -> None:
|
||||
env = await make_env()
|
||||
request = env.request()
|
||||
await env.deliver({**request, "schema_version": "2.0", "attempt": attempt})
|
||||
await asyncio.wait_for(env.worker.wait_idle(), 5)
|
||||
assert [mid for mid, _ in env.published()] == [f"result:{request['job_id']}:{expected}"]
|
||||
|
||||
|
||||
async def test_schema_mismatch_does_not_register_job_id(make_env: EnvFactory) -> None:
|
||||
"""Исправленный запрос с тем же job_id после SCHEMA_MISMATCH обрабатывается (это не повтор)."""
|
||||
env = await make_env()
|
||||
request = env.request()
|
||||
await env.deliver({**request, "schema_version": "2.0"})
|
||||
async with emulator(env):
|
||||
await env.deliver(request)
|
||||
await asyncio.wait_for(env.worker.wait_idle(), 5)
|
||||
assert len(env.hotfolder.puts) == 1
|
||||
|
||||
|
||||
# --- публикация, размер и кодировка -------------------------------------------------------------------------------
|
||||
async def test_publish_failure_is_logged_and_worker_keeps_working(make_env: EnvFactory, log_lines: list[str]) -> None:
|
||||
env = await make_env()
|
||||
env.client.fail = True
|
||||
first = env.request()
|
||||
async with emulator(env):
|
||||
await env.deliver(first)
|
||||
await asyncio.wait_for(env.worker.wait_idle(), 5)
|
||||
assert env.client.published == []
|
||||
assert any(line.startswith("ERROR|") and first["job_id"] in line for line in log_lines)
|
||||
env.client.fail = False
|
||||
second = env.request()
|
||||
body = await env.run(second)
|
||||
assert_error(body, ErrorCode.NOT_IMPLEMENTED, raw=True)
|
||||
assert len(env.hotfolder.puts) == 2
|
||||
|
||||
|
||||
async def test_huge_error_message_is_truncated(make_env: EnvFactory) -> None:
|
||||
huge = "я" * 2_000_000
|
||||
normalizer = StubNormalizer(
|
||||
NormalizeOutcome(status=ResultStatus.ERROR, error=MessageError(code="INVALID_DOCUMENT", message=huge))
|
||||
)
|
||||
env = await make_env(normalizer=normalizer)
|
||||
async with emulator(env):
|
||||
body = await env.run(env.request())
|
||||
assert_error(body, ErrorCode.INVALID_DOCUMENT, raw=True)
|
||||
message = body["error"]["message"]
|
||||
assert len(message) == MAX_ERROR_MESSAGE_CHARS
|
||||
assert message.endswith("…")
|
||||
assert all(len(payload) < 900_000 for _, payload, _ in env.client.published)
|
||||
|
||||
|
||||
async def test_lone_surrogate_in_is_error_message_does_not_break_publishing(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
request = env.request()
|
||||
job_id = request["job_id"]
|
||||
await env.deliver(request)
|
||||
await eventually(lambda: len(env.hotfolder.puts) == 1)
|
||||
env.answer(job_id, b'{"file_uid": "x", "error": {"code": "INVALID_DOCUMENT", "message": "bad \\ud800 tail"}}')
|
||||
await asyncio.wait_for(env.worker.wait_idle(), 5)
|
||||
[(_, body)] = env.results(job_id)
|
||||
assert_error(body, ErrorCode.INVALID_DOCUMENT, raw=True)
|
||||
assert body["error"]["message"].startswith("bad ")
|
||||
assert body["error"]["message"].endswith(" tail")
|
||||
|
||||
|
||||
async def test_lone_surrogate_in_original_name_does_not_break_publishing(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
request = env.request(original_name="скан\udc80.pdf")
|
||||
data = json.dumps(request, ensure_ascii=True).encode("ascii") # в JSON суррогат приходит экранированным
|
||||
await env.deliver(data)
|
||||
await asyncio.wait_for(env.worker.wait_idle(), 5)
|
||||
[(_, body)] = env.results(request["job_id"])
|
||||
assert body["status"] == "error"
|
||||
|
||||
|
||||
# --- параллелизм и остановка --------------------------------------------------------------------------------------
|
||||
async def test_five_jobs_with_four_slots(make_env: EnvFactory) -> None:
|
||||
env = await make_env(SCAN_MAX_INFLIGHT=4)
|
||||
requests = [env.request() for _ in range(5)]
|
||||
for request in requests:
|
||||
await env.deliver(request)
|
||||
await eventually(lambda: len(env.hotfolder.puts) == 4)
|
||||
await eventually(lambda: sum(len(env.acks(r["job_id"])) for r in requests) == 5)
|
||||
await asyncio.sleep(0.05)
|
||||
assert len(env.hotfolder.puts) == 4 # пятый ждёт слот
|
||||
first_job = env.hotfolder.puts[0].split("_", 1)[0]
|
||||
env.answer(first_job, json.dumps({**SUCCESS_ANSWER, "file_uid": first_job}).encode())
|
||||
await eventually(lambda: len(env.hotfolder.puts) == 5)
|
||||
assert len(env.results(first_job)) == 1
|
||||
|
||||
|
||||
async def test_stop_cancels_jobs_and_rejects_new_requests(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
request = env.request()
|
||||
await env.deliver(request)
|
||||
await eventually(lambda: len(env.hotfolder.puts) == 1)
|
||||
await asyncio.wait_for(env.worker.stop(), 1)
|
||||
assert env.worker.active_jobs == 0
|
||||
assert env.results(request["job_id"]) == []
|
||||
await env.deliver(env.request())
|
||||
await asyncio.wait_for(env.worker.wait_idle(), 1)
|
||||
assert len(env.client.published) == 1 # только ack первого запроса
|
||||
await env.worker.stop() # повторная остановка безопасна
|
||||
|
||||
|
||||
async def test_service_info_and_handle_returns_published_message(make_env: EnvFactory) -> None:
|
||||
from src.contract import RecognitionRequest # noqa: PLC0415
|
||||
|
||||
env = await make_env()
|
||||
request = RecognitionRequest.model_validate(env.request(data=None))
|
||||
message = await env.worker.handle(request)
|
||||
assert message.service.name == "scan-gateway"
|
||||
assert message.service.version == env.settings.VERSION
|
||||
assert message.error is not None and message.error.code is ErrorCode.FILE_READ_ERROR
|
||||
[(_, body)] = env.results(str(request.job_id))
|
||||
assert body == message.to_wire()
|
||||
Loading…
x
Reference in New Issue
Block a user