24af6f1b9a
Для `<uuid>_<name>.<ext>` эмулятор собирает все `<name>.json` и `<name>_*.json` из SCAN_FIXTURES_DIR (не префикс `<name>*`) и кладёт их в `Output/<uuid>/` как `<uuid>.json` / `<uuid>_<suffix>.json` с `file_uid = <uuid>`. Набор пишется в `<root>/<SCAN_TMP_DIR>/.<uuid>.answer` и переименовывается в каталог ответа целиком (R-26). Ответ «нет пары» — `data_ocr` объектом с пустыми полями (R-28). Решения исполнителя — design №75 (R-35). Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
623 lines
29 KiB
Python
623 lines
29 KiB
Python
"""Эмулятор ИС Сканирование (spec scan-emulator, D-6, R-11, slice-0-skeleton).
|
||
|
||
Сценарии spec «Точное имя», «Суффикс», «Несколько файлов на одно задание», «Нет пары» — на копиях реальных
|
||
`../fixtures/ocr/*.json` (маркер `fixtures`); остальное — на синтетических фикстурах в `tmp_path`, без внешних
|
||
репозиториев и без Docker.
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import uuid
|
||
from collections.abc import Iterator
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from loguru import logger
|
||
|
||
import src.scan_emulator as emulator
|
||
from src.gateway.hotfolder import LocalHotFolder, inbox_name
|
||
from src.gateway.merge import merge_answers
|
||
from src.normalizer import NullNormalizer
|
||
from src.scan_emulator import EmulatorSettings, process_once, run
|
||
from src.scan_emulator.__main__ import main
|
||
|
||
API_DIR = Path(__file__).resolve().parents[2]
|
||
# `<корень scan-gateway>/../fixtures/ocr` (workspace и worktree `.wt/<срез>/`), как у теста идентичности контракта
|
||
REAL_FIXTURES = API_DIR.parent.parent / "fixtures" / "ocr"
|
||
|
||
|
||
def _uid() -> str:
|
||
return str(uuid.uuid4())
|
||
|
||
|
||
def _hotfolder(tmp_path: Path) -> LocalHotFolder:
|
||
root = tmp_path / "hotfolder"
|
||
root.mkdir()
|
||
return LocalHotFolder(root)
|
||
|
||
|
||
def _fixtures_dir(tmp_path: Path, files: dict[str, Any] | None = None) -> Path:
|
||
path = tmp_path / "fixtures"
|
||
path.mkdir(exist_ok=True)
|
||
for name, content in (files or {}).items():
|
||
text = content if isinstance(content, str) else json.dumps(content, ensure_ascii=False)
|
||
(path / name).write_text(text, encoding="utf-8")
|
||
return path
|
||
|
||
|
||
def _drop(hf: LocalHotFolder, name: str, data: bytes = b"%PDF") -> str:
|
||
"""Положить `<uuid>_<name>` во входящий каталог, вернуть uuid."""
|
||
job_id = _uid()
|
||
hf.put(inbox_name(job_id, name), data)
|
||
return job_id
|
||
|
||
|
||
def _answer(hf: LocalHotFolder, job_id: str, name: str | None = None) -> dict[str, Any]:
|
||
"""Файл ответа каталога `Output/<uuid>/` (по умолчанию основной `<uuid>.json`)."""
|
||
path = hf.answer_path(job_id) if name is None else hf.answer_dir(job_id) / name
|
||
result: dict[str, Any] = json.loads(path.read_text(encoding="utf-8"))
|
||
return result
|
||
|
||
|
||
def _names(hf: LocalHotFolder, job_id: str) -> list[str]:
|
||
return sorted(p.name for p in hf.answer_dir(job_id).iterdir())
|
||
|
||
|
||
# форма `data_ocr` реального ответа-ошибки ИС (aep_5.json); R-28
|
||
EMPTY_DATA_OCR: dict[str, Any] = {
|
||
"label": {"form_name": "", "confidence": 0},
|
||
"organization": {"form_name": "", "confidence": 0},
|
||
"inn": {"form_name": "", "confidence": 0},
|
||
"units": {"form_name": "", "confidence": 0},
|
||
"date": {"form_name": "", "confidence": 0},
|
||
"table_structure": [],
|
||
}
|
||
|
||
|
||
@pytest.fixture
|
||
def log_messages() -> Iterator[list[str]]:
|
||
messages: list[str] = []
|
||
handler_id = logger.add(lambda message: messages.append(str(message)), level="DEBUG", format="{level}|{message}")
|
||
yield messages
|
||
logger.remove(handler_id)
|
||
|
||
|
||
@pytest.fixture
|
||
def real_fixtures(tmp_path: Path) -> Path:
|
||
if not REAL_FIXTURES.is_dir():
|
||
pytest.skip("fixtures недоступен: положите репозиторий fixtures рядом с workspace")
|
||
target = tmp_path / "real-fixtures"
|
||
target.mkdir()
|
||
for source in REAL_FIXTURES.glob("*.json"):
|
||
shutil.copy(source, target / source.name)
|
||
return target
|
||
|
||
|
||
@pytest.mark.fixtures
|
||
class TestRealFixtures:
|
||
@pytest.mark.parametrize("name", ["aep_5.pdf", "aep_5.PDF"])
|
||
def test_exact_name(self, tmp_path: Path, real_fixtures: Path, name: str) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
job_id = _drop(hf, name)
|
||
assert process_once(hf.root, real_fixtures, 0) == [job_id]
|
||
expected = json.loads((real_fixtures / "aep_5.json").read_text(encoding="utf-8"))
|
||
expected["file_uid"] = job_id
|
||
assert _answer(hf, job_id) == expected
|
||
assert (hf.input_dir / f"{job_id}_{name}.done").is_file()
|
||
assert not (hf.input_dir / f"{job_id}_{name}").exists()
|
||
|
||
def test_suffix(self, tmp_path: Path, real_fixtures: Path) -> None:
|
||
"""`aep_8.pdf` -> `aep_8_form4.json` как `<uuid>_form4.json` (раньше — `<uuid>.json`, правило `<name>*`)."""
|
||
hf = _hotfolder(tmp_path)
|
||
job_id = _drop(hf, "aep_8.pdf")
|
||
process_once(hf.root, real_fixtures, 0)
|
||
expected = json.loads((real_fixtures / "aep_8_form4.json").read_text(encoding="utf-8"))
|
||
expected["file_uid"] = job_id
|
||
assert _names(hf, job_id) == [f"{job_id}_form4.json"]
|
||
assert _answer(hf, job_id, f"{job_id}_form4.json") == expected
|
||
assert (hf.input_dir / f"{job_id}_aep_8.pdf.done").is_file()
|
||
|
||
def test_two_files_for_one_job(self, tmp_path: Path, real_fixtures: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
job_id = _drop(hf, "aep_3.pdf")
|
||
assert process_once(hf.root, real_fixtures, 0) == [job_id]
|
||
assert _names(hf, job_id) == [f"{job_id}.json", f"{job_id}_ofr.json"]
|
||
main = json.loads((real_fixtures / "aep_3.json").read_text(encoding="utf-8"))
|
||
ofr = json.loads((real_fixtures / "aep_3_ofr.json").read_text(encoding="utf-8"))
|
||
assert main["error"] == {"code": "", "message": ""}
|
||
assert ofr["error"]["code"] == "INVALID_DOCUMENT"
|
||
assert _answer(hf, job_id) == {**main, "file_uid": job_id}
|
||
assert _answer(hf, job_id, f"{job_id}_ofr.json") == {**ofr, "file_uid": job_id}
|
||
assert (hf.input_dir / f"{job_id}_aep_3.pdf.done").is_file()
|
||
|
||
def test_single_success_file(self, tmp_path: Path, real_fixtures: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
job_id = _drop(hf, "aep_6.pdf")
|
||
process_once(hf.root, real_fixtures, 0)
|
||
assert _names(hf, job_id) == [f"{job_id}.json"]
|
||
expected = json.loads((real_fixtures / "aep_6.json").read_text(encoding="utf-8"))
|
||
assert _answer(hf, job_id) == {**expected, "file_uid": job_id}
|
||
|
||
def test_no_pair(self, tmp_path: Path, real_fixtures: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
job_id = _drop(hf, "unknown.pdf")
|
||
process_once(hf.root, real_fixtures, 0)
|
||
assert _names(hf, job_id) == [f"{job_id}.json"]
|
||
assert _answer(hf, job_id) == {
|
||
"file_uid": job_id,
|
||
"error": {"code": "INVALID_DOCUMENT", "message": "unknown.pdf: нет эталонного ответа в фикстурах"},
|
||
"data_ocr": EMPTY_DATA_OCR,
|
||
}
|
||
assert (hf.input_dir / f"{job_id}_unknown.pdf.done").is_file()
|
||
|
||
def test_no_pair_has_the_shape_of_real_error_answers(self, tmp_path: Path, real_fixtures: Path) -> None:
|
||
"""R-28: форма `data_ocr` совпадает с реальным ответом-ошибкой `aep_5.json` (кроме `0` вместо `0.0`)."""
|
||
hf = _hotfolder(tmp_path)
|
||
job_id = _drop(hf, "unknown.pdf")
|
||
process_once(hf.root, real_fixtures, 0)
|
||
real = json.loads((real_fixtures / "aep_5.json").read_text(encoding="utf-8"))
|
||
assert _answer(hf, job_id)["data_ocr"] == real["data_ocr"]
|
||
assert list(_answer(hf, job_id)["data_ocr"]) == list(real["data_ocr"])
|
||
|
||
|
||
class TestFixtureLookup:
|
||
def test_name_with_other_characters_is_not_taken(self, tmp_path: Path) -> None:
|
||
"""В набор входят `<name>.json` и `<name>_*.json`; `a_b-1.json` (нет `_` после имени) — нет."""
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a_b-1.json": {"v": "dash"}, "a_b.json": {"file_uid": "x", "v": "exact"}})
|
||
job_id = _drop(hf, "a_b.pdf")
|
||
process_once(hf.root, fixtures, 0)
|
||
assert _names(hf, job_id) == [f"{job_id}.json"]
|
||
assert _answer(hf, job_id) == {"file_uid": job_id, "v": "exact"}
|
||
|
||
def test_all_suffix_fixtures_are_taken(self, tmp_path: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a_zz.json": {"v": "zz"}, "a_b.json": {"v": "b"}, "a_c.json": {"v": "c"}})
|
||
job_id = _drop(hf, "a.pdf")
|
||
process_once(hf.root, fixtures, 0)
|
||
assert _names(hf, job_id) == [f"{job_id}_b.json", f"{job_id}_c.json", f"{job_id}_zz.json"]
|
||
assert [_answer(hf, job_id, f"{job_id}_{s}.json")["v"] for s in ("b", "c", "zz")] == ["b", "c", "zz"]
|
||
|
||
def test_foreign_prefix_is_not_taken(self, tmp_path: Path) -> None:
|
||
"""`aep_3` не забирает `aep_30*`: правило `<name>.json` и `<name>_*.json`, а не `<name>*`."""
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(
|
||
tmp_path,
|
||
{
|
||
"aep_3.json": {"v": "main"},
|
||
"aep_3_ofr.json": {"v": "ofr"},
|
||
"aep_30.json": {"v": 30},
|
||
"aep_30_x.json": {"v": 31},
|
||
},
|
||
)
|
||
job_id = _drop(hf, "aep_3.pdf")
|
||
process_once(hf.root, fixtures, 0)
|
||
assert _names(hf, job_id) == [f"{job_id}.json", f"{job_id}_ofr.json"]
|
||
assert _answer(hf, job_id)["v"] == "main"
|
||
assert _answer(hf, job_id, f"{job_id}_ofr.json")["v"] == "ofr"
|
||
|
||
def test_underscore_without_suffix_is_not_taken(self, tmp_path: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a_.json": {"v": 1}})
|
||
job_id = _drop(hf, "a.pdf")
|
||
process_once(hf.root, fixtures, 0)
|
||
assert _answer(hf, job_id)["error"]["code"] == "INVALID_DOCUMENT"
|
||
|
||
def test_file_uid_added_when_absent_and_replaced_when_present(self, tmp_path: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"data_ocr": [1]}, "b.json": {"file_uid": "old", "data_ocr": []}})
|
||
first, second = _drop(hf, "a.pdf"), _drop(hf, "b.pdf")
|
||
process_once(hf.root, fixtures, 0)
|
||
assert _answer(hf, first) == {"file_uid": first, "data_ocr": [1]}
|
||
assert _answer(hf, second) == {"file_uid": second, "data_ocr": []}
|
||
|
||
def test_name_without_extension(self, tmp_path: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"scan.json": {"v": 1}})
|
||
job_id = _drop(hf, "scan")
|
||
process_once(hf.root, fixtures, 0)
|
||
assert _answer(hf, job_id) == {"file_uid": job_id, "v": 1}
|
||
|
||
def test_name_with_glob_characters_is_literal(self, tmp_path: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}})
|
||
job_id = _drop(hf, "[a]*.pdf")
|
||
process_once(hf.root, fixtures, 0)
|
||
assert _answer(hf, job_id)["error"]["code"] == "INVALID_DOCUMENT"
|
||
|
||
def test_non_ascii_not_escaped(self, tmp_path: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"text": "Баланс"}})
|
||
job_id = _drop(hf, "a.pdf")
|
||
process_once(hf.root, fixtures, 0)
|
||
assert "Баланс" in hf.answer_path(job_id).read_text(encoding="utf-8")
|
||
|
||
def test_missing_fixtures_dir_gives_error_answer(self, tmp_path: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
job_id = _drop(hf, "a.pdf")
|
||
process_once(hf.root, tmp_path / "нет-такого", 0)
|
||
assert _answer(hf, job_id)["error"]["code"] == "INVALID_DOCUMENT"
|
||
|
||
@pytest.mark.parametrize("content", ["{не json", "[1, 2]", ""])
|
||
def test_broken_fixture_gives_error_answer(self, tmp_path: Path, log_messages: list[str], content: str) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": content})
|
||
job_id = _drop(hf, "a.pdf")
|
||
assert process_once(hf.root, fixtures, 0) == [job_id]
|
||
answer = _answer(hf, job_id)
|
||
assert answer["file_uid"] == job_id
|
||
assert answer["data_ocr"] == EMPTY_DATA_OCR
|
||
assert answer["error"]["code"] == "INVALID_DOCUMENT"
|
||
assert "a.pdf" in answer["error"]["message"]
|
||
assert "a.json" in answer["error"]["message"]
|
||
assert any(m.startswith("ERROR|") and "a.json" in m for m in log_messages)
|
||
assert (hf.input_dir / f"{job_id}_a.pdf.done").is_file()
|
||
|
||
|
||
class TestProcessOnce:
|
||
def test_returns_processed_uuids(self, tmp_path: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}})
|
||
ids = {_drop(hf, "a.pdf"), _drop(hf, "a.pdf")}
|
||
assert set(process_once(hf.root, fixtures, 0)) == ids
|
||
|
||
def test_nothing_to_do(self, tmp_path: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
assert process_once(hf.root, _fixtures_dir(tmp_path), 0) == []
|
||
|
||
def test_missing_input_dir_is_not_an_error(self, tmp_path: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
assert not hf.input_dir.exists()
|
||
assert process_once(hf.root, _fixtures_dir(tmp_path), 0) == []
|
||
|
||
def test_second_pass_does_not_reprocess(self, tmp_path: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}})
|
||
job_id = _drop(hf, "a.pdf")
|
||
assert process_once(hf.root, fixtures, 0) == [job_id]
|
||
mtime = hf.answer_path(job_id).stat().st_mtime_ns
|
||
assert process_once(hf.root, fixtures, 0) == []
|
||
assert hf.answer_path(job_id).stat().st_mtime_ns == mtime
|
||
|
||
def test_ignores_done_dotfiles_and_non_uuid(self, tmp_path: Path, log_messages: list[str]) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}})
|
||
hf.input_dir.mkdir()
|
||
job_id = _uid()
|
||
ignored = [
|
||
f"{job_id}_a.pdf.done",
|
||
f".{job_id}_a.pdf.part",
|
||
"a.pdf",
|
||
"not-a-uuid_a.pdf",
|
||
f"{job_id}a.pdf",
|
||
f"{job_id}_",
|
||
f"{job_id[:-1]}_a.pdf",
|
||
]
|
||
for name in ignored:
|
||
(hf.input_dir / name).write_bytes(b"x")
|
||
(hf.input_dir / f"{_uid()}_dir.pdf").mkdir()
|
||
assert process_once(hf.root, fixtures, 0) == []
|
||
assert sorted(p.name for p in hf.input_dir.iterdir() if p.is_file()) == sorted(ignored)
|
||
assert not hf.output_dir.exists()
|
||
warnings = [m for m in log_messages if m.startswith("WARNING|")]
|
||
assert len(warnings) == 1
|
||
assert "not-a-uuid_a.pdf" in warnings[0]
|
||
|
||
def test_delay_once_per_pass(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}})
|
||
sleeps: list[float] = []
|
||
monkeypatch.setattr("src.scan_emulator.time.sleep", sleeps.append)
|
||
assert process_once(hf.root, fixtures, 2) == []
|
||
assert sleeps == [] # нет работы — нет задержки
|
||
_drop(hf, "a.pdf")
|
||
_drop(hf, "a.pdf")
|
||
process_once(hf.root, fixtures, 2)
|
||
assert sleeps == [2]
|
||
|
||
def test_answer_exists_before_done_and_is_atomic(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}})
|
||
job_id = _drop(hf, "a.pdf")
|
||
seen: dict[str, bool] = {}
|
||
original_replace = __import__("os").replace
|
||
|
||
def spy(src: Path, dst: Path) -> None:
|
||
if str(dst).endswith(".done"):
|
||
seen["answer_before_done"] = hf.answer_path(job_id).is_file()
|
||
original_replace(src, dst)
|
||
|
||
monkeypatch.setattr("src.scan_emulator.os.replace", spy)
|
||
process_once(hf.root, fixtures, 0)
|
||
assert seen == {"answer_before_done": True}
|
||
assert [p.name for p in hf.answer_path(job_id).parent.iterdir()] == [f"{job_id}.json"]
|
||
|
||
def test_custom_dir_names(self, tmp_path: Path) -> None:
|
||
root = tmp_path / "hf"
|
||
root.mkdir()
|
||
hf = LocalHotFolder(root, "In", "Out")
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}})
|
||
job_id = _drop(hf, "a.pdf")
|
||
assert process_once(root, fixtures, 0, input_dir="In", output_dir="Out") == [job_id]
|
||
assert hf.answer_path(job_id).is_file()
|
||
|
||
def test_interop_put_emulator_poll(self, tmp_path: Path) -> None:
|
||
"""Сквозной «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)
|
||
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 TestAnswerSet:
|
||
def _two_file_job(self, tmp_path: Path) -> tuple[LocalHotFolder, Path, str]:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": "main"}, "a_ofr.json": {"v": "ofr"}})
|
||
return hf, fixtures, _drop(hf, "a.pdf")
|
||
|
||
def test_directory_appears_only_by_rename_of_complete_set(
|
||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""R-26: до rename каталога `Output/<uuid>/` нет вовсе, набор уже собран во временном каталоге `Tmp`;
|
||
шлюз (`poll`) в это время ничего не видит, после rename — видит оба файла сразу."""
|
||
hf, fixtures, job_id = self._two_file_job(tmp_path)
|
||
gateway_side = LocalHotFolder(hf.root, settle_seconds=0)
|
||
seen: dict[str, Any] = {}
|
||
original_replace = os.replace
|
||
|
||
def spy(src: Path, dst: Path) -> None:
|
||
if Path(dst) == hf.answer_dir(job_id):
|
||
seen["output_entries_before"] = sorted(p.name for p in hf.output_dir.iterdir())
|
||
seen["poll_before"] = gateway_side.poll(job_id)
|
||
seen["staging_parent"] = Path(src).parent
|
||
seen["staging_files"] = sorted(p.name for p in Path(src).iterdir())
|
||
original_replace(src, dst)
|
||
|
||
monkeypatch.setattr("src.scan_emulator.os.replace", spy)
|
||
process_once(hf.root, fixtures, 0)
|
||
assert seen["output_entries_before"] == []
|
||
assert seen["poll_before"] is None
|
||
assert seen["staging_parent"] == hf.tmp_dir
|
||
assert seen["staging_files"] == [f"{job_id}.json", f"{job_id}_ofr.json"]
|
||
after = gateway_side.poll(job_id)
|
||
assert after is not None
|
||
assert [f.name for f in after] == [f"{job_id}.json", f"{job_id}_ofr.json"]
|
||
assert list(hf.tmp_dir.iterdir()) == [] # временный каталог переименован, не скопирован
|
||
|
||
def test_failure_while_writing_leaves_no_answer_dir_and_no_done(
|
||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
hf, fixtures, job_id = self._two_file_job(tmp_path)
|
||
calls: list[Path] = []
|
||
original = emulator._write_json
|
||
|
||
def flaky(path: Path, answer: dict[str, Any]) -> None:
|
||
calls.append(path)
|
||
if len(calls) == 2:
|
||
raise OSError("диск переполнен")
|
||
original(path, answer)
|
||
|
||
monkeypatch.setattr(emulator, "_write_json", flaky)
|
||
with pytest.raises(OSError, match="диск"):
|
||
process_once(hf.root, fixtures, 0)
|
||
assert len(calls) == 2
|
||
assert not hf.answer_dir(job_id).exists()
|
||
assert list(hf.tmp_dir.iterdir()) == []
|
||
assert (hf.input_dir / f"{job_id}_a.pdf").is_file()
|
||
assert not (hf.input_dir / f"{job_id}_a.pdf.done").exists()
|
||
monkeypatch.undo()
|
||
assert process_once(hf.root, fixtures, 0) == [job_id] # следующий проход обрабатывает задание целиком
|
||
assert _names(hf, job_id) == [f"{job_id}.json", f"{job_id}_ofr.json"]
|
||
|
||
def test_stale_staging_dir_is_discarded(self, tmp_path: Path) -> None:
|
||
hf, fixtures, job_id = self._two_file_job(tmp_path)
|
||
stale = hf.tmp_dir / f".{job_id}.answer"
|
||
stale.mkdir(parents=True)
|
||
(stale / "junk.json").write_text("{}")
|
||
process_once(hf.root, fixtures, 0)
|
||
assert _names(hf, job_id) == [f"{job_id}.json", f"{job_id}_ofr.json"]
|
||
assert list(hf.tmp_dir.iterdir()) == []
|
||
|
||
def test_existing_answer_dir_is_replaced(self, tmp_path: Path) -> None:
|
||
hf, fixtures, job_id = self._two_file_job(tmp_path)
|
||
hf.answer_dir(job_id).mkdir(parents=True)
|
||
(hf.answer_dir(job_id) / "old.json").write_text("{}")
|
||
process_once(hf.root, fixtures, 0)
|
||
assert _names(hf, job_id) == [f"{job_id}.json", f"{job_id}_ofr.json"]
|
||
|
||
def test_custom_tmp_dir_name(self, tmp_path: Path) -> None:
|
||
root = tmp_path / "hf"
|
||
root.mkdir()
|
||
hf = LocalHotFolder(root, tmp_dir="Work")
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}})
|
||
job_id = _drop(hf, "a.pdf")
|
||
assert process_once(root, fixtures, 0, tmp_dir="Work") == [job_id]
|
||
assert (root / "Work").is_dir()
|
||
assert not (root / "Tmp").exists() or list((root / "Tmp").iterdir()) == []
|
||
|
||
def test_gateway_poll_waits_for_settle_then_returns_whole_set(self, tmp_path: Path) -> None:
|
||
"""Сквозной «put -> эмулятор -> poll (settle, ложные часы) -> merge_answers -> NullNormalizer» на `aep_3`."""
|
||
if not REAL_FIXTURES.is_dir():
|
||
pytest.skip("fixtures недоступен: положите репозиторий fixtures рядом с workspace")
|
||
root = tmp_path / "hotfolder"
|
||
root.mkdir()
|
||
now = [100.0]
|
||
hf = LocalHotFolder(root, settle_seconds=2, clock=lambda: now[0])
|
||
job_id = _uid()
|
||
assert hf.poll(job_id) is None
|
||
hf.put(inbox_name(job_id, "aep_3.pdf"), b"%PDF-1.4")
|
||
assert process_once(root, REAL_FIXTURES, 0) == [job_id]
|
||
assert hf.poll(job_id) is None # набор впервые замечен: settle отсчитывается с этого опроса
|
||
now[0] += 1.9
|
||
assert hf.poll(job_id) is None
|
||
now[0] += 0.2
|
||
files = hf.poll(job_id)
|
||
assert files is not None
|
||
assert [f.name for f in files] == [f"{job_id}.json", f"{job_id}_ofr.json"]
|
||
merged = merge_answers(job_id, files)
|
||
outcome = NullNormalizer().normalize(merged)
|
||
assert outcome.error is not None
|
||
assert outcome.error.code == "NOT_IMPLEMENTED"
|
||
assert outcome.warnings is not None
|
||
assert len(outcome.warnings) == 1
|
||
assert outcome.warnings[0].startswith(f"{job_id}_ofr.json: INVALID_DOCUMENT: ")
|
||
|
||
|
||
class TestSettings:
|
||
def test_defaults(self, tmp_path: Path) -> None:
|
||
settings = EmulatorSettings.from_env({"SCAN_HOTFOLDER_PATH": str(tmp_path)})
|
||
assert settings.hotfolder_path == tmp_path
|
||
assert (settings.input_dir, settings.output_dir) == ("Input", "Output")
|
||
assert settings.delay == 2
|
||
assert settings.poll_interval == 1
|
||
assert settings.tmp_dir == "Tmp"
|
||
assert settings.fixtures_dir == Path("../../fixtures/ocr")
|
||
|
||
def test_overrides(self, tmp_path: Path) -> None:
|
||
settings = EmulatorSettings.from_env(
|
||
{
|
||
"SCAN_HOTFOLDER_PATH": str(tmp_path),
|
||
"SCAN_INPUT_DIR": "In",
|
||
"SCAN_OUTPUT_DIR": "Out",
|
||
"SCAN_TMP_DIR": "Work",
|
||
"SCAN_FIXTURES_DIR": "fx",
|
||
"SCAN_EMULATOR_DELAY": "0.5",
|
||
"SCAN_EMULATOR_POLL_SECONDS": "0.1",
|
||
}
|
||
)
|
||
assert (settings.input_dir, settings.output_dir) == ("In", "Out")
|
||
assert settings.tmp_dir == "Work"
|
||
assert settings.fixtures_dir == Path("fx")
|
||
assert (settings.delay, settings.poll_interval) == (0.5, 0.1)
|
||
|
||
def test_path_is_required(self) -> None:
|
||
with pytest.raises(ValueError, match="SCAN_HOTFOLDER_PATH"):
|
||
EmulatorSettings.from_env({})
|
||
|
||
@pytest.mark.parametrize(
|
||
("name", "value"),
|
||
[("SCAN_EMULATOR_DELAY", "-1"), ("SCAN_EMULATOR_DELAY", "abc"), ("SCAN_EMULATOR_POLL_SECONDS", "0")],
|
||
)
|
||
def test_invalid_numbers(self, tmp_path: Path, name: str, value: str) -> None:
|
||
with pytest.raises(ValueError, match=name):
|
||
EmulatorSettings.from_env({"SCAN_HOTFOLDER_PATH": str(tmp_path), name: value})
|
||
|
||
|
||
def _settings(tmp_path: Path, fixtures: Path, *, poll: float = 0.01) -> EmulatorSettings:
|
||
return EmulatorSettings(
|
||
hotfolder_path=tmp_path / "hotfolder",
|
||
input_dir="Input",
|
||
output_dir="Output",
|
||
fixtures_dir=fixtures,
|
||
delay=0,
|
||
poll_interval=poll,
|
||
)
|
||
|
||
|
||
class TestRun:
|
||
def test_startup_log_and_processing(self, tmp_path: Path, log_messages: list[str]) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}, "b.json": {"v": 2}})
|
||
(fixtures / "readme.txt").write_text("не фикстура")
|
||
job_id = _drop(hf, "a.pdf")
|
||
run(_settings(tmp_path, fixtures), threading.Event(), max_iterations=2)
|
||
assert hf.answer_path(job_id).is_file()
|
||
startup = "\n".join(log_messages)
|
||
assert str(hf.root) in startup
|
||
assert str(hf.input_dir) in startup
|
||
assert str(hf.output_dir) in startup
|
||
assert str(fixtures) in startup
|
||
assert any("фикстур: 2" in m for m in log_messages)
|
||
|
||
def test_missing_fixtures_dir_warns_and_continues(self, tmp_path: Path, log_messages: list[str]) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
missing = tmp_path / "нет"
|
||
job_id = _drop(hf, "a.pdf")
|
||
run(_settings(tmp_path, missing), threading.Event(), max_iterations=1)
|
||
assert any(m.startswith("WARNING|") and str(missing) in m for m in log_messages)
|
||
assert _answer(hf, job_id)["error"]["code"] == "INVALID_DOCUMENT"
|
||
|
||
def test_stop_event_ends_loop_without_waiting(self, tmp_path: Path) -> None:
|
||
_hotfolder(tmp_path)
|
||
stop = threading.Event()
|
||
stop.set()
|
||
run(_settings(tmp_path, _fixtures_dir(tmp_path), poll=60), stop) # вернётся сразу, а не через 60 с
|
||
|
||
def test_loop_survives_pass_error(
|
||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, log_messages: list[str]
|
||
) -> None:
|
||
_hotfolder(tmp_path)
|
||
calls: list[int] = []
|
||
|
||
def boom(*args: object, **kwargs: object) -> list[str]:
|
||
calls.append(1)
|
||
raise OSError("сбой")
|
||
|
||
monkeypatch.setattr("src.scan_emulator.process_once", boom)
|
||
run(_settings(tmp_path, _fixtures_dir(tmp_path)), threading.Event(), max_iterations=2)
|
||
assert len(calls) == 2
|
||
assert any(m.startswith("ERROR|") for m in log_messages)
|
||
|
||
|
||
class TestMain:
|
||
def test_no_hotfolder_path_exits_nonzero(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||
monkeypatch.delenv("SCAN_HOTFOLDER_PATH", raising=False)
|
||
monkeypatch.setenv("ENV_FILE", str(tmp_path / "нет.env"))
|
||
assert main() != 0
|
||
|
||
def test_hotfolder_root_missing_exits_nonzero(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||
monkeypatch.setenv("SCAN_HOTFOLDER_PATH", str(tmp_path / "нет-корня"))
|
||
assert main() != 0
|
||
|
||
def test_clean_exit_on_stop(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}})
|
||
monkeypatch.setenv("SCAN_HOTFOLDER_PATH", str(hf.root))
|
||
monkeypatch.setenv("SCAN_FIXTURES_DIR", str(fixtures))
|
||
monkeypatch.setenv("SCAN_EMULATOR_DELAY", "0")
|
||
monkeypatch.setenv("SCAN_EMULATOR_POLL_SECONDS", "0.01")
|
||
job_id = _drop(hf, "a.pdf")
|
||
stop = threading.Event()
|
||
threading.Timer(0.3, stop.set).start()
|
||
assert main(stop) == 0
|
||
assert hf.answer_path(job_id).is_file()
|
||
|
||
def test_module_starts_without_gateway_settings(self, tmp_path: Path) -> None:
|
||
"""C-1: `python -m src.scan_emulator` не читает настройки шлюза — импорт пакета `src` не создаёт приложение.
|
||
|
||
Окружение — только PATH, пустой ENV_FILE и несуществующий корень: ожидается штатный отказ эмулятора (код 2,
|
||
сообщение про корень горячей папки), а не `ValidationError` настроек шлюза (`OPENBAO`).
|
||
"""
|
||
env = {
|
||
"PATH": os.environ.get("PATH", ""),
|
||
"ENV_FILE": str(tmp_path / "none.env"),
|
||
"SCAN_HOTFOLDER_PATH": str(tmp_path / "missing"),
|
||
}
|
||
proc = subprocess.run(
|
||
[sys.executable, "-m", "src.scan_emulator"],
|
||
cwd=API_DIR,
|
||
env=env,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=60,
|
||
check=False,
|
||
)
|
||
assert proc.returncode == 2, proc.stderr
|
||
assert "корень горячей папки" in proc.stderr
|
||
assert "ValidationError" not in proc.stderr
|
||
assert "OPENBAO" not in proc.stderr
|