diff --git a/api/.env.example b/api/.env.example index 9130ef6..98c650a 100644 --- a/api/.env.example +++ b/api/.env.example @@ -33,11 +33,16 @@ SCAN_HOTFOLDER_MODE=local # Корень горячей папки в режиме local: каталог должен существовать (mkdir -p .hotfolder). # Путь считается от каталога api/; его же читает эмулятор. SCAN_HOTFOLDER_PATH=./.hotfolder -# Подкаталоги корня: «Входящие» (сюда кладётся UUID_имя) и «Исходящие» (оттуда забирается /.json) +# Подкаталоги корня: «Входящие» (сюда кладётся UUID_имя) и «Исходящие» (оттуда забираются все *.json из /) SCAN_INPUT_DIR=Input SCAN_OUTPUT_DIR=Output +# Служебный каталог корня: временный .part пишется сюда и переименовывается во «Входящие» (та же ФС/шара — rename +# атомарен, во «Входящих» не бывает недописанных файлов); создаётся при первой записи +SCAN_TMP_DIR=Tmp # Период опроса «Исходящих», секунды SCAN_POLL_SECONDS=5 +# Набор *.json в / считается готовым, когда имена и размеры файлов не менялись столько секунд (>= 0) +SCAN_SETTLE_SECONDS=2 # Потолок ожидания ответа ИС, секунды (дополнительно ограничен reply_deadline запроса); повторной отправки нет SCAN_TIMEOUT_SECONDS=300 # Сколько заданий обрабатывается одновременно diff --git a/api/.env.test b/api/.env.test index 6634e6b..74b1d06 100644 --- a/api/.env.test +++ b/api/.env.test @@ -20,6 +20,10 @@ SCAN_HOTFOLDER_MODE=local SCAN_HOTFOLDER_PATH=./.hotfolder-test SCAN_INPUT_DIR=Input SCAN_OUTPUT_DIR=Output +SCAN_TMP_DIR=Tmp # Укороченные интервалы, чтобы тесты не ждали по 5 секунд SCAN_POLL_SECONDS=0.1 SCAN_TIMEOUT_SECONDS=5 +# R-23: набор ответа готов сразу (тесты воркера/таймаута/параллелизма не ждут стабилизации); settle проверяется +# отдельными тестами горячей папки с подменёнными часами +SCAN_SETTLE_SECONDS=0 diff --git a/api/src/core/config.py b/api/src/core/config.py index df53a8b..d677e2e 100644 --- a/api/src/core/config.py +++ b/api/src/core/config.py @@ -11,7 +11,7 @@ import os from functools import lru_cache from typing import Literal -from pydantic import PositiveFloat, PositiveInt, model_validator +from pydantic import NonNegativeFloat, PositiveFloat, PositiveInt, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict @@ -68,7 +68,10 @@ class Settings(BaseSettings): SCAN_HOTFOLDER_PATH: str = "" # корень горячей папки в режиме local (каталог должен существовать) SCAN_INPUT_DIR: str = "Input" SCAN_OUTPUT_DIR: str = "Output" + SCAN_TMP_DIR: str = "Tmp" # служебный каталог временных файлов записи (рядом с Input, на той же ФС/шаре) SCAN_POLL_SECONDS: PositiveFloat = 5 # период опроса «Исходящих» + # набор *.json ответа готов, когда имена и размеры не менялись столько секунд; 0 — готов сразу (тесты) + SCAN_SETTLE_SECONDS: NonNegativeFloat = 2 SCAN_TIMEOUT_SECONDS: PositiveFloat = 300 # потолок ожидания ответа ИС (дополнительно к reply_deadline запроса) SCAN_MAX_INFLIGHT: PositiveInt = 4 # одновременно обрабатываемых заданий diff --git a/api/src/core/worker.py b/api/src/core/worker.py index 89b96d3..4a7dd75 100644 --- a/api/src/core/worker.py +++ b/api/src/core/worker.py @@ -47,7 +47,7 @@ from src.contract import ( 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.gateway.hotfolder import AnswerFile, HotFolder, inbox_name from src.normalizer import Normalizer __all__ = [ @@ -610,7 +610,8 @@ class Worker: polls += 1 answer: bytes | None try: - answer = await asyncio.to_thread(self._hotfolder.poll, job_id) + # переходно (Task 3): склейка всех файлов + answer = _first_answer(await asyncio.to_thread(self._hotfolder.poll, job_id)) except Exception as exc: failures += 1 last_error = exc @@ -630,5 +631,13 @@ class Worker: await self._sleep(min(self._settings.SCAN_POLL_SECONDS, (deadline - now).total_seconds())) +def _first_answer(files: list[AnswerFile] | None) -> bytes | None: + """Байты первого (по имени) файла готового набора ответа: один файл — его байты, как раньше. + + Переходно (Task 3): склейка всех файлов набора заменит этот выбор. + """ + return files[0].data if files else None + + def _is_uuid(value: object) -> bool: return isinstance(value, str) and _UUID_RE.fullmatch(value) is not None diff --git a/api/src/gateway/__init__.py b/api/src/gateway/__init__.py index 2bfd56c..4902baa 100644 --- a/api/src/gateway/__init__.py +++ b/api/src/gateway/__init__.py @@ -14,7 +14,13 @@ def create_hotfolder(settings: Settings) -> HotFolder: """Горячая папка для режима `SCAN_HOTFOLDER_MODE`. К файловой системе и шаре не обращается.""" mode = settings.SCAN_HOTFOLDER_MODE if mode == "local": - return LocalHotFolder(settings.SCAN_HOTFOLDER_PATH, settings.SCAN_INPUT_DIR, settings.SCAN_OUTPUT_DIR) + return LocalHotFolder( + settings.SCAN_HOTFOLDER_PATH, + settings.SCAN_INPUT_DIR, + settings.SCAN_OUTPUT_DIR, + settings.SCAN_TMP_DIR, + settle_seconds=settings.SCAN_SETTLE_SECONDS, + ) if mode == "smb": from src.gateway.smb import SmbHotFolder # noqa: PLC0415 — ленивый импорт: в режиме local smb не нужен (R-6) @@ -24,7 +30,9 @@ def create_hotfolder(settings: Settings) -> HotFolder: smb.SMB_PATH, settings.SCAN_INPUT_DIR, settings.SCAN_OUTPUT_DIR, + settings.SCAN_TMP_DIR, login=smb.SMB_LOGIN, password=smb.SMB_PASSWORD, + settle_seconds=settings.SCAN_SETTLE_SECONDS, ) raise ValueError(f"неизвестный SCAN_HOTFOLDER_MODE: {mode!r}") diff --git a/api/src/gateway/hotfolder.py b/api/src/gateway/hotfolder.py index 83e4317..144b6b6 100644 --- a/api/src/gateway/hotfolder.py +++ b/api/src/gateway/hotfolder.py @@ -1,11 +1,20 @@ """Горячие папки ИС Сканирование: протокол `HotFolder` и реализация в обычном каталоге (D-1, R-4, R-5). -Файл кладётся во входящий каталог как `_`, ответ ИС забирается из исходящего каталога по пути -`//.json`. Имена каталогов (`Input`/`Output`) настраиваются, а не зашиты в код. +Файл кладётся во входящий каталог как `_`: сначала временный `.<имя>.part` в служебном каталоге +`Tmp` (на той же ФС/шаре), затем атомарный rename во входящий (OQ-79). Ответ ИС — набор файлов каталога +`//`: ИС кладёт туда все распознанные документы задания, по одному `*.json` на документ (OQ-78). +`poll` отдаёт все `*.json` по возрастанию имени, когда набор готов: есть хотя бы один `*.json`, а набор имён и размеров +не менялся `settle_seconds` (`SCAN_SETTLE_SECONDS`, `SettleTracker`). Имена каталогов (`Input`/`Output`/`Tmp`) +настраиваются, а не зашиты в код. """ import contextlib import os +import threading +import time +from collections import OrderedDict +from collections.abc import Callable +from dataclasses import dataclass from pathlib import Path from typing import Protocol @@ -93,6 +102,76 @@ def _fit_name(job_id: str, base: str) -> str: return (stem or "_") + extension +@dataclass(frozen=True) +class AnswerFile: + """Один файл ответа ИС: имя в каталоге `//` и байты как есть.""" + + name: str + data: bytes + + +# Снимок набора ответа: (имя, размер) всех `*.json` каталога задания по возрастанию имени +Snapshot = tuple[tuple[str, int], ...] +MonotonicFn = Callable[[], float] +# Предел числа заданий, для которых помнится ненастоявшийся набор (задание ушло по таймауту — состояние вытесняется) +MAX_PENDING_JOBS = 1024 + + +def is_answer_name(name: str) -> bool: + """Файл ответа ИС — `*.json`; расширение без учёта регистра (маски Windows/SMB регистронезависимы).""" + return name.lower().endswith(".json") + + +class SettleTracker: + """Готовность набора ответа по стабилизации (settle): набор считается готовым, когда он не пуст и одинаковый + снимок (имена и размеры) наблюдается не меньше `settle_seconds` по часам `clock`. + + Состояние — по `job_id`: снимок и момент, с которого он не менялся. Оно сбрасывается, когда набор признан готовым + или опустел (каталога/json нет), а число заданий ограничено `max_jobs` (самые давние вытесняются: у задания, + ушедшего по таймауту, состояние иначе осталось бы навсегда; вытесненное активное задание настаивается заново). + `settle_seconds == 0` — готов любой непустой набор, состояние не хранится. Потокобезопасно: `poll` вызывается + через `asyncio.to_thread`. + """ + + def __init__( + self, settle_seconds: float, clock: MonotonicFn = time.monotonic, *, max_jobs: int = MAX_PENDING_JOBS + ) -> None: + if settle_seconds < 0: + raise ValueError(f"settle_seconds не может быть отрицательным: {settle_seconds!r}") + self.settle_seconds = settle_seconds + self._clock = clock + self._max_jobs = max_jobs + self._lock = threading.Lock() + self._seen: OrderedDict[str, tuple[Snapshot, float]] = OrderedDict() + + def ready(self, job_id: str, snapshot: Snapshot) -> bool: + """Набор `snapshot` задания готов. Готовый набор забывается: следующий вызов настаивает набор заново.""" + if not snapshot or self.settle_seconds == 0: + self.forget(job_id) + return bool(snapshot) + now = self._clock() + with self._lock: + seen = self._seen.get(job_id) + if seen is not None and seen[0] == snapshot: + if now - seen[1] >= self.settle_seconds: + del self._seen[job_id] + return True + return False + self._seen[job_id] = (snapshot, now) + self._seen.move_to_end(job_id) + while len(self._seen) > self._max_jobs: + self._seen.popitem(last=False) + return False + + def forget(self, job_id: str) -> None: + with self._lock: + self._seen.pop(job_id, None) + + def __len__(self) -> int: + with self._lock: + return len(self._seen) + + class HotFolder(Protocol): """Обмен файлами с ИС Сканирование через пару каталогов. @@ -100,12 +179,15 @@ class HotFolder(Protocol): """ def put(self, name: str, data: bytes) -> None: - """Атомарно положить файл `name` (простое имя, без пути) во входящий каталог. Имя собирает вызывающий - (`inbox_name`). Если горячая папка недоступна — исключение (воркер превращает его в `RECOGNITION_FAILED`).""" + """Атомарно положить файл `name` (простое имя, без пути) во входящий каталог: временный файл — в служебном + каталоге `Tmp`, затем rename. Имя собирает вызывающий (`inbox_name`). Если горячая папка недоступна — + исключение (воркер превращает его в `RECOGNITION_FAILED`).""" ... - def poll(self, job_id: str) -> bytes | None: - """Байты ответа ИС `//.json` или `None`, пока ответа нет.""" + def poll(self, job_id: str) -> list[AnswerFile] | None: + """Все `*.json` каталога `//` по возрастанию имени, когда набор готов (не пуст и не менялся + `SCAN_SETTLE_SECONDS`); `None`, пока ответа нет или набор ещё пишется. Вызывается повторно с тем же `job_id`: + реализация помнит снимок набора между вызовами.""" ... def healthy(self) -> bool: @@ -116,29 +198,47 @@ class HotFolder(Protocol): class LocalHotFolder: """Горячая папка в обычном каталоге: dev, compose-том, unit-тесты. - `root`, `input_dir`, `output_dir` и `answer_path` открыты намеренно (R-4): эмулятор ИС Сканирование работает с - ними обычными операциями файловой системы (листинг входящих, `.done`, запись ответа). + `root`, `input_dir`, `output_dir`, `tmp_dir`, `answer_dir` и `answer_path` открыты намеренно (R-4): эмулятор + ИС Сканирование работает с ними обычными операциями файловой системы (листинг входящих, `.done`, запись ответа). + `settle_seconds` и `clock` (монотонные часы) — готовность набора ответа (`SettleTracker`). """ - def __init__(self, root: Path | str, input_dir: str = "Input", output_dir: str = "Output") -> None: + def __init__( # noqa: PLR0913 — имена трёх каталогов + параметры готовности набора + self, + root: Path | str, + input_dir: str = "Input", + output_dir: str = "Output", + tmp_dir: str = "Tmp", + *, + settle_seconds: float = 2, + clock: MonotonicFn = time.monotonic, + ) -> None: self.root: Path = Path(root) self.input_dir: Path = self.root / input_dir self.output_dir: Path = self.root / output_dir + self.tmp_dir: Path = self.root / tmp_dir + self._settle = SettleTracker(settle_seconds, clock) + + def answer_dir(self, job_id: str) -> Path: + """Каталог ответа ИС: `/`.""" + validate_job_id(job_id) + return self.output_dir / job_id def answer_path(self, job_id: str) -> Path: - """Путь ответа ИС: `//.json`.""" - validate_job_id(job_id) - return self.output_dir / job_id / f"{job_id}.json" + """Путь основного файла ответа `//.json` (так его называет ИС и эмулятор).""" + return self.answer_dir(job_id) / f"{job_id}.json" def put(self, name: str, data: bytes) -> None: - """Атомарная запись: `..part` в том же каталоге, затем `os.replace` (существующий файл заменяется). + """Атомарная запись: `..part` в служебном каталоге `tmp_dir`, затем `os.replace` в `input_dir/` + (существующий файл заменяется; оба каталога на одной ФС — rename атомарен). - Корень не создаётся (нет корня — `FileNotFoundError`), входящий каталог создаётся без родителей. При сбое - временный файл удаляется (ошибка удаления не маскирует исходную). + Корень не создаётся (нет корня — `FileNotFoundError`), служебный и входящий каталоги создаются без родителей. + При сбое временный файл удаляется (ошибка удаления не маскирует исходную). """ validate_name(name) + self.tmp_dir.mkdir(exist_ok=True) self.input_dir.mkdir(exist_ok=True) - part = self.input_dir / f".{name}.part" + part = self.tmp_dir / f".{name}.part" try: part.write_bytes(data) os.replace(part, self.input_dir / name) @@ -148,15 +248,32 @@ class LocalHotFolder: part.unlink(missing_ok=True) raise - def poll(self, job_id: str) -> bytes | None: - """Содержимое ответа или `None`, пока файла нет (в том числе если он исчез между проверкой и чтением).""" - path = self.answer_path(job_id) - if not path.exists(): - return None - try: - return path.read_bytes() - except FileNotFoundError: + def poll(self, job_id: str) -> list[AnswerFile] | None: + """Готовый набор `*.json` каталога ответа или `None` (нет корня/каталога/json-файлов, набор ещё + настаивается, файл исчез или изменился между листингом и чтением). Другие ошибки ФС поднимаются.""" + folder = self.answer_dir(job_id) + snapshot = self._snapshot(folder) + if not self._settle.ready(job_id, snapshot): return None + files: list[AnswerFile] = [] + for name, size in snapshot: + try: + data = (folder / name).read_bytes() + except FileNotFoundError: + return None + if len(data) != size: # файл изменился после листинга — набор настаивается заново + return None + files.append(AnswerFile(name, data)) + return files + + @property + def settle_seconds(self) -> float: + """`SCAN_SETTLE_SECONDS`: сколько набор ответа должен не меняться, чтобы считаться готовым.""" + return self._settle.settle_seconds + + def pending_jobs(self) -> int: + """Число заданий с ненастоявшимся набором (для тестов и диагностики).""" + return len(self._settle) def healthy(self) -> bool: """Корень существует и является каталогом.""" @@ -164,3 +281,20 @@ class LocalHotFolder: return self.root.is_dir() except (OSError, ValueError): return False + + @staticmethod + def _snapshot(folder: Path) -> Snapshot: + try: + with os.scandir(folder) as entries: + found = [] + for entry in entries: + if not is_answer_name(entry.name): + continue + try: + if entry.is_file(): + found.append((entry.name, entry.stat().st_size)) + except FileNotFoundError: + continue + except (FileNotFoundError, NotADirectoryError): + return () + return tuple(sorted(found)) diff --git a/api/src/gateway/smb.py b/api/src/gateway/smb.py index 2669d02..4bad421 100644 --- a/api/src/gateway/smb.py +++ b/api/src/gateway/smb.py @@ -4,7 +4,9 @@ r"""Горячая папка на SMB-шаре платформы (`SCAN_HOTFOL в lifespan приложения (`app.py`); логин/пароль, если заданы, передаются ещё и в каждый вызов `smbclient`, чтобы сессия поднималась заново по требованию (после сбоя регистрации при старте или сброса соединения). -Пути — UNC: `\\\\\<имя>` и `\\\\\\.json`. +Пути — UNC: временный файл `\\\\\.<имя>.part` переименовывается во входящий +`\\\\\<имя>` (одна шара — rename атомарен, OQ-79); ответ — все `*.json` каталога +`\\\\\\` (OQ-78), готовность — по стабилизации набора (`SettleTracker`). Ошибки smbprotocol — `SMBOSError` (подкласс `OSError`, но НЕ `FileNotFoundError`): «нет файла/каталога» распознаётся по `errno == ENOENT`. """ @@ -12,12 +14,21 @@ r"""Горячая папка на SMB-шаре платформы (`SCAN_HOTFOL import contextlib import errno import ntpath +import time from typing import Any import smbclient from smbprotocol.header import NtStatus -from src.gateway.hotfolder import validate_job_id, validate_name +from src.gateway.hotfolder import ( + AnswerFile, + MonotonicFn, + SettleTracker, + Snapshot, + is_answer_name, + validate_job_id, + validate_name, +) # Ответ есть, но пока не читается: ИС держит файл открытым на запись или файл удаляется — опрос повторится позже _NOT_READY = frozenset({NtStatus.STATUS_SHARING_VIOLATION, NtStatus.STATUS_DELETE_PENDING}) @@ -49,37 +60,47 @@ def build_unc_root(host: str, path: str) -> str: class SmbHotFolder: """Горячая папка ИС Сканирование на SMB-шаре. Методы синхронные (блокирующий SMB), семантика как у - `LocalHotFolder`: те же правила имён, атомарная запись `..part` + replace, `Input` создаётся без родителей, - корень не создаётся.""" + `LocalHotFolder`: те же правила имён, атомарная запись `Tmp\\..part` + replace в `Input`, `Tmp` и `Input` + создаются без родителей, корень не создаётся, готовность набора ответа — по стабилизации (`SettleTracker`).""" - def __init__( # noqa: PLR0913 — адрес и учётные данные D-1 + имена каталогов, как у LocalHotFolder + def __init__( # noqa: PLR0913 — адрес и учётные данные D-1 + имена каталогов и готовность, как у LocalHotFolder self, host: str, path: str, input_dir: str = "Input", output_dir: str = "Output", + tmp_dir: str = "Tmp", *, login: str = "", password: str = "", + settle_seconds: float = 2, + clock: MonotonicFn = time.monotonic, ) -> None: self.root: str = build_unc_root(host, path) self.input_dir: str = ntpath.join(self.root, input_dir) self.output_dir: str = ntpath.join(self.root, output_dir) + self.tmp_dir: str = ntpath.join(self.root, tmp_dir) self._credentials: dict[str, Any] = {"username": login, "password": password} if login else {} + self._settle = SettleTracker(settle_seconds, clock) + + def answer_dir(self, job_id: str) -> str: + r"""UNC-путь каталога ответа ИС: `\`.""" + validate_job_id(job_id) + return ntpath.join(self.output_dir, job_id) def answer_path(self, job_id: str) -> str: - r"""UNC-путь ответа ИС: `\\.json`.""" - validate_job_id(job_id) - return ntpath.join(self.output_dir, job_id, f"{job_id}.json") + r"""UNC-путь основного файла ответа: `\\.json`.""" + return ntpath.join(self.answer_dir(job_id), f"{job_id}.json") def put(self, name: str, data: bytes) -> None: - """Записать `..part` во входящий каталог и переименовать в `name` (`smbclient.replace`: существующий - файл заменяется). Имя проверяется до любого обращения к шаре. При сбое временный файл удаляется (ошибка - удаления не маскирует исходную). Нет корня — `SMBOSError` (ENOENT) от `mkdir`. + """Записать `..part` в служебный каталог и переименовать в `\\` (`smbclient.replace`: + существующий файл заменяется). Имя проверяется до любого обращения к шаре. При сбое временный файл удаляется + (ошибка удаления не маскирует исходную). Нет корня — `SMBOSError` (ENOENT) от первого `mkdir`. """ validate_name(name) - self._ensure_input_dir() - part = ntpath.join(self.input_dir, f".{name}.part") + self._ensure_dir(self.tmp_dir) + self._ensure_dir(self.input_dir) + part = ntpath.join(self.tmp_dir, f".{name}.part") try: with smbclient.open_file(part, mode="wb", **self._credentials) as fd: fd.write(data) @@ -89,21 +110,39 @@ class SmbHotFolder: smbclient.remove(part, **self._credentials) raise - def poll(self, job_id: str) -> bytes | None: - r"""Байты `\\.json` или `None`, пока его нет (нет файла, каталога задания или - `Output`) или он ещё не готов (ИС держит его открытым — `STATUS_SHARING_VIOLATION`, удаляется — - `STATUS_DELETE_PENDING`). Остальные ошибки SMB поднимаются.""" - path = self.answer_path(job_id) + def poll(self, job_id: str) -> list[AnswerFile] | None: + r"""Готовый набор `*.json` каталога `\\` по возрастанию имени или `None`: каталога/`Output`/ + json-файлов нет, набор ещё настаивается, файл исчез или изменился между листингом и чтением, ИС ещё держит + файл (`STATUS_SHARING_VIOLATION`, удаляется — `STATUS_DELETE_PENDING`). Остальные ошибки SMB поднимаются.""" + folder = self.answer_dir(job_id) try: - # Монопольное открытие (share_access по умолчанию 0): пока ИС пишет файл, оно не проходит, и недописанный - # JSON не читается - with smbclient.open_file(path, mode="rb", **self._credentials) as fd: - data: bytes = fd.read() + snapshot = self._snapshot(folder) + if not self._settle.ready(job_id, snapshot): + return None + files: list[AnswerFile] = [] + for name, size in snapshot: + # Монопольное открытие (share_access по умолчанию 0): пока ИС пишет файл, оно не проходит, и + # недописанный JSON не читается + with smbclient.open_file(ntpath.join(folder, name), mode="rb", **self._credentials) as fd: + data: bytes = fd.read() + if len(data) != size: # файл изменился после листинга — набор настаивается заново + return None + files.append(AnswerFile(name, data)) except OSError as err: if _has_errno(err, errno.ENOENT, FileNotFoundError) or getattr(err, "ntstatus", None) in _NOT_READY: + self._settle.forget(job_id) return None raise - return data + return files + + @property + def settle_seconds(self) -> float: + """`SCAN_SETTLE_SECONDS`: сколько набор ответа должен не меняться, чтобы считаться готовым.""" + return self._settle.settle_seconds + + def pending_jobs(self) -> int: + """Число заданий с ненастоявшимся набором (для тестов и диагностики).""" + return len(self._settle) def healthy(self) -> bool: """Корень доступен и является каталогом (`smbclient.listdir` проходит). Не бросает исключений.""" @@ -113,23 +152,33 @@ class SmbHotFolder: return False return True - def _ensure_input_dir(self) -> None: - """`mkdir` входящего каталога, как `os.makedirs(exist_ok=True)` без родителей. + def _snapshot(self, folder: str) -> Snapshot: + """(имя, размер) `*.json`-файлов каталога по возрастанию имени; размер — из листинга (`end_of_file`), без + отдельного stat. Нет каталога — ENOENT-ошибка `scandir` (её обрабатывает `poll`).""" + found = [ + (entry.name, int(entry.smb_info.end_of_file)) + for entry in smbclient.scandir(folder, **self._credentials) + if is_answer_name(entry.name) and entry.is_file() + ] + return tuple(sorted(found)) + + def _ensure_dir(self, path: str) -> None: + """`mkdir` служебного или входящего каталога, как `os.makedirs(exist_ok=True)` без родителей. «Уже существует» (EEXIST) — не ошибка. Другая ошибка (например, `STATUS_ACCESS_DENIED`: учётке нельзя - создавать подкаталоги в корне, хотя писать в `Input` можно) — тоже не ошибка, если каталог уже есть + создавать подкаталоги в корне, хотя писать в каталог можно) — тоже не ошибка, если каталог уже есть (`smbclient.path.isdir`); иначе поднимается исходная ошибка `mkdir`. ENOENT (нет корня) — сразу наружу. """ try: - smbclient.mkdir(self.input_dir, **self._credentials) + smbclient.mkdir(path, **self._credentials) except OSError as err: if _has_errno(err, errno.EEXIST, FileExistsError): return - if _has_errno(err, errno.ENOENT, FileNotFoundError) or not self._input_dir_exists(): + if _has_errno(err, errno.ENOENT, FileNotFoundError) or not self._dir_exists(path): raise - def _input_dir_exists(self) -> bool: + def _dir_exists(self, path: str) -> bool: try: - return bool(smbclient.path.isdir(self.input_dir, **self._credentials)) + return bool(smbclient.path.isdir(path, **self._credentials)) except Exception: return False diff --git a/api/tests/unit/conftest.py b/api/tests/unit/conftest.py index 42c75e2..32697eb 100644 --- a/api/tests/unit/conftest.py +++ b/api/tests/unit/conftest.py @@ -43,6 +43,7 @@ async def make_env(tmp_path: Path) -> AsyncIterator[EnvFactory]: "SCAN_HOTFOLDER_PATH": str(root), "SCAN_POLL_SECONDS": 0.01, "SCAN_TIMEOUT_SECONDS": 5, + "SCAN_SETTLE_SECONDS": 0, # R-23: набор ответа готов сразу, тесты не ждут стабилизации "PUBLISH_RETRIES": 1, **overrides, } @@ -52,7 +53,14 @@ async def make_env(tmp_path: Path) -> AsyncIterator[EnvFactory]: 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) + local = LocalHotFolder( + root, + settings.SCAN_INPUT_DIR, + settings.SCAN_OUTPUT_DIR, + settings.SCAN_TMP_DIR, + settle_seconds=settings.SCAN_SETTLE_SECONDS, + ) + hotfolder = SpyHotFolder(local, events) kwargs: dict[str, Any] = {} if clock is not None: kwargs["clock"] = clock diff --git a/api/tests/unit/test_config.py b/api/tests/unit/test_config.py index 29bb5b1..5d70a2e 100644 --- a/api/tests/unit/test_config.py +++ b/api/tests/unit/test_config.py @@ -29,6 +29,8 @@ def test_defaults_local_mode_without_smb_and_nats_credentials(monkeypatch, minim assert s.SCAN_HOTFOLDER_MODE == "local" assert s.SCAN_INPUT_DIR == "Input" assert s.SCAN_OUTPUT_DIR == "Output" + assert s.SCAN_TMP_DIR == "Tmp" + assert s.SCAN_SETTLE_SECONDS == 2 assert s.SCAN_POLL_SECONDS == 5 assert s.SCAN_TIMEOUT_SECONDS == 300 assert s.SCAN_MAX_INFLIGHT == 4 @@ -65,6 +67,19 @@ def test_smb_mode_with_smb_block_ok(monkeypatch): assert s.OPENBAO.SMB.SMB_HOST == "smb.example" +def test_settle_seconds_zero_is_allowed(monkeypatch): + monkeypatch.setenv("SCAN_SETTLE_SECONDS", "0") + + assert get_settings().SCAN_SETTLE_SECONDS == 0 + + +def test_negative_settle_seconds_rejected(monkeypatch): + monkeypatch.setenv("SCAN_SETTLE_SECONDS", "-1") + + with pytest.raises(ValidationError, match="SCAN_SETTLE_SECONDS"): + get_settings() + + def test_unknown_hotfolder_mode_rejected(monkeypatch): monkeypatch.setenv("SCAN_HOTFOLDER_MODE", "ftp") @@ -124,6 +139,8 @@ def test_env_test_file_loads_for_tests(): assert s.SCAN_HOTFOLDER_PATH == "./.hotfolder-test" assert s.SCAN_POLL_SECONDS == 0.1 assert s.SCAN_TIMEOUT_SECONDS == 5 + assert s.SCAN_TMP_DIR == "Tmp" + assert s.SCAN_SETTLE_SECONDS == 0 # R-23: тесты воркера не ждут стабилизации набора assert s.OPENBAO.MINIO.MINIO_URL == "127.0.0.1:9000" assert s.OPENBAO.MINIO.MINIO_SECURE is False assert s.OPENBAO.NATS.NATS_URL == "nats://127.0.0.1:4222" @@ -138,5 +155,7 @@ def test_env_example_loads(monkeypatch): assert s.SCAN_HOTFOLDER_MODE == "local" assert s.SCAN_INPUT_DIR == "Input" + assert s.SCAN_TMP_DIR == "Tmp" + assert s.SCAN_SETTLE_SECONDS == 2 assert s.OPENBAO.MINIO.MINIO_BUCKET == "arm-ka" assert s.OPENBAO.NATS.NATS_JWT == "" diff --git a/api/tests/unit/test_emulator.py b/api/tests/unit/test_emulator.py index 098775a..fd63a79 100644 --- a/api/tests/unit/test_emulator.py +++ b/api/tests/unit/test_emulator.py @@ -267,17 +267,20 @@ class TestProcessOnce: assert hf.answer_path(job_id).is_file() def test_interop_put_emulator_poll(self, tmp_path: Path) -> None: - """Сквозной «put -> эмулятор -> poll» без Docker (D-9).""" - hf = _hotfolder(tmp_path) + """Сквозной «put -> эмулятор -> poll» без Docker (D-9); готовность набора без ожидания (settle 0).""" + root = tmp_path / "hotfolder" + root.mkdir() + hf = LocalHotFolder(root, settle_seconds=0) fixtures = _fixtures_dir(tmp_path, {"report.json": {"file_uid": "x", "data_ocr": [{"k": "v"}]}}) job_id = _uid() assert hf.poll(job_id) is None hf.put(inbox_name(job_id, "report.pdf"), b"%PDF-1.4") assert hf.poll(job_id) is None process_once(hf.root, fixtures, 0) - raw = hf.poll(job_id) - assert raw is not None - assert json.loads(raw) == {"file_uid": job_id, "data_ocr": [{"k": "v"}]} + files = hf.poll(job_id) + assert files is not None + assert [f.name for f in files] == [f"{job_id}.json"] + assert json.loads(files[0].data) == {"file_uid": job_id, "data_ocr": [{"k": "v"}]} class TestSettings: diff --git a/api/tests/unit/test_health.py b/api/tests/unit/test_health.py index 9c95782..2517573 100644 --- a/api/tests/unit/test_health.py +++ b/api/tests/unit/test_health.py @@ -7,7 +7,7 @@ from pathlib import Path import pytest from src.core.health import HealthProvider -from src.gateway.hotfolder import LocalHotFolder +from src.gateway.hotfolder import AnswerFile, LocalHotFolder pytestmark = pytest.mark.usefixtures("log_lines") @@ -29,7 +29,7 @@ class CountingHotFolder: def put(self, name: str, data: bytes) -> None: raise NotImplementedError - def poll(self, job_id: str) -> bytes | None: + def poll(self, job_id: str) -> list[AnswerFile] | None: raise NotImplementedError def healthy(self) -> bool: diff --git a/api/tests/unit/test_hotfolder_factory.py b/api/tests/unit/test_hotfolder_factory.py index 0c9fc75..d88c7f5 100644 --- a/api/tests/unit/test_hotfolder_factory.py +++ b/api/tests/unit/test_hotfolder_factory.py @@ -16,7 +16,13 @@ API_DIR = Path(__file__).resolve().parents[2] def _local_settings(root: Path) -> Settings: return get_settings().model_copy( - update={"SCAN_HOTFOLDER_MODE": "local", "SCAN_HOTFOLDER_PATH": str(root), "SCAN_INPUT_DIR": "In"} + update={ + "SCAN_HOTFOLDER_MODE": "local", + "SCAN_HOTFOLDER_PATH": str(root), + "SCAN_INPUT_DIR": "In", + "SCAN_TMP_DIR": "Служебный", + "SCAN_SETTLE_SECONDS": 1.5, + } ) @@ -27,6 +33,8 @@ def _smb_settings() -> Settings: update={ "SCAN_HOTFOLDER_MODE": "smb", "SCAN_OUTPUT_DIR": "Out", + "SCAN_TMP_DIR": "T", + "SCAN_SETTLE_SECONDS": 3, "OPENBAO": settings.OPENBAO.model_copy(update={"SMB": smb}), } ) @@ -38,6 +46,8 @@ def test_local_mode_builds_local_hotfolder(tmp_path: Path) -> None: assert hf.root == tmp_path assert hf.input_dir == tmp_path / "In" assert hf.output_dir == tmp_path / "Output" + assert hf.tmp_dir == tmp_path / "Служебный" + assert hf.settle_seconds == 1.5 assert hf.healthy() is True @@ -49,6 +59,8 @@ def test_smb_mode_builds_smb_hotfolder() -> None: assert hf.root == "\\\\smb.example.local\\share\\scan" assert hf.input_dir == "\\\\smb.example.local\\share\\scan\\Input" assert hf.output_dir == "\\\\smb.example.local\\share\\scan\\Out" + assert hf.tmp_dir == "\\\\smb.example.local\\share\\scan\\T" + assert hf.settle_seconds == 3 def test_smb_mode_does_not_touch_share(monkeypatch: pytest.MonkeyPatch) -> None: @@ -57,7 +69,7 @@ def test_smb_mode_does_not_touch_share(monkeypatch: pytest.MonkeyPatch) -> None: def forbidden(*args: object, **kwargs: object) -> None: raise AssertionError("фабрика не должна обращаться к SMB") - for name in ("open_file", "listdir", "mkdir", "replace", "remove", "register_session"): + for name in ("open_file", "listdir", "scandir", "mkdir", "replace", "remove", "register_session"): monkeypatch.setattr(smb.smbclient, name, forbidden) create_hotfolder(_smb_settings()) diff --git a/api/tests/unit/test_local_hotfolder.py b/api/tests/unit/test_local_hotfolder.py index 80e1812..aa89ac0 100644 --- a/api/tests/unit/test_local_hotfolder.py +++ b/api/tests/unit/test_local_hotfolder.py @@ -1,27 +1,45 @@ """Горячая папка в обычном каталоге (spec image-recognition-gateway, D-1, R-4/R-5, slice-0-skeleton). -Быстрые тесты на `tmp_path`: атомарная запись входящего файла, опрос ответа ИС, проверка здоровья, валидация имён. +Быстрые тесты на `tmp_path`: атомарная запись входящего файла через служебный каталог `Tmp`, опрос набора ответов ИС +с готовностью по стабилизации (settle, часы подменены), проверка здоровья, валидация имён. """ +import os import uuid from pathlib import Path +from typing import Any import pytest -from src.gateway.hotfolder import HotFolder, LocalHotFolder, inbox_name, validate_name +from src.gateway.hotfolder import AnswerFile, HotFolder, LocalHotFolder, SettleTracker, inbox_name, validate_name def _job_id() -> str: return str(uuid.uuid4()) -def _write_answer(hf: LocalHotFolder, job_id: str, data: bytes) -> Path: - path = hf.answer_path(job_id) - path.parent.mkdir(parents=True) +class FakeClock: + """Монотонные часы теста: время двигается только вручную.""" + + def __init__(self) -> None: + self.now = 100.0 + + def __call__(self) -> float: + return self.now + + +def _write_answer(hf: LocalHotFolder, job_id: str, data: bytes, name: str | None = None) -> Path: + path = hf.answer_dir(job_id) / (name or f"{job_id}.json") + path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data) return path +def _settled(tmp_path: Path, settle: float = 2) -> tuple[LocalHotFolder, FakeClock]: + clock = FakeClock() + return LocalHotFolder(tmp_path, settle_seconds=settle, clock=clock), clock + + class TestInboxName: def test_joins_job_id_and_name(self) -> None: job_id = _job_id() @@ -155,19 +173,31 @@ class TestPaths: assert hf.root == tmp_path assert hf.input_dir == tmp_path / "Input" assert hf.output_dir == tmp_path / "Output" + assert hf.tmp_dir == tmp_path / "Tmp" + assert hf.answer_dir(job_id) == tmp_path / "Output" / job_id assert hf.answer_path(job_id) == tmp_path / "Output" / job_id / f"{job_id}.json" def test_custom_dir_names_and_str_root(self, tmp_path: Path) -> None: - hf = LocalHotFolder(str(tmp_path), input_dir="Входящие", output_dir="Исходящие") + hf = LocalHotFolder(str(tmp_path), input_dir="Входящие", output_dir="Исходящие", tmp_dir="Служебный") assert hf.root == tmp_path assert hf.input_dir == tmp_path / "Входящие" assert hf.output_dir == tmp_path / "Исходящие" + assert hf.tmp_dir == tmp_path / "Служебный" assert hf.answer_path("abc") == tmp_path / "Исходящие" / "abc" / "abc.json" @pytest.mark.parametrize("job_id", ["", ".", "..", "../x", "a/b", "a\\b", "a:b", "a?b", "a*b", "a\x00b", "a\nb"]) def test_answer_path_rejects_unsafe_job_id(self, tmp_path: Path, job_id: str) -> None: + hf = LocalHotFolder(tmp_path) with pytest.raises(ValueError, match="job_id"): - LocalHotFolder(tmp_path).answer_path(job_id) + hf.answer_path(job_id) + with pytest.raises(ValueError, match="job_id"): + hf.answer_dir(job_id) + with pytest.raises(ValueError, match="job_id"): + hf.poll(job_id) + + def test_negative_settle_is_rejected(self, tmp_path: Path) -> None: + with pytest.raises(ValueError, match="settle"): + LocalHotFolder(tmp_path, settle_seconds=-1) def test_implements_protocol(self, tmp_path: Path) -> None: hf: HotFolder = LocalHotFolder(tmp_path) @@ -187,14 +217,54 @@ class TestPut: name = inbox_name(_job_id(), "aep_5.pdf") hf.put(name, b"data") assert [p.name for p in hf.input_dir.iterdir()] == [name] + assert list(hf.tmp_dir.iterdir()) == [] - def test_creates_input_dir_only(self, tmp_path: Path) -> None: + def test_part_is_written_in_tmp_not_in_input(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """OQ-79: во время записи во входящем каталоге нет временного файла — он в `Tmp/`, затем rename в `Input`.""" + hf = LocalHotFolder(tmp_path) + name = inbox_name(_job_id(), "aep_5.pdf") + seen: dict[str, Any] = {} + real_replace = os.replace + + def spy(src: Any, dst: Any) -> None: + seen["src"], seen["dst"] = Path(src), Path(dst) + seen["input"] = sorted(p.name for p in hf.input_dir.iterdir()) + seen["tmp"] = sorted(p.name for p in hf.tmp_dir.iterdir()) + seen["part"] = Path(src).read_bytes() + real_replace(src, dst) + + monkeypatch.setattr("src.gateway.hotfolder.os.replace", spy) + hf.put(name, b"data") + assert seen == { + "src": hf.tmp_dir / f".{name}.part", + "dst": hf.input_dir / name, + "input": [], + "tmp": [f".{name}.part"], + "part": b"data", + } + assert [p.name for p in hf.input_dir.iterdir()] == [name] + assert list(hf.tmp_dir.iterdir()) == [] + + def test_creates_input_and_tmp_dirs_only(self, tmp_path: Path) -> None: hf = LocalHotFolder(tmp_path) assert not hf.input_dir.exists() + assert not hf.tmp_dir.exists() hf.put("a.pdf", b"1") assert hf.input_dir.is_dir() + assert hf.tmp_dir.is_dir() assert not hf.output_dir.exists() + def test_existing_tmp_dir_is_fine(self, tmp_path: Path) -> None: + (tmp_path / "Tmp").mkdir() + LocalHotFolder(tmp_path).put("a.pdf", b"1") + assert (tmp_path / "Input" / "a.pdf").read_bytes() == b"1" + + def test_custom_tmp_dir_name(self, tmp_path: Path) -> None: + hf = LocalHotFolder(tmp_path, tmp_dir="Служебный") + hf.put("a.pdf", b"1") + assert (tmp_path / "Служебный").is_dir() + assert not (tmp_path / "Tmp").exists() + def test_existing_input_dir_is_fine(self, tmp_path: Path) -> None: (tmp_path / "Input").mkdir() LocalHotFolder(tmp_path).put("a.pdf", b"1") @@ -259,6 +329,7 @@ class TestPut: with pytest.raises(OSError, match="диск недоступен"): hf.put("a.pdf", b"data") assert list(hf.input_dir.iterdir()) == [] + assert list(hf.tmp_dir.iterdir()) == [] def test_cleanup_failure_does_not_mask_original_error( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch @@ -279,8 +350,10 @@ class TestPut: class TestPoll: - def test_none_until_answer_exists(self, tmp_path: Path) -> None: - hf = LocalHotFolder(tmp_path) + """`poll` — все `*.json` каталога `//` по имени, когда набор настоялся `settle_seconds`.""" + + def test_none_until_answer_dir_has_json(self, tmp_path: Path) -> None: + hf = LocalHotFolder(tmp_path, settle_seconds=0) job_id = _job_id() assert hf.poll(job_id) is None (tmp_path / "Output").mkdir() @@ -288,28 +361,160 @@ class TestPoll: (tmp_path / "Output" / job_id).mkdir() assert hf.poll(job_id) is None - def test_returns_bytes_when_answer_appears(self, tmp_path: Path) -> None: - hf = LocalHotFolder(tmp_path) + def test_missing_root_is_none(self, tmp_path: Path) -> None: + assert LocalHotFolder(tmp_path / "absent", settle_seconds=0).poll(_job_id()) is None + + def test_answer_dir_is_a_file_is_none(self, tmp_path: Path) -> None: + hf = LocalHotFolder(tmp_path, settle_seconds=0) job_id = _job_id() - payload = b'{"file_uid": "x", "data_ocr": []}' + (tmp_path / "Output").mkdir() + (tmp_path / "Output" / job_id).write_bytes(b"{}") assert hf.poll(job_id) is None + + def test_settle_zero_returns_at_once(self, tmp_path: Path) -> None: + hf = LocalHotFolder(tmp_path, settle_seconds=0) + job_id = _job_id() + payload = b'{"file_uid": "x", "data_ocr": {}}' _write_answer(hf, job_id, payload) - assert hf.poll(job_id) == payload + assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", payload)] + + def test_one_json_after_settle(self, tmp_path: Path) -> None: + hf, clock = _settled(tmp_path) + job_id = _job_id() + _write_answer(hf, job_id, b"{}") + assert hf.poll(job_id) is None # набор впервые увиден + clock.now += 1.9 + assert hf.poll(job_id) is None + clock.now += 0.1 + assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b"{}")] + + def test_two_jsons_ascending_by_name(self, tmp_path: Path) -> None: + hf, clock = _settled(tmp_path) + job_id = _job_id() + _write_answer(hf, job_id, b'{"n": 2}', f"{job_id}_ofr.json") + _write_answer(hf, job_id, b'{"n": 1}', f"{job_id}.json") + assert hf.poll(job_id) is None + clock.now += 2 + assert hf.poll(job_id) == [ + AnswerFile(f"{job_id}.json", b'{"n": 1}'), + AnswerFile(f"{job_id}_ofr.json", b'{"n": 2}'), + ] + + def test_non_json_files_and_subdirs_are_ignored(self, tmp_path: Path) -> None: + hf, clock = _settled(tmp_path) + job_id = _job_id() + _write_answer(hf, job_id, b"text", "readme.txt") + _write_answer(hf, job_id, b"{}", f".{job_id}.json.part") + (hf.answer_dir(job_id) / "nested.json").mkdir() + assert hf.poll(job_id) is None + clock.now += 10 + assert hf.poll(job_id) is None # json-файлов нет — ответа нет, сколько ни жди + _write_answer(hf, job_id, b"{}") + assert hf.poll(job_id) is None + clock.now += 2 + assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b"{}")] + + def test_json_extension_is_case_insensitive(self, tmp_path: Path) -> None: + hf = LocalHotFolder(tmp_path, settle_seconds=0) + job_id = _job_id() + _write_answer(hf, job_id, b"{}", "B.JSON") + _write_answer(hf, job_id, b"[]", "a.Json") + assert hf.poll(job_id) == [AnswerFile("B.JSON", b"{}"), AnswerFile("a.Json", b"[]")] + + def test_growing_file_is_not_ready_until_stable(self, tmp_path: Path) -> None: + hf, clock = _settled(tmp_path) + job_id = _job_id() + path = _write_answer(hf, job_id, b'{"part": ') + assert hf.poll(job_id) is None + clock.now += 1.5 + path.write_bytes(b'{"part": 1, "more": ') # размер вырос — таймер заново + assert hf.poll(job_id) is None + clock.now += 1.5 + path.write_bytes(b'{"part": 1, "more": 2}') + assert hf.poll(job_id) is None + clock.now += 1.9 + assert hf.poll(job_id) is None + clock.now += 0.1 + assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b'{"part": 1, "more": 2}')] + + def test_new_file_resets_timer(self, tmp_path: Path) -> None: + hf, clock = _settled(tmp_path) + job_id = _job_id() + _write_answer(hf, job_id, b"{}") + assert hf.poll(job_id) is None + clock.now += 1.5 + _write_answer(hf, job_id, b"[]", f"{job_id}_ofr.json") + assert hf.poll(job_id) is None + clock.now += 1.5 + assert hf.poll(job_id) is None # с момента появления второго файла прошло 1,5 с + clock.now += 0.5 + assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b"{}"), AnswerFile(f"{job_id}_ofr.json", b"[]")] + + def test_removed_file_resets_timer(self, tmp_path: Path) -> None: + hf, clock = _settled(tmp_path) + job_id = _job_id() + _write_answer(hf, job_id, b"{}") + extra = _write_answer(hf, job_id, b"[]", f"{job_id}_ofr.json") + assert hf.poll(job_id) is None + clock.now += 1.5 + extra.unlink() + assert hf.poll(job_id) is None + clock.now += 1.5 + assert hf.poll(job_id) is None + clock.now += 0.5 + assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b"{}")] + + def test_removal_of_all_json_is_none_and_restarts(self, tmp_path: Path) -> None: + hf, clock = _settled(tmp_path) + job_id = _job_id() + path = _write_answer(hf, job_id, b"{}") + assert hf.poll(job_id) is None + clock.now += 1 + path.unlink() + assert hf.poll(job_id) is None + clock.now += 5 + path.write_bytes(b"{}") # тот же набор снова: настаивается заново, а не «с прошлого раза» + assert hf.poll(job_id) is None + clock.now += 2 + assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b"{}")] + + def test_state_is_dropped_after_answer(self, tmp_path: Path) -> None: + hf, clock = _settled(tmp_path) + job_id = _job_id() + _write_answer(hf, job_id, b"{}") + assert hf.poll(job_id) is None + clock.now += 2 + assert hf.poll(job_id) is not None + assert hf.pending_jobs() == 0 + assert hf.poll(job_id) is None # состояние сброшено: повторный опрос настаивает набор заново def test_other_job_is_unaffected(self, tmp_path: Path) -> None: - hf = LocalHotFolder(tmp_path) + hf, clock = _settled(tmp_path) job_id, other = _job_id(), _job_id() _write_answer(hf, other, b"{}") + assert hf.poll(other) is None + clock.now += 1 + _write_answer(hf, job_id, b"[]") assert hf.poll(job_id) is None - assert hf.poll(other) == b"{}" + clock.now += 1 + assert hf.poll(job_id) is None + assert hf.poll(other) == [AnswerFile(f"{other}.json", b"{}")] + clock.now += 1 + assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b"[]")] - def test_missing_root_is_none(self, tmp_path: Path) -> None: - assert LocalHotFolder(tmp_path / "absent").poll(_job_id()) is None - - def test_file_removed_between_exists_and_read_is_none( + def test_file_changed_between_listing_and_read_is_none( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: - hf = LocalHotFolder(tmp_path) + hf = LocalHotFolder(tmp_path, settle_seconds=0) + job_id = _job_id() + _write_answer(hf, job_id, b"{}") + monkeypatch.setattr(Path, "read_bytes", lambda self: b'{"longer": true}') + assert hf.poll(job_id) is None + + def test_file_removed_between_listing_and_read_is_none( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + hf = LocalHotFolder(tmp_path, settle_seconds=0) job_id = _job_id() path = _write_answer(hf, job_id, b"{}") @@ -321,6 +526,48 @@ class TestPoll: assert hf.poll(job_id) is None +class TestSettleTracker: + def test_snapshot_must_be_stable_for_settle_seconds(self) -> None: + clock = FakeClock() + tracker = SettleTracker(2, clock) + snap = (("a.json", 2),) + assert tracker.ready("job", snap) is False + clock.now += 2 + assert tracker.ready("job", snap) is True + assert len(tracker) == 0 + + def test_empty_snapshot_is_never_ready(self) -> None: + assert SettleTracker(0).ready("job", ()) is False + + def test_zero_settle_keeps_no_state(self) -> None: + tracker = SettleTracker(0) + assert tracker.ready("job", (("a.json", 2),)) is True + assert len(tracker) == 0 + + def test_state_is_bounded(self) -> None: + """Задания, ушедшие по таймауту с ненастоявшимся набором, не копятся: старые вытесняются.""" + clock = FakeClock() + tracker = SettleTracker(2, clock, max_jobs=3) + for job in ("j1", "j2", "j3", "j4"): + assert tracker.ready(job, (("a.json", 1),)) is False + assert len(tracker) == 3 + clock.now += 2 + assert tracker.ready("j1", (("a.json", 1),)) is False # вытеснено — настаивается заново + assert tracker.ready("j4", (("a.json", 1),)) is True + + def test_forget(self) -> None: + clock = FakeClock() + tracker = SettleTracker(2, clock) + tracker.ready("job", (("a.json", 1),)) + tracker.forget("job") + tracker.forget("unknown") + assert len(tracker) == 0 + + def test_negative_settle_is_rejected(self) -> None: + with pytest.raises(ValueError, match="settle"): + SettleTracker(-0.1) + + class TestHealthy: def test_true_for_existing_dir(self, tmp_path: Path) -> None: assert LocalHotFolder(tmp_path).healthy() is True diff --git a/api/tests/unit/test_smb_hotfolder.py b/api/tests/unit/test_smb_hotfolder.py index ba9c70e..e814f0f 100644 --- a/api/tests/unit/test_smb_hotfolder.py +++ b/api/tests/unit/test_smb_hotfolder.py @@ -14,13 +14,22 @@ import pytest from smbprotocol.exceptions import NtStatus, SMBException, SMBOSError from src.gateway import smb -from src.gateway.hotfolder import HotFolder, inbox_name +from src.gateway.hotfolder import AnswerFile, HotFolder, inbox_name from src.gateway.smb import SmbHotFolder HOST = "smb.example.local" ROOT = "\\\\smb.example.local\\share\\scan" INPUT = ROOT + "\\Input" OUTPUT = ROOT + "\\Output" +TMP = ROOT + "\\Tmp" + + +class FakeClock: + def __init__(self) -> None: + self.now = 100.0 + + def __call__(self) -> float: + return self.now class FakeSmbClient: @@ -78,6 +87,25 @@ class FakeSmbClient: raise SMBOSError(NtStatus.STATUS_OBJECT_NAME_NOT_FOUND, path) local.unlink() + def scandir(self, path: str, **kwargs: Any) -> list[SimpleNamespace]: + """Как `smbclient.scandir`: элементы с `name`, `is_file()` и `smb_info.end_of_file` (размер из листинга).""" + self._record("scandir", (path,), kwargs) + local = self._local(path) + if not local.parent.is_dir(): + raise SMBOSError(NtStatus.STATUS_OBJECT_PATH_NOT_FOUND, path) + if not local.exists(): + raise SMBOSError(NtStatus.STATUS_OBJECT_NAME_NOT_FOUND, path) + if not local.is_dir(): + raise SMBOSError(NtStatus.STATUS_NOT_A_DIRECTORY, path) + return [ + SimpleNamespace( + name=p.name, + is_file=p.is_file, + smb_info=SimpleNamespace(end_of_file=p.stat().st_size if p.is_file() else 0), + ) + for p in local.iterdir() + ] + def listdir(self, path: str, **kwargs: Any) -> list[str]: self._record("listdir", (path,), kwargs) local = self._local(path) @@ -111,13 +139,16 @@ def _job_id() -> str: def _hotfolder(**kwargs: Any) -> SmbHotFolder: + kwargs.setdefault("settle_seconds", 0) return SmbHotFolder(HOST, "share/scan", **kwargs) -def _write_answer(share_root: Path, job_id: str, data: bytes) -> None: +def _write_answer(share_root: Path, job_id: str, data: bytes, name: str | None = None) -> Path: folder = share_root / "Output" / job_id - folder.mkdir(parents=True) - (folder / f"{job_id}.json").write_bytes(data) + folder.mkdir(parents=True, exist_ok=True) + path = folder / (name or f"{job_id}.json") + path.write_bytes(data) + return path class TestPaths: @@ -127,6 +158,8 @@ class TestPaths: assert hf.root == ROOT assert hf.input_dir == INPUT assert hf.output_dir == OUTPUT + assert hf.tmp_dir == TMP + assert hf.answer_dir(job_id) == f"{OUTPUT}\\{job_id}" assert hf.answer_path(job_id) == f"{OUTPUT}\\{job_id}\\{job_id}.json" @pytest.mark.parametrize("path", ["share/scan", "share\\scan", "/share/scan/", "\\share\\scan\\"]) @@ -141,9 +174,10 @@ class TestPaths: assert SmbHotFolder(HOST, "share").root == "\\\\smb.example.local\\share" def test_custom_dir_names(self) -> None: - hf = SmbHotFolder(HOST, "share/scan", input_dir="Входящие", output_dir="Исходящие") + hf = SmbHotFolder(HOST, "share/scan", input_dir="Входящие", output_dir="Исходящие", tmp_dir="Служебный") assert hf.input_dir == ROOT + "\\Входящие" assert hf.output_dir == ROOT + "\\Исходящие" + assert hf.tmp_dir == ROOT + "\\Служебный" assert hf.answer_path("abc") == ROOT + "\\Исходящие\\abc\\abc.json" @pytest.mark.parametrize(("host", "path"), [(HOST, ""), (HOST, "/"), ("", "share/scan"), ("", "\\\\srv")]) @@ -155,6 +189,12 @@ class TestPaths: def test_answer_path_rejects_unsafe_job_id(self, job_id: str) -> None: with pytest.raises(ValueError, match="job_id"): _hotfolder().answer_path(job_id) + with pytest.raises(ValueError, match="job_id"): + _hotfolder().answer_dir(job_id) + + def test_negative_settle_is_rejected(self) -> None: + with pytest.raises(ValueError, match="settle"): + _hotfolder(settle_seconds=-1) def test_constructor_does_not_touch_share(self, fake: FakeSmbClient) -> None: _hotfolder(login="user", password="secret") @@ -166,18 +206,63 @@ class TestPaths: class TestPut: - def test_writes_part_then_replaces(self, fake: FakeSmbClient, share_root: Path) -> None: + def test_writes_part_in_tmp_then_replaces_into_input(self, fake: FakeSmbClient, share_root: Path) -> None: name = inbox_name(_job_id(), "aep_5.PDF") data = bytes(range(256)) * 10 _hotfolder().put(name, data) - part = f"{INPUT}\\.{name}.part" + part = f"{TMP}\\.{name}.part" assert fake.calls == [ + ("mkdir", (TMP,), {}), ("mkdir", (INPUT,), {}), ("open_file", (part, "wb"), {}), ("replace", (part, f"{INPUT}\\{name}"), {}), ] assert (share_root / "Input" / name).read_bytes() == data assert [p.name for p in (share_root / "Input").iterdir()] == [name] + assert list((share_root / "Tmp").iterdir()) == [] + + def test_input_has_no_part_during_write(self, fake: FakeSmbClient, share_root: Path) -> None: + """OQ-79: пока пишется временный файл, во входящем каталоге его нет — он в `Tmp`.""" + name = "a.pdf" + seen: dict[str, list[str]] = {} + real_replace = fake.replace + + def spy(src: str, dst: str, **kwargs: Any) -> None: + seen["input"] = sorted(p.name for p in (share_root / "Input").iterdir()) + seen["tmp"] = sorted(p.name for p in (share_root / "Tmp").iterdir()) + real_replace(src, dst, **kwargs) + + fake.replace = spy # type: ignore[method-assign] + _hotfolder().put(name, b"data") + assert seen == {"input": [], "tmp": [".a.pdf.part"]} + assert [p.name for p in (share_root / "Input").iterdir()] == [name] + + def test_existing_tmp_dir_is_fine(self, fake: FakeSmbClient, share_root: Path) -> None: + (share_root / "Tmp").mkdir() + _hotfolder().put("a.pdf", b"1") + assert (share_root / "Input" / "a.pdf").read_bytes() == b"1" + + def test_tmp_mkdir_denied_but_tmp_exists_is_fine(self, fake: FakeSmbClient, share_root: Path) -> None: + (share_root / "Tmp").mkdir() + (share_root / "Input").mkdir() + fake.fail["mkdir"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, TMP) + _hotfolder().put("a.pdf", b"1") + assert (share_root / "Input" / "a.pdf").read_bytes() == b"1" + assert fake.calls[:4] == [ + ("mkdir", (TMP,), {}), + ("isdir", (TMP,), {}), + ("mkdir", (INPUT,), {}), + ("isdir", (INPUT,), {}), + ] + + def test_tmp_mkdir_denied_and_no_tmp_raises_original(self, fake: FakeSmbClient, share_root: Path) -> None: + (share_root / "Input").mkdir() + denied = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, TMP) + fake.fail["mkdir"] = denied + with pytest.raises(SMBOSError) as excinfo: + _hotfolder().put("a.pdf", b"1") + assert excinfo.value is denied + assert fake.funcs() == ["mkdir", "isdir"] def test_existing_input_dir_is_fine(self, fake: FakeSmbClient, share_root: Path) -> None: (share_root / "Input").mkdir() @@ -193,25 +278,27 @@ class TestPut: def test_mkdir_denied_but_input_exists_is_fine(self, fake: FakeSmbClient, share_root: Path) -> None: """Урезанные ACL: нельзя создавать подкаталоги в корне, но `Input` уже есть и в него писать можно.""" + (share_root / "Tmp").mkdir() (share_root / "Input").mkdir() fake.fail["mkdir"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT) _hotfolder().put("a.pdf", b"1") assert (share_root / "Input" / "a.pdf").read_bytes() == b"1" - assert fake.funcs() == ["mkdir", "isdir", "open_file", "replace"] - assert fake.calls[1] == ("isdir", (INPUT,), {}) + assert fake.funcs() == ["mkdir", "isdir", "mkdir", "isdir", "open_file", "replace"] + assert fake.calls[3] == ("isdir", (INPUT,), {}) def test_mkdir_denied_and_no_input_raises_original(self, fake: FakeSmbClient, share_root: Path) -> None: + (share_root / "Tmp").mkdir() denied = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT) fake.fail["mkdir"] = denied with pytest.raises(SMBOSError) as excinfo: _hotfolder().put("a.pdf", b"1") assert excinfo.value is denied - assert fake.funcs() == ["mkdir", "isdir"] + assert fake.funcs() == ["mkdir", "isdir", "mkdir", "isdir"] def test_mkdir_denied_and_isdir_fails_raises_original(self, fake: FakeSmbClient, share_root: Path) -> None: - denied = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT) + denied = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, TMP) fake.fail["mkdir"] = denied - fake.fail["isdir"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT) + fake.fail["isdir"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, TMP) with pytest.raises(SMBOSError) as excinfo: _hotfolder().put("a.pdf", b"1") assert excinfo.value is denied @@ -266,56 +353,111 @@ class TestPut: fake.fail["replace"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT) with pytest.raises(OSError): _hotfolder().put("a.pdf", b"data") - part = f"{INPUT}\\.a.pdf.part" + part = f"{TMP}\\.a.pdf.part" assert fake.calls[-1] == ("remove", (part,), {}) assert list((share_root / "Input").iterdir()) == [] + assert list((share_root / "Tmp").iterdir()) == [] def test_cleanup_error_does_not_mask_original(self, fake: FakeSmbClient, share_root: Path) -> None: fake.fail["replace"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT) fake.fail["remove"] = SMBException("соединение потеряно") with pytest.raises(SMBOSError): _hotfolder().put("a.pdf", b"data") - assert fake.funcs() == ["mkdir", "open_file", "replace", "remove"] + assert fake.funcs() == ["mkdir", "mkdir", "open_file", "replace", "remove"] def test_passes_credentials_to_every_call(self, fake: FakeSmbClient, share_root: Path) -> None: _hotfolder(login="user", password="secret").put("a.pdf", b"1") - assert [kwargs for _, _, kwargs in fake.calls] == [{"username": "user", "password": "secret"}] * 3 + assert [kwargs for _, _, kwargs in fake.calls] == [{"username": "user", "password": "secret"}] * 4 class TestPoll: - def test_none_until_answer_exists(self, fake: FakeSmbClient, share_root: Path) -> None: + def test_none_until_answer_dir_has_json(self, fake: FakeSmbClient, share_root: Path) -> None: hf = _hotfolder() job_id = _job_id() assert hf.poll(job_id) is None # нет Output: STATUS_OBJECT_PATH_NOT_FOUND (share_root / "Output").mkdir() - assert hf.poll(job_id) is None # нет Output/ + assert hf.poll(job_id) is None # нет Output/: STATUS_OBJECT_NAME_NOT_FOUND (share_root / "Output" / job_id).mkdir() - assert hf.poll(job_id) is None # нет .json: STATUS_OBJECT_NAME_NOT_FOUND + assert hf.poll(job_id) is None # каталог пуст + assert fake.funcs() == ["scandir", "scandir", "scandir"] - def test_returns_bytes_when_answer_appears(self, fake: FakeSmbClient, share_root: Path) -> None: + def test_lists_answer_dir_and_reads_every_json(self, fake: FakeSmbClient, share_root: Path) -> None: hf = _hotfolder() job_id = _job_id() - payload = b'{"file_uid": "x", "data_ocr": []}' - _write_answer(share_root, job_id, payload) - assert hf.poll(job_id) == payload - assert fake.calls == [("open_file", (f"{OUTPUT}\\{job_id}\\{job_id}.json", "rb"), {})] + _write_answer(share_root, job_id, b'{"n": 2}', f"{job_id}_ofr.json") + _write_answer(share_root, job_id, b'{"n": 1}') + _write_answer(share_root, job_id, b"text", "readme.txt") + (share_root / "Output" / job_id / "nested.json").mkdir() + assert hf.poll(job_id) == [ + AnswerFile(f"{job_id}.json", b'{"n": 1}'), + AnswerFile(f"{job_id}_ofr.json", b'{"n": 2}'), + ] + folder = f"{OUTPUT}\\{job_id}" + assert fake.calls == [ + ("scandir", (folder,), {}), + ("open_file", (f"{folder}\\{job_id}.json", "rb"), {}), + ("open_file", (f"{folder}\\{job_id}_ofr.json", "rb"), {}), + ] + + def test_one_json_after_settle(self, fake: FakeSmbClient, share_root: Path) -> None: + clock = FakeClock() + hf = _hotfolder(settle_seconds=2, clock=clock) + job_id = _job_id() + _write_answer(share_root, job_id, b"{}") + assert hf.poll(job_id) is None + clock.now += 1.9 + assert hf.poll(job_id) is None + clock.now += 0.1 + assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b"{}")] + assert fake.funcs() == ["scandir", "scandir", "scandir", "open_file"] # до готовности файлы не читаются + + def test_growing_file_is_not_ready_until_stable(self, fake: FakeSmbClient, share_root: Path) -> None: + clock = FakeClock() + hf = _hotfolder(settle_seconds=2, clock=clock) + job_id = _job_id() + path = _write_answer(share_root, job_id, b'{"a": ') + assert hf.poll(job_id) is None + clock.now += 1.5 + path.write_bytes(b'{"a": 1}') + assert hf.poll(job_id) is None + clock.now += 1.5 + assert hf.poll(job_id) is None + clock.now += 0.5 + assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b'{"a": 1}')] def test_reads_only_own_answer(self, fake: FakeSmbClient, share_root: Path) -> None: job_id, other = _job_id(), _job_id() _write_answer(share_root, other, b"{}") - (share_root / "Output" / other / "extra.json").write_bytes(b"[]") hf = _hotfolder() assert hf.poll(job_id) is None - assert hf.poll(other) == b"{}" - assert fake.funcs() == ["open_file", "open_file"] + assert hf.poll(other) == [AnswerFile(f"{other}.json", b"{}")] def test_missing_root_is_none(self, fake: FakeSmbClient) -> None: assert _hotfolder().poll(_job_id()) is None def test_plain_file_not_found_is_none(self, fake: FakeSmbClient) -> None: - fake.fail["open_file"] = FileNotFoundError("нет файла") + fake.fail["scandir"] = FileNotFoundError("нет каталога") assert _hotfolder().poll(_job_id()) is None + def test_file_vanished_before_read_is_none(self, fake: FakeSmbClient, share_root: Path) -> None: + job_id = _job_id() + _write_answer(share_root, job_id, b"{}") + fake.fail["open_file"] = SMBOSError(NtStatus.STATUS_OBJECT_NAME_NOT_FOUND, OUTPUT) + assert _hotfolder().poll(job_id) is None + + def test_file_changed_between_listing_and_read_is_none(self, fake: FakeSmbClient, share_root: Path) -> None: + job_id = _job_id() + path = _write_answer(share_root, job_id, b"{}") + real_scandir = fake.scandir + + def scandir_then_grow(folder: str, **kwargs: Any) -> list[SimpleNamespace]: + entries = real_scandir(folder, **kwargs) + path.write_bytes(b'{"more": 1}') + return entries + + fake.scandir = scandir_then_grow # type: ignore[method-assign] + assert _hotfolder().poll(job_id) is None + @pytest.mark.parametrize("status", [NtStatus.STATUS_SHARING_VIOLATION, NtStatus.STATUS_DELETE_PENDING]) def test_answer_being_written_or_deleted_is_none( self, fake: FakeSmbClient, share_root: Path, status: NtStatus @@ -326,10 +468,13 @@ class TestPoll: fake.fail["open_file"] = SMBOSError(status, OUTPUT) assert _hotfolder().poll(job_id) is None - def test_other_errors_propagate(self, fake: FakeSmbClient, share_root: Path) -> None: - fake.fail["open_file"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, OUTPUT) + @pytest.mark.parametrize("func", ["scandir", "open_file"]) + def test_other_errors_propagate(self, fake: FakeSmbClient, share_root: Path, func: str) -> None: + job_id = _job_id() + _write_answer(share_root, job_id, b"{}") + fake.fail[func] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, OUTPUT) with pytest.raises(SMBOSError): - _hotfolder().poll(_job_id()) + _hotfolder().poll(job_id) def test_invalid_job_id_rejected_before_any_call(self, fake: FakeSmbClient) -> None: with pytest.raises(ValueError, match="job_id"): @@ -337,8 +482,10 @@ class TestPoll: assert fake.calls == [] def test_passes_credentials(self, fake: FakeSmbClient, share_root: Path) -> None: - _hotfolder(login="user", password="secret").poll(_job_id()) - assert fake.calls[0][2] == {"username": "user", "password": "secret"} + job_id = _job_id() + _write_answer(share_root, job_id, b"{}") + _hotfolder(login="user", password="secret").poll(job_id) + assert [kwargs for _, _, kwargs in fake.calls] == [{"username": "user", "password": "secret"}] * 2 class TestHealthy: diff --git a/api/tests/unit/test_worker.py b/api/tests/unit/test_worker.py index 7ea88fb..ce631ed 100644 --- a/api/tests/unit/test_worker.py +++ b/api/tests/unit/test_worker.py @@ -24,6 +24,7 @@ from src.contract import ( ) from src.core.clients import JSON_CONTENT_TYPE, OCTET_STREAM from src.core.worker import MAX_ERROR_MESSAGE_CHARS +from src.gateway.hotfolder import AnswerFile from src.normalizer import NOT_IMPLEMENTED_MESSAGE, NormalizeOutcome from tests.worker_helpers import ( API_DIR, @@ -286,12 +287,22 @@ 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] + env.hotfolder.poll_script = [None, [AnswerFile("x.json", full[:10])], [AnswerFile("x.json", 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_several_answer_files_first_by_name_is_used(make_env: EnvFactory) -> None: + """Переходно (Task 3 заменит склейкой всех файлов): из готового набора берётся первый файл (по имени).""" + env = await make_env() + first = json.dumps({**SUCCESS_ANSWER, "file_uid": "x"}).encode() + env.hotfolder.poll_script = [[AnswerFile("x.json", first), AnswerFile("x_ofr.json", b'{"file_uid": "second"}')]] + 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")] == first + + # --- таймаут ------------------------------------------------------------------------------------------------------ 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) diff --git a/api/tests/worker_helpers.py b/api/tests/worker_helpers.py index 766ef8c..502fe44 100644 --- a/api/tests/worker_helpers.py +++ b/api/tests/worker_helpers.py @@ -26,7 +26,7 @@ from src.core.clients import JSON_CONTENT_TYPE, OCTET_STREAM, S3ObjectNotFoundEr from src.core.config import Settings from src.core.nats import NatsGateway from src.core.worker import Worker -from src.gateway.hotfolder import LocalHotFolder +from src.gateway.hotfolder import AnswerFile, LocalHotFolder from src.scan_emulator import process_once API_DIR = Path(__file__).resolve().parents[1] @@ -135,14 +135,14 @@ class SpyHotFolder: self.polls = 0 self.poll_errors: list[BaseException] = [] self.poll_always: BaseException | None = None - self.poll_script: list[bytes | None] = [] + self.poll_script: list[list[AnswerFile] | 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: + def poll(self, job_id: str) -> list[AnswerFile] | None: self.polls += 1 if self.poll_always is not None: raise self.poll_always @@ -240,8 +240,8 @@ class Env: 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) + def answer(self, job_id: str, content: bytes, name: str | None = None) -> None: + path = self.local.answer_dir(job_id) / (name or f"{job_id}.json") path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(content)