feat(worker): очистка горячей папки под SCAN_CLEANUP, RECOGNITION_FAILED без расширения, OcrNormalizer в lifespan (D-2, D-4)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
Andrey Epifancev
parent
c00823c673
commit
c1782b9524
@ -3,11 +3,13 @@
|
||||
- `create_app(settings)` строит горячую папку (`create_hotfolder`, без обращения к ФС/шаре) и `HealthProvider`;
|
||||
`GET /healthcheck` отдаёт `uptime`, `version`, `nats` (`ok`/`down`/`disabled`), `hotfolder` (`ok`/`down`).
|
||||
- lifespan: регистрация SMB-сессии (только `SCAN_HOTFOLDER_MODE=smb`), S3 (`S3Storage`), NATS (`NatsGateway`), воркер
|
||||
(`Worker` с `NullNormalizer`). Подключение к NATS блокирует до доступности сервера, поэтому идёт фоновой задачей:
|
||||
HTTP отвечает сразу (`nats = down`), подписка `recognition.image.request` появляется после подключения. Сбой этой
|
||||
задачи (например, `NatsConfigError`) — лог error с трассировкой и `nats = down`, HTTP-сервер жив.
|
||||
(`Worker` с `OcrNormalizer`). Справочники нормализатора загружаются из `SCAN_RULES_DIR` в `create_app`: битый
|
||||
справочник — `ValueError`, сервис не стартует. Подключение к NATS блокирует до доступности сервера, поэтому идёт
|
||||
фоновой задачей: HTTP отвечает сразу (`nats = down`), подписка `recognition.image.request` появляется после
|
||||
подключения. Сбой этой задачи (например, `NatsConfigError`) — лог error с трассировкой и `nats = down`,
|
||||
HTTP-сервер жив.
|
||||
Остановка: отмена подключения, остановка воркера (задания в работе отменяются), закрытие NATS и S3.
|
||||
- Объекты lifespan доступны в `app.state`: `worker`, `nats`, `storage`, `health` (тесты, задачи 13–14).
|
||||
- Объекты lifespan доступны в `app.state`: `worker`, `nats`, `storage`, `normalizer`, `health`.
|
||||
- `NATS_ENABLED=false` — подключения нет (`nats = disabled`), воркер запросов не получает (dev без NATS).
|
||||
"""
|
||||
|
||||
@ -29,7 +31,8 @@ from src.core.health import HealthProvider
|
||||
from src.core.nats import NatsGateway
|
||||
from src.core.worker import Worker
|
||||
from src.gateway import create_hotfolder
|
||||
from src.normalizer import NullNormalizer
|
||||
from src.normalizer import OcrNormalizer
|
||||
from src.normalizer.rules import load_rules
|
||||
|
||||
|
||||
async def run_nats(gateway: NatsGateway, worker: Worker) -> None:
|
||||
@ -60,6 +63,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
|
||||
version = settings.VERSION
|
||||
up_time = datetime.now(UTC)
|
||||
hotfolder = create_hotfolder(settings)
|
||||
normalizer = OcrNormalizer(load_rules(settings.SCAN_RULES_DIR)) # битый справочник — сервис не стартует (D-7)
|
||||
health = HealthProvider(nats_enabled=settings.NATS_ENABLED, hotfolder=hotfolder)
|
||||
|
||||
@asynccontextmanager
|
||||
@ -76,7 +80,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
|
||||
logger.exception("Ошибка при регистрации сессии SMB")
|
||||
storage = S3Storage.from_settings(settings.OPENBAO.MINIO)
|
||||
gateway = NatsGateway(settings)
|
||||
worker = Worker(settings, storage=storage, publisher=gateway, hotfolder=hotfolder, normalizer=NullNormalizer())
|
||||
worker = Worker(settings, storage=storage, publisher=gateway, hotfolder=hotfolder, normalizer=normalizer)
|
||||
health.nats = gateway
|
||||
app.state.storage = storage
|
||||
app.state.nats = gateway
|
||||
@ -99,6 +103,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
|
||||
|
||||
app = FastAPI(title="scan-gateway", version=version, root_path=settings.ROOT_PATH, lifespan=lifespan)
|
||||
app.state.health = health
|
||||
app.state.normalizer = normalizer
|
||||
|
||||
@app.get("/healthcheck", status_code=200)
|
||||
async def healthcheck():
|
||||
|
||||
@ -1,7 +1,7 @@
|
||||
"""Воркер scan-gateway: запрос NATS → ack → S3 → горячая папка → набор ответов ИС → склейка → S3 `raw/` →
|
||||
нормализатор → конверт.
|
||||
очистка горячей папки (под флагом) → нормализатор → конверт.
|
||||
|
||||
Spec image-recognition-gateway; design D-2, D-3, D-4, D-5, D-7; решения R-14..R-20, R-24, R-25, R-30.
|
||||
Spec image-recognition-gateway; design D-1, D-2, D-3, D-4, D-5, D-7; решения R-14..R-20, R-24, R-25, R-30.
|
||||
|
||||
- `on_request(msg)` — колбэк подписки NATS. Только разбирает и проверяет тело, регистрирует `job_id` и запускает
|
||||
задачу задания; ничего не ждёт (nats-py вызывает колбэки одной подписки последовательно, T11.4).
|
||||
@ -14,6 +14,11 @@ Spec image-recognition-gateway; design D-2, D-3, D-4, D-5, D-7; решения R
|
||||
- Ошибки — результат, не молчание (таблица D-7): каждое задание заканчивается опубликованным конвертом; исключение
|
||||
задания наружу не выходит (кроме отмены при остановке); неудачная публикация — лог error (ядро закроет задание по
|
||||
`reply_deadline`).
|
||||
- Входящий файл — `<job_id>.<ext>` (D-1): расширение берётся из `original_name`; нет допустимого расширения —
|
||||
`RECOGNITION_FAILED` без записи в горячую папку.
|
||||
- Очистка (D-2): сразу после сохранения сырого ответа в `raw/` (любой исход разбора) воркер вызывает
|
||||
`HotFolder.cleanup`; удаляет она только при `SCAN_CLEANUP=true`. При `TIMEOUT`, сбое чтения S3 или записи в горячую
|
||||
папку очистки нет; ошибка очистки — одна строка лога, исход задания не меняется.
|
||||
- Методы горячей папки синхронные (R-9) — вызываются через `asyncio.to_thread`; нормализатор — тоже в потоке.
|
||||
"""
|
||||
|
||||
@ -161,9 +166,10 @@ class _Rejection:
|
||||
|
||||
@dataclass
|
||||
class _JobContext:
|
||||
"""Состояние задания, нужное для конверта при непредвиденной ошибке: сырой ответ уже в S3."""
|
||||
"""Состояние задания для конверта при непредвиденной ошибке и для очистки: сырой ответ уже в S3, имя файла."""
|
||||
|
||||
raw_s3_key: str | None = None
|
||||
inbox_name: str | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@ -445,6 +451,7 @@ class Worker:
|
||||
if isinstance(sent, RecognitionResultMessage):
|
||||
return sent
|
||||
name, started = sent
|
||||
context.inbox_name = name
|
||||
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))
|
||||
@ -511,14 +518,15 @@ class Worker:
|
||||
f"исходный файл {source.bucket}/{source.key} не прочитан из S3: {_describe(exc)}",
|
||||
)
|
||||
|
||||
# 2. Во входящий каталог как <job_id>_<original_name> (R-5, R-15)
|
||||
# 2. Во входящий каталог как <job_id>.<ext> (D-1; расширение — из original_name)
|
||||
try:
|
||||
name = inbox_name(job_id, request.original_name)
|
||||
except ValueError:
|
||||
return self._error(
|
||||
request,
|
||||
ErrorCode.RECOGNITION_FAILED,
|
||||
f"недопустимое имя файла {_fragment(request.original_name)}: файл не передан в ИС Сканирование",
|
||||
f"у файла {_fragment(request.original_name)} нет допустимого расширения: "
|
||||
"файл не передан в ИС Сканирование",
|
||||
)
|
||||
try:
|
||||
await asyncio.to_thread(self._hotfolder.put, name, data)
|
||||
@ -557,6 +565,8 @@ class Worker:
|
||||
request, ErrorCode.UNKNOWN_ERROR, f"сырой ответ ИС не сохранён в S3 ({key}): {_describe(exc)}"
|
||||
)
|
||||
context.raw_s3_key = key
|
||||
# набор прочитан и сохранён — следы задания в горячей папке больше не нужны (D-2)
|
||||
await self._cleanup(job_id, context.inbox_name)
|
||||
|
||||
# 5. Нормализатор получает склеенный ответ (D-5); его исключение — UNKNOWN_ERROR с сохранённым сырым ответом
|
||||
try:
|
||||
@ -609,6 +619,23 @@ class Worker:
|
||||
metrics=metrics,
|
||||
)
|
||||
|
||||
async def _cleanup(self, job_id: str, name: str | None) -> None:
|
||||
"""Очистка горячей папки после прочтения набора (D-2): удаляет только при `SCAN_CLEANUP` (решает горячая папка);
|
||||
ошибка — одна строка лога, исход задания не меняется."""
|
||||
if name is None:
|
||||
return
|
||||
try:
|
||||
removed = await asyncio.to_thread(self._hotfolder.cleanup, job_id, name)
|
||||
except Exception as exc: # noqa: BLE001 — любая ошибка удаления не влияет на конверт
|
||||
logger.warning(
|
||||
"Очистка горячей папки не удалась, задание продолжается. job_id={job_id}, error={error}",
|
||||
job_id=job_id,
|
||||
error=_describe(exc),
|
||||
)
|
||||
return
|
||||
if removed:
|
||||
logger.info("Горячая папка очищена. job_id={job_id}, name={name}", job_id=job_id, name=name)
|
||||
|
||||
async def _wait_answer(self, job_id: str, deadline: datetime) -> _WaitResult:
|
||||
"""Опрос `poll(job_id)` каждые `SCAN_POLL_SECONDS` до дедлайна (последний опрос — в момент дедлайна).
|
||||
|
||||
|
||||
@ -13,7 +13,7 @@ from src.core.nats import NatsGateway
|
||||
from src.core.worker import Worker
|
||||
from src.gateway.hotfolder import LocalHotFolder
|
||||
from src.normalizer import NullNormalizer
|
||||
from tests.worker_helpers import SUCCESS_ANSWER, Env, EnvFactory, FakeNatsClient, FakeStorage, SpyHotFolder
|
||||
from tests.worker_helpers import PDF_BYTES, SUCCESS_ANSWER, Env, EnvFactory, FakeNatsClient, FakeStorage, SpyHotFolder
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@ -38,6 +38,7 @@ async def make_env(tmp_path: Path) -> AsyncIterator[EnvFactory]:
|
||||
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")
|
||||
(fixtures / "ok_doc.pdf").write_bytes(PDF_BYTES) # пара ответа по sha256 содержимого (эмулятор, Task 14)
|
||||
settings = get_settings().model_copy(
|
||||
update={
|
||||
"SCAN_HOTFOLDER_PATH": str(root),
|
||||
@ -59,6 +60,7 @@ async def make_env(tmp_path: Path) -> AsyncIterator[EnvFactory]:
|
||||
settings.SCAN_OUTPUT_DIR,
|
||||
settings.SCAN_TMP_DIR,
|
||||
settle_seconds=settings.SCAN_SETTLE_SECONDS,
|
||||
cleanup=settings.SCAN_CLEANUP,
|
||||
)
|
||||
hotfolder = SpyHotFolder(local, events)
|
||||
kwargs: dict[str, Any] = {}
|
||||
|
||||
@ -15,6 +15,7 @@ from src.app import create_app
|
||||
from src.core.config import get_settings
|
||||
from src.core.nats import NatsGateway
|
||||
from src.core.worker import Worker
|
||||
from src.normalizer import OcrNormalizer
|
||||
|
||||
pytestmark = pytest.mark.usefixtures("log_lines")
|
||||
|
||||
@ -215,3 +216,15 @@ async def test_unhandled_error_has_json_500(local_env: Path) -> None:
|
||||
|
||||
assert response.status_code == 500
|
||||
assert response.json() == {"detail": "внутренняя ошибка сервера"}
|
||||
|
||||
|
||||
def test_create_app_fails_on_broken_rules(tmp_path: Path) -> None:
|
||||
(tmp_path / "statement_lines.yaml").write_text("editions: []\n", encoding="utf-8")
|
||||
settings = get_settings().model_copy(update={"SCAN_RULES_DIR": str(tmp_path), "SCAN_HOTFOLDER_PATH": str(tmp_path)})
|
||||
with pytest.raises(ValueError, match="statement_lines.yaml"):
|
||||
create_app(settings)
|
||||
|
||||
|
||||
def test_create_app_uses_ocr_normalizer() -> None:
|
||||
app = create_app(get_settings().model_copy(update={"SCAN_HOTFOLDER_PATH": "."}))
|
||||
assert isinstance(app.state.normalizer, OcrNormalizer)
|
||||
|
||||
@ -27,7 +27,8 @@ from src.core.clients import JSON_CONTENT_TYPE
|
||||
from src.core.worker import MAX_ERROR_MESSAGE_CHARS, MAX_WARNING_CHARS, MAX_WARNINGS
|
||||
from src.gateway.hotfolder import AnswerFile
|
||||
from src.gateway.merge import dump_merged, merge_answers
|
||||
from src.normalizer import NOT_IMPLEMENTED_MESSAGE, NOT_OBJECT_MESSAGE, NormalizeOutcome
|
||||
from src.normalizer import NOT_IMPLEMENTED_MESSAGE, NOT_OBJECT_MESSAGE, NormalizeOutcome, OcrNormalizer
|
||||
from src.normalizer.rules import load_rules
|
||||
from tests.worker_helpers import (
|
||||
API_DIR,
|
||||
BUCKET,
|
||||
@ -112,16 +113,22 @@ async def test_ack_is_published_before_s3_read_and_before_file_appears(make_env:
|
||||
async with emulator(env):
|
||||
await env.run(request)
|
||||
job_id = request["job_id"]
|
||||
name = f"{job_id}_ok_doc.pdf"
|
||||
name = f"{job_id}.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)]
|
||||
# очистка вызывается всегда после записи (шпион её журналирует; удаляет `HotFolder.cleanup` только при SCAN_CLEANUP)
|
||||
assert kinds == [
|
||||
("publish", f"ack:{job_id}:1"),
|
||||
("s3_get", request["s3"]["key"]),
|
||||
("put", name),
|
||||
("cleanup", 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"
|
||||
path = env.local.input_dir / f"{request['job_id']}.pdf"
|
||||
await eventually(path.exists)
|
||||
assert path.read_bytes() == PDF_BYTES
|
||||
await env.worker.stop()
|
||||
@ -137,8 +144,8 @@ async def test_repeated_request_gets_second_ack_and_one_file(make_env: EnvFactor
|
||||
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 env.inbox() == [f"{job_id}.pdf"]
|
||||
assert env.hotfolder.puts == [f"{job_id}.pdf"]
|
||||
assert len(env.storage.get_calls) == 1
|
||||
|
||||
|
||||
@ -152,7 +159,7 @@ async def test_repeated_request_after_result_gets_ack_without_reprocessing(make_
|
||||
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"]
|
||||
assert env.hotfolder.puts == [f"{job_id}.pdf"]
|
||||
|
||||
|
||||
async def test_done_cache_is_bounded_lru(make_env: EnvFactory) -> None:
|
||||
@ -201,7 +208,7 @@ async def test_success_answer_gives_not_implemented_with_raw_key(make_env: EnvFa
|
||||
assert _raw(env, job_id) == merged_raw(job_id, (f"{job_id}.json", answer, None))
|
||||
# сырой ответ сохранён раньше публикации результата
|
||||
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"]
|
||||
assert env.inbox() == [f"{job_id}.pdf.done"]
|
||||
|
||||
|
||||
@pytest.mark.fixtures
|
||||
@ -212,6 +219,7 @@ async def test_real_fixture_error_is_passed_through(make_env: EnvFactory, tmp_pa
|
||||
real.mkdir()
|
||||
for source in REAL_FIXTURES.glob("*.json"):
|
||||
shutil.copy(source, real / source.name)
|
||||
(real / "aep_5.pdf").write_bytes(PDF_BYTES) # пара ответа `aep_5.json` по sha256 содержимого запроса (Task 14)
|
||||
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")
|
||||
@ -226,11 +234,11 @@ async def test_real_fixture_error_is_passed_through(make_env: EnvFactory, tmp_pa
|
||||
|
||||
async def test_unknown_name_gets_emulator_invalid_document(make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
request = env.request(original_name="unknown.pdf")
|
||||
request = env.request(original_name="unknown.pdf", data=b"%PDF unknown content") # пары по содержимому нет
|
||||
async with emulator(env):
|
||||
body = await env.run(request)
|
||||
assert_error(body, ErrorCode.INVALID_DOCUMENT, raw=True)
|
||||
assert "unknown.pdf" in body["error"]["message"]
|
||||
assert "нет пары по содержимому" in body["error"]["message"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("status", [ResultStatus.DONE, ResultStatus.NEEDS_CHECK])
|
||||
@ -556,8 +564,8 @@ async def test_timeout_by_scan_timeout_keeps_file_and_never_resends(make_env: En
|
||||
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.inbox() == [f"{request['job_id']}.pdf"] # входящий файл остаётся
|
||||
assert env.hotfolder.puts == [f"{request['job_id']}.pdf"] # повторной отправки нет
|
||||
assert env.hotfolder.polls >= 5
|
||||
|
||||
|
||||
@ -613,7 +621,7 @@ async def test_unusable_original_name_is_recognition_failed(make_env: EnvFactory
|
||||
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 "расширени" in body["error"]["message"]
|
||||
assert env.hotfolder.puts == []
|
||||
|
||||
|
||||
@ -881,3 +889,77 @@ async def test_service_info_and_handle_returns_published_message(make_env: EnvFa
|
||||
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()
|
||||
|
||||
|
||||
# --- Срез 3a: имя <job_id>.<ext>, очистка под флагом, OcrNormalizer (D-1, D-2, решения №2, №3, №26) -----------------
|
||||
class TestSlice3a:
|
||||
async def test_no_extension_is_recognition_failed(self, make_env: EnvFactory) -> None:
|
||||
env = await make_env()
|
||||
body = env.request(original_name="scan")
|
||||
result = await env.run(body)
|
||||
assert result["status"] == "error" and result["error"]["code"] == "RECOGNITION_FAILED"
|
||||
assert "расширени" in result["error"]["message"]
|
||||
assert env.inbox() == []
|
||||
|
||||
async def test_cleanup_after_answer_when_enabled(self, make_env: EnvFactory) -> None:
|
||||
env = await make_env(SCAN_CLEANUP=True)
|
||||
body = env.request()
|
||||
async with emulator(env):
|
||||
result = await env.run(body)
|
||||
assert result["status"] == "error" and result["error"]["code"] == "NOT_IMPLEMENTED" # NullNormalizer
|
||||
assert env.inbox() == [] # и <job>.pdf, и <job>.pdf.done удалены
|
||||
assert not env.local.answer_dir(body["job_id"]).exists()
|
||||
assert ("cleanup", f"{body['job_id']}.pdf") in env.events
|
||||
|
||||
async def test_no_cleanup_when_disabled(self, make_env: EnvFactory) -> None:
|
||||
env = await make_env(SCAN_CLEANUP=False)
|
||||
body = env.request()
|
||||
async with emulator(env):
|
||||
await env.run(body)
|
||||
assert env.inbox() == [f"{body['job_id']}.pdf.done"]
|
||||
assert env.local.answer_dir(body["job_id"]).is_dir()
|
||||
|
||||
async def test_no_cleanup_on_timeout(self, make_env: EnvFactory) -> None:
|
||||
env = await make_env(SCAN_CLEANUP=True)
|
||||
body = env.request(deadline_s=0.2)
|
||||
result = await env.run(body)
|
||||
assert result["error"]["code"] == "TIMEOUT"
|
||||
assert env.inbox() == [f"{body['job_id']}.pdf"]
|
||||
assert ("cleanup", f"{body['job_id']}.pdf") not in env.events
|
||||
|
||||
async def test_cleanup_error_does_not_change_result(
|
||||
self, make_env: EnvFactory, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
env = await make_env(SCAN_CLEANUP=True)
|
||||
|
||||
def deny(job_id: str, name: str) -> bool:
|
||||
raise PermissionError("нет прав")
|
||||
|
||||
monkeypatch.setattr(env.hotfolder.inner, "cleanup", deny)
|
||||
body = env.request()
|
||||
async with emulator(env):
|
||||
result = await env.run(body)
|
||||
assert result["status"] == "error" and result["error"]["code"] == "NOT_IMPLEMENTED"
|
||||
assert "metrics" not in result or "warnings" not in result.get("metrics", {})
|
||||
|
||||
async def test_existing_input_waits_for_answer(self, make_env: EnvFactory) -> None:
|
||||
"""Review Focus 4: входящий файл уже лежит (повтор задания), ответа нет — TIMEOUT, файл не перезаписан."""
|
||||
env = await make_env()
|
||||
body = env.request(deadline_s=0.3)
|
||||
env.local.put(f"{body['job_id']}.pdf", b"already there")
|
||||
result = await env.run(body)
|
||||
assert result["error"]["code"] == "TIMEOUT"
|
||||
assert (env.local.input_dir / f"{body['job_id']}.pdf").read_bytes() == b"already there"
|
||||
assert env.inbox() == [f"{body['job_id']}.pdf"]
|
||||
|
||||
async def test_ocr_normalizer_produces_documents(self, make_env: EnvFactory) -> None:
|
||||
env = await make_env(normalizer=OcrNormalizer(load_rules(API_DIR / "rules")))
|
||||
body = env.request()
|
||||
async with emulator(env):
|
||||
result = await env.run(body)
|
||||
assert result["status"] == "needs_check" # SUCCESS_ANSWER: confidence < 100 в шапке
|
||||
key = result["result_s3"]["key"]
|
||||
stored = json.loads(env.storage.objects[(BUCKET, key)])
|
||||
assert stored["documents"][0]["document_type"] == "BALANCE"
|
||||
assert stored["documents"][0]["lines"][0]["line_code"] == "1600"
|
||||
assert result["raw_s3_key"] == f"raw/{body['job_id']}.json"
|
||||
|
||||
@ -76,8 +76,8 @@ class SteppedSleep:
|
||||
|
||||
|
||||
def job_of(name: str) -> str:
|
||||
"""`job_id` из имени файла входящего каталога `<job_id>_<original_name>`."""
|
||||
return name.split("_", 1)[0]
|
||||
"""`job_id` из имени файла входящего каталога `<job_id>.<ext>`."""
|
||||
return name.split(".", 1)[0]
|
||||
|
||||
|
||||
def answer(env: Env, job_id: str) -> None:
|
||||
@ -152,7 +152,7 @@ async def test_five_jobs_with_four_slots(make_env: EnvFactory) -> None:
|
||||
waiting_request = requests[jobs.index(waiting)]
|
||||
assert max(index_of(env, ("publish", f"ack:{job}:1")) for job in jobs) < freed_result
|
||||
assert freed_result < index_of(env, ("s3_get", source_key(waiting_request)))
|
||||
assert freed_result < index_of(env, ("put", f"{waiting}_ok_doc.pdf"))
|
||||
assert freed_result < index_of(env, ("put", f"{waiting}.pdf"))
|
||||
|
||||
# Все пять заданий заканчиваются конвертами
|
||||
await finish_all(env, poll_sleep, jobs)
|
||||
@ -188,7 +188,7 @@ async def test_single_slot_serialises_jobs(make_env: EnvFactory) -> None:
|
||||
]
|
||||
expected: list[tuple[str, str]] = []
|
||||
for job in order:
|
||||
expected += [("put", f"{job}_ok_doc.pdf"), ("publish", f"result:{job}:1")]
|
||||
expected += [("put", f"{job}.pdf"), ("publish", f"result:{job}:1")]
|
||||
assert files_and_results == expected
|
||||
await finish_all(env, poll_sleep, jobs)
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user