From 7dc0317b99cae819541678b6e35c05050a9ffe2d Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Fri, 9 Oct 2026 16:34:39 +0400 Subject: [PATCH] =?UTF-8?q?feat(hotfolder):=20local=20=D1=82=D0=BE=D0=B6?= =?UTF-8?q?=D0=B5=20=D0=BF=D0=B8=D1=88=D0=B5=D1=82=20=D0=BF=D1=80=D1=8F?= =?UTF-8?q?=D0=BC=D0=BE=20=D0=B2=20Input;=20=D0=B1=D0=B5=D0=B7=20SCAN=5FTM?= =?UTF-8?q?P=5FDIR;=20=D1=82=D0=B0=D0=B9=D0=BC=D0=B0=D1=83=D1=82=201200=20?= =?UTF-8?q?=D1=81=20(#26)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- api/src/core/config.py | 6 +- api/src/gateway/__init__.py | 1 - api/src/gateway/hotfolder.py | 86 +----- api/tests/integration/test_roundtrip.py | 21 +- api/tests/unit/conftest.py | 1 - api/tests/unit/test_config.py | 18 +- api/tests/unit/test_hotfolder_factory.py | 3 +- api/tests/unit/test_local_hotfolder.py | 344 ++++------------------- api/tests/unit/test_worker.py | 5 +- 9 files changed, 87 insertions(+), 398 deletions(-) diff --git a/api/src/core/config.py b/api/src/core/config.py index b14165f..db9eb68 100644 --- a/api/src/core/config.py +++ b/api/src/core/config.py @@ -83,13 +83,13 @@ 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 запроса) + # Потолок ожидания ответа ИС (дополнительно к reply_deadline запроса): 20 мин, в паре со сроком ответа OCR ядра + SCAN_TIMEOUT_SECONDS: PositiveFloat = 1200 SCAN_MAX_INFLIGHT: PositiveInt = 4 # одновременно обрабатываемых заданий - # Очистка горячей папки после прочтения набора ответа (D-2): любое удаление, включая .part в Tmp, только при true. + # Очистка горячей папки после прочтения набора ответа (D-2): ответ, входящий файл, .done — только при true. # На SMB-шаре контура права на удаление не подтверждены — по умолчанию выключено SCAN_CLEANUP: bool = False SCAN_RULES_DIR: str = "rules" # справочники нормализатора (yaml), путь от текущего каталога (api/) diff --git a/api/src/gateway/__init__.py b/api/src/gateway/__init__.py index 5d5265e..fb726df 100644 --- a/api/src/gateway/__init__.py +++ b/api/src/gateway/__init__.py @@ -18,7 +18,6 @@ def create_hotfolder(settings: Settings) -> HotFolder: settings.SCAN_HOTFOLDER_PATH, settings.SCAN_INPUT_DIR, settings.SCAN_OUTPUT_DIR, - settings.SCAN_TMP_DIR, settle_seconds=settings.SCAN_SETTLE_SECONDS, cleanup=settings.SCAN_CLEANUP, ) diff --git a/api/src/gateway/hotfolder.py b/api/src/gateway/hotfolder.py index 145d910..e1a2a01 100644 --- a/api/src/gateway/hotfolder.py +++ b/api/src/gateway/hotfolder.py @@ -1,17 +1,16 @@ """Горячие папки ИС Сканирование: протокол `HotFolder` и реализация в обычном каталоге (D-1, D-2, R-4, R-5). -Файл кладётся во входящий каталог как `.` (D-1): сначала временный `.<имя>.part` в служебном каталоге -`Tmp` (на той же ФС/шаре), затем жёсткая ссылка (без замены) во входящий (решение №17). Ответ ИС — набор файлов каталога +Файл кладётся во входящий каталог как `.` (D-1) сразу, открытием `wb` с перезаписью существующего — как в +старом АРМ (решение владельца 09.10.2026, change `hotfolder-direct-write`). Ответ ИС — набор файлов каталога `//`: ИС кладёт туда все распознанные документы задания, по одному `*.json` на документ (OQ-78). `poll` отдаёт все `*.json` по возрастанию имени, когда набор готов: есть хотя бы один `*.json`, а набор имён и размеров -не менялся `settle_seconds` (`SCAN_SETTLE_SECONDS`, `SettleTracker`). Имена каталогов (`Input`/`Output`/`Tmp`) -настраиваются, а не зашиты в код. +не менялся `settle_seconds` (`SCAN_SETTLE_SECONDS`, `SettleTracker`). Имена каталогов (`Input`/`Output`) настраиваются, +а не зашиты в код; `Tmp` — служебный каталог эмулятора (сборка набора ответа), шлюз его не использует. -Очистка (D-2) удаляет следы задания: ответ, входящий файл, временные файлы — при `SCAN_CLEANUP=true`. +Очистка (D-2) удаляет следы задания: ответ, входящий файл и его `.done` — при `SCAN_CLEANUP=true`. """ import contextlib -import errno import os import re import shutil @@ -29,8 +28,6 @@ _ILLEGAL_CHARS = frozenset(':<>"|?*') | frozenset(chr(code) for code in range(0x _FORBIDDEN_CHARS = _ILLEGAL_CHARS | frozenset(_PATH_SEPARATORS) # Расширение входящего файла — суффикс после последней точки, в нижнем регистре, 1–16 символов ASCII [0-9a-z] _EXTENSION_RE = re.compile(r"[0-9a-z]{1,16}") -# errno `os.link`, означающие «ФС не умеет жёсткие ссылки» (запасной путь записи — `os.replace`) -_NO_HARD_LINKS = (errno.EPERM, errno.EXDEV, errno.ENOTSUP, errno.EOPNOTSUPP) def validate_name(name: str) -> None: @@ -171,7 +168,8 @@ class LocalHotFolder: """Горячая папка в обычном каталоге: dev, compose-том, unit-тесты. `root`, `input_dir`, `output_dir`, `tmp_dir`, `answer_dir` и `answer_path` открыты намеренно (R-4): эмулятор - ИС Сканирование работает с ними обычными операциями файловой системы (листинг входящих, `.done`, запись ответа). + ИС Сканирование работает с ними обычными операциями файловой системы (листинг входящих, `.done`, запись ответа); + `tmp_dir` нужен только эмулятору (сборка набора ответа), `put` его не использует. `settle_seconds` и `clock` (монотонные часы) — готовность набора ответа (`SettleTracker`). `cleanup_enabled` (D-2) — очистка включена при `SCAN_CLEANUP`. """ @@ -204,76 +202,22 @@ class LocalHotFolder: return self.answer_dir(job_id) / f"{job_id}.json" def put(self, name: str, data: bytes) -> None: - """Атомарная запись без замены: `..part` в `tmp_dir`, затем жёсткая ссылка в `input_dir/` и - удаление `.part` как часть успешной публикации (R-8). Назначение уже есть — файл считается доставленным - (повтор задания, R-17): ничего не пишется, `.part` снимается только при включённой очистке. Корень не - создаётся (нет корня — `FileNotFoundError`).""" + """Записать `input_dir/` сразу (`wb`, перезапись существующего, как в старом АРМ — решение владельца + 09.10.2026). `Input` создаётся без родителей, корень — нет (нет корня — `FileNotFoundError`). При ошибке записи + недописанный файл снимается только при включённой очистке (R-3), ошибка снятия подавляется.""" validate_name(name) - self.tmp_dir.mkdir(exist_ok=True) self.input_dir.mkdir(exist_ok=True) - part = self.tmp_dir / f".{name}.part" target = self.input_dir / name - if target.exists(): - # R-17: `.part` от прошлой публикации может быть жёсткой ссылкой на этот файл (удаление после публикации - # подавляется, R-8, OQ-144) — писать и даже удалять его незачем, файл уже доставлен - self._drop_part(part) - return - # `.part` создаётся заново (№32): остаток от прерванной записи не должен ронять задание, поэтому ошибка его - # удаления подавляется; если он всё же остался, `_open_part` не даст записать «поверх» ссылки на чужой файл - with contextlib.suppress(OSError): - part.unlink(missing_ok=True) - fd = self._open_part(part) try: - with os.fdopen(fd, "wb") as handle: - handle.write(data) + target.write_bytes(data) except BaseException: - self._drop_part(part) - raise - try: - os.link(part, target) # без замены: существующее назначение даёт FileExistsError - except FileExistsError: - self._drop_part(part) - return - except OSError as err: - # ФС без жёстких ссылок (некоторые тома Docker/сетевые монтирования): запасной путь через replace с - # проверкой назначения (окно гонки крошечное, повтор задания с тем же job_id редок). Только для самой - # линковки: ошибка записи выше поднимается как есть - if err.errno in _NO_HARD_LINKS and not target.exists(): - self._replace_part(part, target) - return - self._drop_part(part) - raise - except BaseException: - self._drop_part(part) - raise - # завершение успешной записи (как rename в старой версии), не управляется флагом очистки: - # файл уже доставлен, ошибка удаления .part её не должна отменить (R-8) - with contextlib.suppress(OSError): - part.unlink(missing_ok=True) - - @staticmethod - def _open_part(part: Path) -> int: - """Открыть `.part` на запись: новый файл (`O_EXCL`). Если старый не удалился (`FileExistsError`) — усечь его - только когда это обычный файл без других ссылок (`st_nlink == 1`); ссылку на опубликованный файл не трогаем, - исключение поднимается как есть (R-17).""" - try: - return os.open(part, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644) - except FileExistsError: - if os.lstat(part).st_nlink != 1: - raise - return os.open(part, os.O_WRONLY | os.O_TRUNC) - - def _replace_part(self, part: Path, target: Path) -> None: - try: - os.replace(part, target) - except BaseException: - self._drop_part(part) + self._drop_partial(target) raise - def _drop_part(self, part: Path) -> None: + def _drop_partial(self, path: Path) -> None: if self.cleanup_enabled: with contextlib.suppress(OSError): - part.unlink(missing_ok=True) + path.unlink(missing_ok=True) def cleanup(self, job_id: str, name: str) -> bool: if not self.cleanup_enabled: @@ -282,7 +226,7 @@ class LocalHotFolder: folder = self.answer_dir(job_id) if folder.is_dir(): shutil.rmtree(folder) - for path in (self.input_dir / name, self.input_dir / f"{name}.done", self.tmp_dir / f".{name}.part"): + for path in (self.input_dir / name, self.input_dir / f"{name}.done"): try: os.unlink(path) except FileNotFoundError: diff --git a/api/tests/integration/test_roundtrip.py b/api/tests/integration/test_roundtrip.py index b74aaef..20f9363 100644 --- a/api/tests/integration/test_roundtrip.py +++ b/api/tests/integration/test_roundtrip.py @@ -3,7 +3,7 @@ Настоящий lifespan приложения (`create_app` → S3, NATS, воркер, `OcrNormalizer` со справочниками `api/rules/`) и эмулятор ИС Сканирование (`src.scan_emulator.run` в потоке, `SCAN_EMULATOR_DEFAULT` пуст) работают в процессе теста; NATS 127.0.0.1:4222 и MinIO 127.0.0.1:9000 — из `api/.env.test` (`make up`), горячая папка — tmp-каталог. -Входящий файл — `.pdf` (расширение `original_name` в нижнем регистре, `.part` живёт в `Tmp`), пару фикстур +Входящий файл — `.pdf` (расширение `original_name` в нижнем регистре, пишется прямо в `Input`), пару фикстур эмулятор находит по sha256 содержимого: в его каталоге фикстур лежат PDF и ответы (реальные `aep_*` из `fixtures/ocr` и синтетический `ok_doc`). @@ -16,7 +16,7 @@ 3. `aep_6.pdf` (один успешный файл) → `needs_check`, `result_s3` с двумя документами (`BALANCE`, `INCOME`); 4. `ok_doc.pdf` (синтетический успешный ответ в форме реальных) → `needs_check` (редакция `fsbu4` пуста) с `result_s3`; 5. `unknown.pdf` (содержимого нет в фикстурах) → ack, затем `INVALID_DOCUMENT` эмулятора; -6. `aep_6.pdf` при `SCAN_CLEANUP=true` → после результата `Input/` и `Output//` пусты, `Tmp/` без `.part`. +6. `aep_6.pdf` при `SCAN_CLEANUP=true` → после результата `Input/` и `Output//` пусты. Каждый конверт проверяется схемой `recognition-result-message`, `service.name` и `Nats-Msg-Id`; файл результата из MinIO — схемой `recognition-result`; склеенный сырой ответ в MinIO содержит ровно те файлы, что лежат в каталоге ответа, в порядке возрастания имени, с телами байт в байт; входящий файл совпадает с объектом S3. Готовность набора @@ -178,10 +178,6 @@ class Roundtrip: def input_dir(self) -> Path: return self.root / self.settings.SCAN_INPUT_DIR - @property - def tmp_dir(self) -> Path: - return self.root / self.settings.SCAN_TMP_DIR - def answer_dir(self, job_id: str) -> Path: return self.root / self.settings.SCAN_OUTPUT_DIR / job_id @@ -260,12 +256,6 @@ class Roundtrip: def inbox(self, job_id: str) -> list[str]: return sorted(p.name for p in self.input_dir.iterdir() if p.name.startswith(job_id)) - def stray_parts(self) -> list[str]: - """Временные `.part` записи во входящий каталог: после успешной публикации их в `Tmp` нет (решение №26).""" - return ( - sorted(p.name for p in self.tmp_dir.iterdir() if p.name.endswith(".part")) if self.tmp_dir.is_dir() else [] - ) - async def run(self, original_name: str, data: bytes, *, cleanup: bool = False) -> tuple[dict[str, Any], Envelope]: """Запрос → ack → итоговый конверт; общие проверки сквозного сценария. Возвращает запрос и итог.""" body = await self.request(original_name, data) @@ -283,8 +273,7 @@ class Roundtrip: """Сырой ответ в MinIO — склейка ответа эмулятора (D-4); входящий файл `.pdf` = объект S3. Без очистки (`SCAN_CLEANUP=false`) каталог ответа и входящий файл (с суффиксом `.done` эмулятора) остаются; при - очистке (`SCAN_CLEANUP=true`) во входящем каталоге пусто, каталога `Output//` нет. `.part` в `Tmp` нет - в обоих режимах. + очистке (`SCAN_CLEANUP=true`) во входящем каталоге пусто, каталога `Output//` нет. """ job_id = body["job_id"] assert result.body["file_id"] == body["file_id"] @@ -297,7 +286,6 @@ class Roundtrip: assert entry["response"]["file_uid"] == job_id stat = await self.s3.client.stat_object(self.s3.bucket, f"raw/{job_id}.json") assert stat.content_type == JSON_CONTENT_TYPE - assert self.stray_parts() == [] if cleanup: assert list(self.input_dir.iterdir()) == [] assert not self.answer_dir(job_id).exists() @@ -393,7 +381,7 @@ def _start_emulator(stack: AsyncExitStack, settings: Settings, root: Path, fixtu hotfolder_path=root, input_dir=settings.SCAN_INPUT_DIR, output_dir=settings.SCAN_OUTPUT_DIR, - tmp_dir=settings.SCAN_TMP_DIR, + tmp_dir="Tmp", # служебный каталог эмулятора; у шлюза его нет (change hotfolder-direct-write) fixtures_dir=fixtures, delay=EMULATOR_DELAY, poll_interval=0.05, @@ -650,4 +638,3 @@ async def test_cleanup_enabled_leaves_input_and_answer_empty(roundtrip_cleanup: assert [entry["name"] for entry in raw["files"]] == [f"{job_id}.json"] assert list(roundtrip_cleanup.input_dir.iterdir()) == [] assert not roundtrip_cleanup.answer_dir(job_id).exists() - assert roundtrip_cleanup.stray_parts() == [] diff --git a/api/tests/unit/conftest.py b/api/tests/unit/conftest.py index 83d676b..ccb0fb1 100644 --- a/api/tests/unit/conftest.py +++ b/api/tests/unit/conftest.py @@ -58,7 +58,6 @@ async def make_env(tmp_path: Path) -> AsyncIterator[EnvFactory]: root, settings.SCAN_INPUT_DIR, settings.SCAN_OUTPUT_DIR, - settings.SCAN_TMP_DIR, settle_seconds=settings.SCAN_SETTLE_SECONDS, cleanup=settings.SCAN_CLEANUP, ) diff --git a/api/tests/unit/test_config.py b/api/tests/unit/test_config.py index 1f29787..04d29e5 100644 --- a/api/tests/unit/test_config.py +++ b/api/tests/unit/test_config.py @@ -29,10 +29,10 @@ 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 not hasattr(s, "SCAN_TMP_DIR") # служебный каталог только у эмулятора (change hotfolder-direct-write) assert s.SCAN_SETTLE_SECONDS == 2 assert s.SCAN_POLL_SECONDS == 5 - assert s.SCAN_TIMEOUT_SECONDS == 300 + assert s.SCAN_TIMEOUT_SECONDS == 1200 assert s.SCAN_MAX_INFLIGHT == 4 assert s.OPENBAO.SMB.SMB_HOST == "" assert s.OPENBAO.NATS.NATS_SEED == "" @@ -80,6 +80,16 @@ def test_negative_settle_seconds_rejected(monkeypatch): get_settings() +def test_legacy_scan_tmp_dir_is_ignored(monkeypatch, minimal_env_file): + """Старый `.env` с `SCAN_TMP_DIR` (до change hotfolder-direct-write) не ломает старт: `extra="ignore"`.""" + _clean_env(monkeypatch) + monkeypatch.setenv("SCAN_TMP_DIR", "Tmp") + + s = get_settings() + + assert not hasattr(s, "SCAN_TMP_DIR") + + @pytest.mark.parametrize("settle", ["10", "11"]) def test_settle_not_less_than_timeout_rejected(monkeypatch, settle): """M-4: набор не может настояться за срок, который короче или равен settle — каждое задание давало бы TIMEOUT.""" @@ -156,7 +166,7 @@ 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 not hasattr(s, "SCAN_TMP_DIR") 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 @@ -172,7 +182,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 not hasattr(s, "SCAN_TMP_DIR") 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_hotfolder_factory.py b/api/tests/unit/test_hotfolder_factory.py index b0bc971..75d6046 100644 --- a/api/tests/unit/test_hotfolder_factory.py +++ b/api/tests/unit/test_hotfolder_factory.py @@ -20,7 +20,6 @@ def _local_settings(root: Path) -> Settings: "SCAN_HOTFOLDER_MODE": "local", "SCAN_HOTFOLDER_PATH": str(root), "SCAN_INPUT_DIR": "In", - "SCAN_TMP_DIR": "Служебный", "SCAN_SETTLE_SECONDS": 1.5, } ) @@ -45,7 +44,7 @@ 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.tmp_dir == tmp_path / "Tmp" # служебный каталог эмулятора, шлюз его не использует assert hf.settle_seconds == 1.5 assert hf.healthy() is True diff --git a/api/tests/unit/test_local_hotfolder.py b/api/tests/unit/test_local_hotfolder.py index a76bb25..910ede2 100644 --- a/api/tests/unit/test_local_hotfolder.py +++ b/api/tests/unit/test_local_hotfolder.py @@ -1,6 +1,6 @@ """Горячая папка в обычном каталоге (spec image-recognition-gateway, D-1, R-4/R-5, slice-0-skeleton). -Быстрые тесты на `tmp_path`: атомарная запись входящего файла через служебный каталог `Tmp`, опрос набора ответов ИС +Быстрые тесты на `tmp_path`: запись входящего файла прямо в `Input` с перезаписью, опрос набора ответов ИС с готовностью по стабилизации (settle, часы подменены), проверка здоровья, валидация имён. """ @@ -41,6 +41,17 @@ def _settled(tmp_path: Path, settle: float = 2) -> tuple[LocalHotFolder, FakeClo return LocalHotFolder(tmp_path, settle_seconds=settle, clock=clock), clock +def _failing_write(self: Path, data: bytes) -> int: + """Недописанный файл: на диск попадают первые два байта, затем ошибка ввода-вывода.""" + with open(self, "wb") as handle: + handle.write(data[:2]) + raise OSError(errno.ENOSPC, "диск полон") + + +def _failing_unlink(self: Path, missing_ok: bool = False) -> None: + raise PermissionError(errno.EACCES, "нет прав") + + class TestInboxName: def test_job_id_and_lowercase_extension(self) -> None: job_id = _job_id() @@ -57,34 +68,12 @@ class TestInboxName: inbox_name("job", original) -class TestPutWithoutReplace: - def test_existing_destination_is_kept(self, tmp_path: Path) -> None: - hf = LocalHotFolder(tmp_path) - hf.put("job.pdf", b"first") - hf.put("job.pdf", b"second") - assert (hf.input_dir / "job.pdf").read_bytes() == b"first" - # R-17: назначение есть — повтор ничего не пишет; прежнего `.part` после успешной публикации нет - assert not (hf.tmp_dir / ".job.pdf.part").exists() - - def test_existing_destination_part_removed_when_cleanup_enabled(self, tmp_path: Path) -> None: - hf = LocalHotFolder(tmp_path, cleanup=True) - hf.put("job.pdf", b"first") - hf.put("job.pdf", b"second") - assert (hf.input_dir / "job.pdf").read_bytes() == b"first" - assert not (hf.tmp_dir / ".job.pdf.part").exists() - - def test_no_temp_files_in_input(self, tmp_path: Path) -> None: - hf = LocalHotFolder(tmp_path) - hf.put("job.pdf", b"x") - assert [p.name for p in hf.input_dir.iterdir()] == ["job.pdf"] - assert not (hf.tmp_dir / ".job.pdf.part").exists() # после успешной публикации .part нет - - class TestCleanup: def _fill(self, hf: LocalHotFolder, job_id: str) -> None: hf.put(f"{job_id}.pdf", b"x") (hf.input_dir / f"{job_id}.pdf.done").write_bytes(b"") - (hf.tmp_dir / f".{job_id}.pdf.part").write_bytes(b"orphan") + hf.tmp_dir.mkdir(exist_ok=True) + (hf.tmp_dir / f".{job_id}.pdf.part").write_bytes(b"legacy") # остаток прежних версий шлюза — не наш _write_answer(hf, job_id, b"{}") _write_answer(hf, job_id, b"{}", name=f"{job_id}_ofr.json") (hf.answer_dir(job_id) / "readme.txt").write_bytes(b"keep") @@ -104,7 +93,7 @@ class TestCleanup: assert hf.cleanup(job_id, f"{job_id}.pdf") is True assert not (hf.input_dir / f"{job_id}.pdf").exists() assert not (hf.input_dir / f"{job_id}.pdf.done").exists() - assert not (hf.tmp_dir / f".{job_id}.pdf.part").exists() + assert (hf.tmp_dir / f".{job_id}.pdf.part").exists() # Tmp очистка не трогает assert not hf.answer_dir(job_id).exists() # каталог удалён вместе с readme.txt assert hf.cleanup(job_id, f"{job_id}.pdf") is True # повтор — не ошибка @@ -181,60 +170,42 @@ class TestPut: hf.put(name, data) assert (tmp_path / "Input" / name).read_bytes() == data - def test_leaves_no_part_files(self, tmp_path: Path) -> None: + def test_creates_input_dir_only(self, tmp_path: Path) -> None: hf = LocalHotFolder(tmp_path) - job_id = _job_id() - 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_part_is_written_in_tmp_not_in_input(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - """OQ-79: во время записи во входящем каталоге нет временного файла — он в `Tmp/`, затем link в `Input`.""" - hf = LocalHotFolder(tmp_path) - job_id = _job_id() - name = inbox_name(job_id, "aep_5.pdf") - seen: dict[str, Any] = {} - real_link = os.link - - 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_link(src, dst) - - monkeypatch.setattr("src.gateway.hotfolder.os.link", 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() + assert sorted(p.name for p in tmp_path.iterdir()) == ["Input"] # Tmp не создаётся - 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_existing_file_is_overwritten(self, tmp_path: Path) -> None: + """Как в старом АРМ (решение владельца 09.10.2026): повтор задания перезаписывает входящий файл.""" + hf = LocalHotFolder(tmp_path, settle_seconds=0) + hf.put("job.pdf", b"first, longer") + hf.put("job.pdf", b"second") + assert (hf.input_dir / "job.pdf").read_bytes() == b"second" - 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_write_failure_keeps_partial_file_when_cleanup_disabled( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + hf = LocalHotFolder(tmp_path, settle_seconds=0) + monkeypatch.setattr(Path, "write_bytes", _failing_write) + with pytest.raises(OSError, match="диск полон"): + hf.put("job.pdf", b"data") + assert (hf.input_dir / "job.pdf").read_bytes() == b"da" + + def test_write_failure_removes_partial_file_when_cleanup_enabled( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + hf = LocalHotFolder(tmp_path, settle_seconds=0, cleanup=True) + monkeypatch.setattr(Path, "write_bytes", _failing_write) + with pytest.raises(OSError, match="диск полон"): + hf.put("job.pdf", b"data") + assert not (hf.input_dir / "job.pdf").exists() + + def test_remove_failure_does_not_mask_write_error(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + hf = LocalHotFolder(tmp_path, settle_seconds=0, cleanup=True) + monkeypatch.setattr(Path, "write_bytes", _failing_write) + monkeypatch.setattr(Path, "unlink", _failing_unlink) + with pytest.raises(OSError, match="диск полон"): + hf.put("job.pdf", b"data") def test_existing_input_dir_is_fine(self, tmp_path: Path) -> None: (tmp_path / "Input").mkdir() @@ -285,227 +256,6 @@ class TestPut: hf.put(name, b"x") assert sorted(p.name for p in hf.input_dir.iterdir()) == sorted(names) - def test_cleans_part_file_on_failure(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - """При cleanup=True и ошибке link: .part удаляется.""" - hf = LocalHotFolder(tmp_path, cleanup=True) - - def boom(src: Any, dst: Any) -> None: - raise OSError("диск недоступен") - - monkeypatch.setattr("src.gateway.hotfolder.os.link", boom) - 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_keeps_part_file_on_failure_when_cleanup_disabled( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch - ) -> None: - """При cleanup=False и ошибке link: .part остаётся в Tmp (T11.1).""" - hf = LocalHotFolder(tmp_path, cleanup=False) - - def boom(src: Any, dst: Any) -> None: - raise OSError("диск недоступен") - - monkeypatch.setattr("src.gateway.hotfolder.os.link", boom) - with pytest.raises(OSError, match="диск недоступен"): - hf.put("a.pdf", b"data") - assert list(hf.input_dir.iterdir()) == [] - assert [p.name for p in hf.tmp_dir.iterdir()] == [".a.pdf.part"] - - def test_successful_put_despite_final_unlink_failure(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - """Успешный link, но ошибка в завершающем unlink .part (R-8): put НЕ падает, файл доставлен (T11.2).""" - hf = LocalHotFolder(tmp_path) - - unlink_calls = [] - - real_unlink = os.unlink - - def unlink_with_failure(path: Any) -> None: - unlink_calls.append(path) - if str(path).endswith(".part"): - raise PermissionError("удаление запрещено") - real_unlink(path) - - monkeypatch.setattr("src.gateway.hotfolder.os.unlink", unlink_with_failure) - hf.put("a.pdf", b"data") # НЕ падает несмотря на ошибку unlink - assert (hf.input_dir / "a.pdf").read_bytes() == b"data" - assert (hf.tmp_dir / ".a.pdf.part").exists() # .part не удалён из-за ошибки - assert unlink_calls # попытка удаления была - - def test_cleanup_failure_does_not_mask_original_error( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch - ) -> None: - """Ошибка удаления `.part` на пути ошибки link подавляется (T11.1); исходное исключение проходит.""" - hf = LocalHotFolder(tmp_path, cleanup=True) - - def boom(src: Any, dst: Any) -> None: - raise OSError("диск недоступен") - - def unlink_fails(self: Path, missing_ok: bool = False) -> None: - raise PermissionError("удаление .part запрещено") - - monkeypatch.setattr("src.gateway.hotfolder.os.link", boom) - monkeypatch.setattr(Path, "unlink", unlink_fails) - with pytest.raises(OSError, match="диск недоступен"): - hf.put("a.pdf", b"data") - - -class TestPartIsCreatedFresh: - """Решение №32: `.part` создаётся заново (удаление старого + `O_EXCL`), fallback — только вокруг `os.link`.""" - - @pytest.mark.parametrize("cleanup", [False, True]) - def test_existing_destination_is_not_touched_by_repeat(self, tmp_path: Path, cleanup: bool) -> None: - """R-17: назначение есть, `.part` — жёсткая ссылка на него: байты не меняются, put не падает.""" - hf = LocalHotFolder(tmp_path, cleanup=cleanup) - hf.put("job.pdf", b"first") - published = hf.input_dir / "job.pdf" - stale = hf.tmp_dir / ".job.pdf.part" - os.link(published, stale) - hf.put("job.pdf", b"different data") - assert published.read_bytes() == b"first" - assert stale.exists() is (not cleanup) # очистка снимает `.part`, иначе он остаётся как был - if not cleanup: - assert stale.read_bytes() == b"first" and published.stat().st_nlink == 2 - - def test_existing_destination_with_undeletable_part_does_not_fail( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch - ) -> None: - hf = LocalHotFolder(tmp_path) # очистка выключена: удаление `.part` вообще не пытается - hf.put("job.pdf", b"first") - os.link(hf.input_dir / "job.pdf", hf.tmp_dir / ".job.pdf.part") - - def deny(self: Path, missing_ok: bool = False) -> None: - raise PermissionError("удаление запрещено") - - monkeypatch.setattr(Path, "unlink", deny) - hf.put("job.pdf", b"second") - assert (hf.input_dir / "job.pdf").read_bytes() == b"first" - - def test_undeletable_stale_part_with_single_link_is_truncated( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch - ) -> None: - """Назначения нет, старый `.part` не удаляется, но это обычный файл (nlink == 1): put успешен.""" - hf = LocalHotFolder(tmp_path) - hf.tmp_dir.mkdir() - stale = hf.tmp_dir / ".job.pdf.part" - stale.write_bytes(b"old and longer content") - - def deny(self: Path, missing_ok: bool = False) -> None: - raise PermissionError("удаление запрещено") - - monkeypatch.setattr(Path, "unlink", deny) - hf.put("job.pdf", b"new") - assert (hf.input_dir / "job.pdf").read_bytes() == b"new" - - def test_undeletable_stale_part_with_other_links_is_not_written( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch - ) -> None: - """Назначения нет, старый `.part` не удаляется и имеет nlink == 2 (синтетически): запись не идёт, ошибка не - маскируется, второй файл не изменён.""" - hf = LocalHotFolder(tmp_path) - hf.tmp_dir.mkdir() - other = tmp_path / "other.bin" - other.write_bytes(b"precious") - os.link(other, hf.tmp_dir / ".job.pdf.part") - - def deny(self: Path, missing_ok: bool = False) -> None: - raise PermissionError("удаление запрещено") - - monkeypatch.setattr(Path, "unlink", deny) - with pytest.raises(FileExistsError): - hf.put("job.pdf", b"new") - assert other.read_bytes() == b"precious" - assert not (hf.input_dir / "job.pdf").exists() - - def test_stale_part_is_replaced_before_publication(self, tmp_path: Path) -> None: - hf = LocalHotFolder(tmp_path) - hf.tmp_dir.mkdir() - (hf.tmp_dir / ".job.pdf.part").write_bytes(b"old and longer content") - hf.put("job.pdf", b"new") - assert (hf.input_dir / "job.pdf").read_bytes() == b"new" - assert not (hf.tmp_dir / ".job.pdf.part").exists() - - def test_part_is_opened_exclusively(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - hf = LocalHotFolder(tmp_path) - flags: list[int] = [] - real_open = os.open - - def spy(path: Any, flag: int, mode: int = 0o777, **kwargs: Any) -> int: - flags.append(flag) - return real_open(path, flag, mode, **kwargs) - - monkeypatch.setattr("src.gateway.hotfolder.os.open", spy) - hf.put("job.pdf", b"x") - assert flags and flags[0] & os.O_EXCL and flags[0] & os.O_CREAT - - def test_write_error_propagates_as_itself(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - """EPERM записи не превращается в `FileNotFoundError` от `os.replace` несуществующего `.part`.""" - hf = LocalHotFolder(tmp_path, cleanup=True) - - class FailingFile: - def __enter__(self) -> "FailingFile": - return self - - def __exit__(self, *exc: object) -> None: - return None - - def write(self, data: bytes) -> int: - raise OSError(errno.EPERM, "запись запрещена") - - monkeypatch.setattr("src.gateway.hotfolder.os.fdopen", lambda fd, mode: (os.close(fd), FailingFile())[1]) - with pytest.raises(OSError, match="запись запрещена") as caught: - hf.put("job.pdf", b"data") - assert caught.value.errno == errno.EPERM and not isinstance(caught.value, FileNotFoundError) - assert list(hf.input_dir.iterdir()) == [] and list(hf.tmp_dir.iterdir()) == [] - - @pytest.mark.parametrize("code", [errno.EPERM, errno.EXDEV, errno.ENOTSUP, errno.EOPNOTSUPP]) - def test_fallback_to_replace_when_link_unsupported( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, code: int - ) -> None: - hf = LocalHotFolder(tmp_path) - - def no_links(src: Any, dst: Any) -> None: - raise OSError(code, "жёсткие ссылки не поддерживаются") - - monkeypatch.setattr("src.gateway.hotfolder.os.link", no_links) - hf.put("job.pdf", b"data") - assert (hf.input_dir / "job.pdf").read_bytes() == b"data" - assert list(hf.tmp_dir.iterdir()) == [] # `.part` стал назначением - - def test_no_fallback_when_destination_appeared_during_link( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch - ) -> None: - """Назначение появилось между проверкой и линковкой, а `os.link` дал EPERM: `os.replace` не затирает.""" - hf = LocalHotFolder(tmp_path, cleanup=True) - hf.input_dir.mkdir() - - def no_links(src: Any, dst: Any) -> None: - Path(dst).write_bytes(b"first") # файл успел появиться (параллельный повтор того же задания) - raise OSError(errno.EPERM, "жёсткие ссылки не поддерживаются") - - monkeypatch.setattr("src.gateway.hotfolder.os.link", no_links) - with pytest.raises(OSError, match="жёсткие ссылки"): - hf.put("job.pdf", b"second") - assert (hf.input_dir / "job.pdf").read_bytes() == b"first" - - def test_replace_failure_after_link_fallback_propagates( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch - ) -> None: - hf = LocalHotFolder(tmp_path, cleanup=True) - - def no_links(src: Any, dst: Any) -> None: - raise OSError(errno.EPERM, "жёсткие ссылки не поддерживаются") - - def broken_replace(src: Any, dst: Any) -> None: - raise OSError(errno.EIO, "замена не удалась") - - monkeypatch.setattr("src.gateway.hotfolder.os.link", no_links) - monkeypatch.setattr("src.gateway.hotfolder.os.replace", broken_replace) - with pytest.raises(OSError, match="замена не удалась"): - hf.put("job.pdf", b"data") - assert list(hf.tmp_dir.iterdir()) == [] # при очистке `.part` снят - class TestPoll: """`poll` — все `*.json` каталога `//` по имени, когда набор настоялся `settle_seconds`.""" diff --git a/api/tests/unit/test_worker.py b/api/tests/unit/test_worker.py index 2ae1dac..0be8fc7 100644 --- a/api/tests/unit/test_worker.py +++ b/api/tests/unit/test_worker.py @@ -953,13 +953,14 @@ class TestSlice3a: assert "Очистка горячей папки не удалась" in warnings[0] and job_id in warnings[0] and "нет прав" in warnings[0] async def test_existing_input_waits_for_answer(self, make_env: EnvFactory) -> None: - """Review Focus 4: входящий файл уже лежит (повтор задания), ответа нет — TIMEOUT, файл не перезаписан.""" + """Входящий файл уже лежит (повтор задания), ответа нет — TIMEOUT; файл перезаписан объектом S3, как в старом + АРМ (change hotfolder-direct-write, решение владельца 09.10.2026).""" 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.local.input_dir / f"{body['job_id']}.pdf").read_bytes() == PDF_BYTES assert env.inbox() == [f"{body['job_id']}.pdf"] async def test_ocr_normalizer_produces_documents(self, make_env: EnvFactory) -> None: