793516fa61
T14.2(a): Добавлен тест с шпионом на build_index, проверяющий что индекс не пересчитывается повторно на каждом проходе при неизменных фикстурах. 5 проходов — ровно 1 вызов build_index (при инициализации run), остальные используют refresh. T14.2(b): Добавлен тест корректной обработки появления каталога фикстур после стартапа — run не выбрасывает исключение и может использовать фикстуры после их создания. T14.1 remainder (double refresh): Исправлено поведение process_once: если index=None, то build_index один раз; если index передан, то только refresh. Обновлена docstring процесса. T14.2(c): Добавлен мутационный тест test_skip_suffixes_not_logged_as_regex_mismatch, проверяющий что файлы с _SKIP_SUFFIXES пропускаются молча и НЕ логируются как не совпадающие с шаблоном. T14.4: Переписан test_uppercase_extension_accepted — теперь пишет файл напрямую .PDF и проверяет нормализацию в .pdf.done, вместо использования _drop которая нормализует расширение. Дополнительно: - Обновлена docstring run() — точное описание: один refresh в конце каждого прохода - Удалено "(Review Focus 5)" из docstring test_ignores_part_tmp_done - Добавлен @pytest.mark.skipif для chmod теста при запуске от root 915 тестов пройдено, 10 деселектировано (10 избранных тестов требуют fixtures). Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
873 lines
42 KiB
Python
873 lines
42 KiB
Python
"""Эмулятор ИС Сканирование (D-3, spec scan-emulator, slice-3a-ocr-normalizer).
|
||
|
||
Сценарии spec «Точное имя», «Суффикс», «Несколько файлов на одно задание», «Нет пары» — на копиях реальных
|
||
`../fixtures/ocr/*.json` (маркер `fixtures`); остальное — на синтетических фикстурах в `tmp_path` с вычисляемым
|
||
содержимым, соответствующим по sha256.
|
||
"""
|
||
|
||
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
|
||
from src.gateway.merge import merge_answers
|
||
from src.normalizer import NullNormalizer
|
||
from src.scan_emulator import EmulatorSettings, build_index, process_once, resolve_stem, 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():
|
||
if name.lower().endswith(".json"):
|
||
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 | None = None) -> str:
|
||
"""Положить файл во входящий каталог в формате `<uuid>.<ext>`, вернуть uuid.
|
||
|
||
По умолчанию содержимое берётся из одноимённой фикстуры в фиксутрах (если она существует);
|
||
иначе используется переданное data или b"%PDF <name>".
|
||
"""
|
||
job_id = _uid()
|
||
ext = Path(name).suffix.lstrip(".").lower() or "pdf"
|
||
if data is None:
|
||
data = b"%PDF " + name.encode()
|
||
hf.put(f"{job_id}.{ext}", 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("*"):
|
||
if source.is_file() and (source.suffix.lower() in (".json", ".pdf")):
|
||
shutil.copy(source, target / source.name)
|
||
return target
|
||
|
||
|
||
class TestHashIndex:
|
||
def test_resolves_by_content(self, tmp_path: Path) -> None:
|
||
fixtures = _fixtures_dir(
|
||
tmp_path, {"ok.json": {"file_uid": "x", "error": {"code": "", "message": ""}, "data_ocr": {}}}
|
||
)
|
||
(fixtures / "ok.pdf").write_bytes(b"%PDF ok")
|
||
index = build_index(fixtures)
|
||
assert resolve_stem(b"%PDF ok", index, "") == ("ok", "hash")
|
||
assert resolve_stem(b"other", index, "") == (None, "none")
|
||
assert resolve_stem(b"other", index, "ok") == ("ok", "default")
|
||
|
||
def test_refresh_only_when_listing_changes(self, tmp_path: Path) -> None:
|
||
fixtures = _fixtures_dir(tmp_path)
|
||
(fixtures / "a.pdf").write_bytes(b"a")
|
||
index = build_index(fixtures)
|
||
assert index.refresh(fixtures) is index
|
||
(fixtures / "b.pdf").write_bytes(b"b")
|
||
refreshed = index.refresh(fixtures)
|
||
assert refreshed is not index and set(refreshed.by_hash.values()) == {"a", "b"}
|
||
|
||
def test_ignores_part_tmp_done(self, tmp_path: Path) -> None:
|
||
"""Временные и обработанные файлы не обрабатываются."""
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path)
|
||
hf.input_dir.mkdir(exist_ok=True)
|
||
hf.tmp_dir.mkdir(exist_ok=True)
|
||
job = _uid()
|
||
(hf.tmp_dir / f".{job}.pdf.part").write_bytes(b"x")
|
||
(hf.input_dir / f"{job}.pdf.done").write_bytes(b"x")
|
||
(hf.input_dir / f"{job}.pdf.tmp").write_bytes(b"x")
|
||
assert process_once(hf.root, fixtures, 0) == []
|
||
assert not hf.output_dir.exists() or list(hf.output_dir.iterdir()) == []
|
||
|
||
def test_default_set_on_miss(self, tmp_path: Path, log_messages: list[str]) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(
|
||
tmp_path, {"dflt.json": {"file_uid": "x", "error": {"code": "", "message": ""}, "data_ocr": {}}}
|
||
)
|
||
(fixtures / "dflt.pdf").write_bytes(b"dflt data")
|
||
job = _drop(hf, "any.pdf", data=b"unknown bytes")
|
||
assert process_once(hf.root, fixtures, 0, default="dflt") == [job]
|
||
assert _answer(hf, job)["error"]["code"] == ""
|
||
# M-6: Log message uses "по умолчанию"
|
||
assert any("по умолчанию" in m for m in log_messages)
|
||
|
||
def test_miss_without_default_is_invalid_document(self, tmp_path: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path)
|
||
job = _drop(hf, "any.pdf", data=b"unknown bytes")
|
||
assert process_once(hf.root, fixtures, 0) == [job]
|
||
answer = _answer(hf, job)
|
||
assert answer["error"]["code"] == "INVALID_DOCUMENT" and "нет пары по содержимому" in answer["error"]["message"]
|
||
|
||
def test_name_does_not_matter(self, tmp_path: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(
|
||
tmp_path, {"ok.json": {"file_uid": "x", "error": {"code": "", "message": ""}, "data_ocr": {}}}
|
||
)
|
||
(fixtures / "ok.pdf").write_bytes(b"%PDF ok")
|
||
job = _drop(hf, "whatever.PDF", data=b"%PDF ok")
|
||
assert process_once(hf.root, fixtures, 0) == [job]
|
||
assert (hf.input_dir / f"{job}.pdf.done").exists()
|
||
assert _answer(hf, job)["file_uid"] == job
|
||
|
||
@pytest.mark.skipif(os.geteuid() == 0, reason="chmod test not applicable when running as root")
|
||
def test_upload_file_read_error_continues(self, tmp_path: Path, log_messages: list[str]) -> None:
|
||
"""T14.3: OSError при чтении входящего файла логируется одной строкой, проход продолжается."""
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}})
|
||
(fixtures / "a.pdf").write_bytes(b"a data")
|
||
hf.input_dir.mkdir()
|
||
|
||
job_id = _uid()
|
||
bad_file = hf.input_dir / f"{job_id}.pdf"
|
||
bad_file.write_bytes(b"test")
|
||
|
||
# Сделать файл нечитаемым
|
||
os.chmod(bad_file, 0o000)
|
||
|
||
try:
|
||
job_id2 = _drop(hf, "a.pdf", data=b"a data")
|
||
result = process_once(hf.root, fixtures, 0)
|
||
|
||
# Первый файл должен быть пропущен с логом, второй обработан
|
||
assert job_id2 in result
|
||
has_warning = any(
|
||
m.startswith("WARNING|") and "ошибка чтения файла" in m and bad_file.name in m for m in log_messages
|
||
)
|
||
assert has_warning
|
||
finally:
|
||
# Восстановить права для очистки
|
||
os.chmod(bad_file, 0o644)
|
||
|
||
def test_uppercase_extension_accepted(self, tmp_path: Path) -> None:
|
||
"""M-4: Файл <uuid>.PDF (верхний регистр) принимается и переименовывается в .pdf.done."""
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}})
|
||
(fixtures / "a.pdf").write_bytes(b"a data")
|
||
# Писать файл напрямую, чтобы проверить обработку .PDF (hf.put нормализует расширение)
|
||
job_id = str(uuid.uuid4())
|
||
hf.input_dir.mkdir(parents=True, exist_ok=True)
|
||
(hf.input_dir / f"{job_id}.PDF").write_bytes(b"a data")
|
||
|
||
assert process_once(hf.root, fixtures, 0) == [job_id]
|
||
# Файл переименован в .pdf.done (расширение нормализовано в lowercase)
|
||
assert (hf.input_dir / f"{job_id}.pdf.done").exists()
|
||
# Ответ создан
|
||
assert hf.answer_path(job_id).exists()
|
||
|
||
def test_ignores_part_tmp_done_robust(self, tmp_path: Path) -> None:
|
||
"""T14.2c: Усиленный тест — падает при отсутствии skip-списка."""
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}})
|
||
(fixtures / "a.pdf").write_bytes(b"a data")
|
||
hf.input_dir.mkdir(exist_ok=True)
|
||
hf.tmp_dir.mkdir(exist_ok=True)
|
||
|
||
job = _uid()
|
||
# Файлы для пропуска
|
||
(hf.input_dir / f"{job}.pdf.done").write_bytes(b"skip")
|
||
(hf.input_dir / f"{job}.pdf.tmp").write_bytes(b"skip")
|
||
(hf.input_dir / f"{job}.pdf.part").write_bytes(b"skip")
|
||
(hf.tmp_dir / f".{job}.pdf.part").write_bytes(b"skip")
|
||
|
||
# Нормальный файл
|
||
job2 = _drop(hf, "a.pdf", data=b"a data")
|
||
|
||
result = process_once(hf.root, fixtures, 0)
|
||
assert result == [job2]
|
||
|
||
# Все skip-файлы остаются
|
||
assert (hf.input_dir / f"{job}.pdf.done").exists()
|
||
assert (hf.input_dir / f"{job}.pdf.tmp").exists()
|
||
assert (hf.input_dir / f"{job}.pdf.part").exists()
|
||
assert (hf.tmp_dir / f".{job}.pdf.part").exists()
|
||
|
||
|
||
@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)
|
||
pdf_data = (real_fixtures / "aep_5.pdf").read_bytes()
|
||
job_id = _drop(hf, name, data=pdf_data)
|
||
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}.pdf.done").is_file()
|
||
assert not (hf.input_dir / f"{job_id}.pdf").exists()
|
||
|
||
def test_suffix(self, tmp_path: Path, real_fixtures: Path) -> None:
|
||
"""`aep_8.pdf` -> `aep_8_form4.json` как `<uuid>_form4.json` при совпадении содержимого."""
|
||
hf = _hotfolder(tmp_path)
|
||
pdf_data = (real_fixtures / "aep_8.pdf").read_bytes()
|
||
job_id = _drop(hf, "aep_8.pdf", data=pdf_data)
|
||
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}.pdf.done").is_file()
|
||
|
||
def test_two_files_for_one_job(self, tmp_path: Path, real_fixtures: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
pdf_data = (real_fixtures / "aep_3.pdf").read_bytes()
|
||
job_id = _drop(hf, "aep_3.pdf", data=pdf_data)
|
||
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}.pdf.done").is_file()
|
||
|
||
def test_single_success_file(self, tmp_path: Path, real_fixtures: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
pdf_data = (real_fixtures / "aep_6.pdf").read_bytes()
|
||
job_id = _drop(hf, "aep_6.pdf", data=pdf_data)
|
||
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", data=b"unknown content")
|
||
process_once(hf.root, real_fixtures, 0)
|
||
assert _names(hf, job_id) == [f"{job_id}.json"]
|
||
answer = _answer(hf, job_id)
|
||
assert answer["file_uid"] == job_id
|
||
assert answer["error"]["code"] == "INVALID_DOCUMENT"
|
||
assert "нет пары по содержимому" in answer["error"]["message"]
|
||
assert answer["data_ocr"] == EMPTY_DATA_OCR
|
||
assert (hf.input_dir / f"{job_id}.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", data=b"unknown content")
|
||
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"}})
|
||
(fixtures / "a_b.pdf").write_bytes(b"exact content")
|
||
job_id = _drop(hf, "a_b.pdf", data=b"exact content")
|
||
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"}})
|
||
(fixtures / "a.pdf").write_bytes(b"test data")
|
||
job_id = _drop(hf, "a.pdf", data=b"test data")
|
||
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*`: правило `<stem>.json` и `<stem>_*.json`, а не `<stem>*`."""
|
||
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},
|
||
},
|
||
)
|
||
(fixtures / "aep_3.pdf").write_bytes(b"aep3 content")
|
||
job_id = _drop(hf, "aep_3.pdf", data=b"aep3 content")
|
||
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}})
|
||
(fixtures / "a.pdf").write_bytes(b"content")
|
||
job_id = _drop(hf, "a.pdf", data=b"content")
|
||
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": []}})
|
||
(fixtures / "a.pdf").write_bytes(b"a data")
|
||
(fixtures / "b.pdf").write_bytes(b"b data")
|
||
first, second = _drop(hf, "a.pdf", data=b"a data"), _drop(hf, "b.pdf", data=b"b data")
|
||
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}})
|
||
(fixtures / "scan").write_bytes(b"no ext")
|
||
job_id = _drop(hf, "scan", data=b"no ext")
|
||
process_once(hf.root, fixtures, 0)
|
||
assert _answer(hf, job_id) == {"file_uid": job_id, "v": 1}
|
||
|
||
def test_non_ascii_not_escaped(self, tmp_path: Path) -> None:
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"text": "Баланс"}})
|
||
(fixtures / "a.pdf").write_bytes(b"cyrillic")
|
||
job_id = _drop(hf, "a.pdf", data=b"cyrillic")
|
||
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", data=b"content")
|
||
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})
|
||
(fixtures / "a.pdf").write_bytes(b"broken")
|
||
job_id = _drop(hf, "a.pdf", data=b"broken")
|
||
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.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}.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}})
|
||
(fixtures / "a.pdf").write_bytes(b"a content")
|
||
ids = {_drop(hf, "a.pdf", data=b"a content"), _drop(hf, "a.pdf", data=b"a content")}
|
||
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}})
|
||
(fixtures / "a.pdf").write_bytes(b"a content")
|
||
job_id = _drop(hf, "a.pdf", data=b"a content")
|
||
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}.pdf.done",
|
||
f".{job_id}.pdf",
|
||
"a.pdf",
|
||
"not-a-uuid.pdf",
|
||
f"{job_id}a.pdf",
|
||
f"{job_id}.pdf.part",
|
||
f"{job_id}.pdf.tmp",
|
||
]
|
||
for name in ignored:
|
||
(hf.input_dir / name).write_bytes(b"x")
|
||
(hf.input_dir / f"{_uid()}.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" 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}})
|
||
(fixtures / "a.pdf").write_bytes(b"a data")
|
||
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", data=b"a data")
|
||
_drop(hf, "a.pdf", data=b"a data")
|
||
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}})
|
||
(fixtures / "a.pdf").write_bytes(b"a data")
|
||
job_id = _drop(hf, "a.pdf", data=b"a data")
|
||
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}})
|
||
(fixtures / "a.pdf").write_bytes(b"a data")
|
||
job_id = _drop(hf, "a.pdf", data=b"a data")
|
||
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; готовность набора без ожидания (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"}]}})
|
||
(fixtures / "report.pdf").write_bytes(b"%PDF-1.4 report")
|
||
job_id = _uid()
|
||
assert hf.poll(job_id) is None
|
||
hf.put(f"{job_id}.pdf", b"%PDF-1.4 report")
|
||
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"}})
|
||
(fixtures / "a.pdf").write_bytes(b"two file content")
|
||
return hf, fixtures, _drop(hf, "a.pdf", data=b"two file content")
|
||
|
||
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}.pdf").is_file()
|
||
assert not (hf.input_dir / f"{job_id}.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}})
|
||
(fixtures / "a.pdf").write_bytes(b"custom work")
|
||
job_id = _drop(hf, "a.pdf", data=b"custom work")
|
||
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()) == []
|
||
|
||
@pytest.mark.fixtures
|
||
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])
|
||
pdf_data = (REAL_FIXTURES / "aep_3.pdf").read_bytes()
|
||
job_id = _uid()
|
||
assert hf.poll(job_id) is None
|
||
hf.put(f"{job_id}.pdf", pdf_data)
|
||
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")
|
||
assert settings.default == ""
|
||
|
||
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",
|
||
"SCAN_EMULATOR_DEFAULT": "aep_6",
|
||
}
|
||
)
|
||
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)
|
||
assert settings.default == "aep_6"
|
||
|
||
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 / "a.pdf").write_bytes(b"a data")
|
||
(fixtures / "b.pdf").write_bytes(b"b data")
|
||
job_id = _drop(hf, "a.pdf", data=b"a data")
|
||
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_build_index_not_called_repeatedly_when_fixtures_unchanged(
|
||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""T14.2(a): build_index() не вызывается повторно на каждом проходе, если файлы в фикстурах не меняются.
|
||
|
||
Минимум 5 проходов (max_iterations=5) без изменения списка файлов должны привести к ровно 1 вызову
|
||
build_index() (при инициализации run), остальные проходы используют refresh().
|
||
"""
|
||
_hotfolder(tmp_path) # Создаём структуру горячей папки
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}})
|
||
(fixtures / "a.pdf").write_bytes(b"a data")
|
||
|
||
build_index_calls: list[int] = []
|
||
original_build_index = emulator.build_index
|
||
|
||
def spy_build_index(*args: object, **kwargs: object) -> emulator.FixtureIndex:
|
||
build_index_calls.append(1)
|
||
return original_build_index(*args, **kwargs)
|
||
|
||
monkeypatch.setattr(emulator, "build_index", spy_build_index)
|
||
run(_settings(tmp_path, fixtures), threading.Event(), max_iterations=5)
|
||
|
||
# build_index вызван ровно один раз (при инициализации run)
|
||
assert len(build_index_calls) == 1
|
||
|
||
def test_run_with_missing_then_appearing_fixtures_dir(self, tmp_path: Path) -> None:
|
||
"""T14.2(b): run() обрабатывает появление каталога фикстур ПОСЛЕ стартапа без ошибок.
|
||
|
||
Стартуем с несуществующим каталогом фикстур, затем каталог создаётся.
|
||
Эмулятор не должен выбросить исключение, а индекс должен обновиться.
|
||
"""
|
||
_hotfolder(tmp_path) # Создаём структуру горячей папки
|
||
fixtures = tmp_path / "fixtures"
|
||
|
||
# Стартуем с несуществующим каталогом - не должно быть ошибки
|
||
stop = threading.Event()
|
||
stop.set() # Стоп после первого прохода (в run цикл ждёт stop.is_set)
|
||
run(_settings(tmp_path, fixtures), stop, max_iterations=1)
|
||
|
||
# После успешного запуска с отсутствующим каталогом создаём его
|
||
fixtures.mkdir(exist_ok=True)
|
||
(fixtures / "a.json").write_text('{"v": 1}')
|
||
(fixtures / "a.pdf").write_bytes(b"test data")
|
||
|
||
# Второй запуск должен использовать новый каталог
|
||
stop.clear()
|
||
stop.set()
|
||
run(_settings(tmp_path, fixtures), stop, max_iterations=1)
|
||
# Если мы сюда дошли без исключения - тест прошёл
|
||
|
||
def test_skip_suffixes_not_logged_as_regex_mismatch(
|
||
self, tmp_path: Path, log_messages: list[str], monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""T14.2(c): Файлы в _SKIP_SUFFIXES пропускаются молча, не логируются как не совпадающие с шаблоном.
|
||
|
||
Мутационный тест: если убрать проверку _SKIP_SUFFIXES, тест сломается (файлы будут залогированы).
|
||
"""
|
||
hf = _hotfolder(tmp_path)
|
||
fixtures = _fixtures_dir(tmp_path, {"a.json": {"v": 1}})
|
||
(fixtures / "a.pdf").write_bytes(b"a data")
|
||
hf.input_dir.mkdir(exist_ok=True)
|
||
hf.tmp_dir.mkdir(exist_ok=True)
|
||
|
||
job_id = _uid()
|
||
# Кладём файлы с _SKIP_SUFFIXES
|
||
(hf.input_dir / f"{job_id}.pdf.done").write_bytes(b"skip")
|
||
(hf.input_dir / f"{job_id}.pdf.tmp").write_bytes(b"skip")
|
||
(hf.input_dir / f"{job_id}.pdf.part").write_bytes(b"skip")
|
||
|
||
# Кладём нормальный файл для проверки, что обработка работает
|
||
job_id2 = _drop(hf, "a.pdf", data=b"a data")
|
||
|
||
result = process_once(hf.root, fixtures, 0)
|
||
assert job_id2 in result
|
||
|
||
# Логи НЕ должны содержать сообщения о "не по шаблону" для файлов с _SKIP_SUFFIXES
|
||
foreign_warnings = [m for m in log_messages if m.startswith("WARNING|") and "не по шаблону" in m]
|
||
assert len(foreign_warnings) == 0 # Ни одного предупреждения про "не по шаблону"
|
||
|
||
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", data=b"unknown")
|
||
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)
|
||
|
||
|
||
def test_settings_default_from_env(tmp_path: Path) -> None:
|
||
settings = EmulatorSettings.from_env({"SCAN_HOTFOLDER_PATH": str(tmp_path), "SCAN_EMULATOR_DEFAULT": "aep_6"})
|
||
assert settings.default == "aep_6"
|
||
|
||
|
||
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}})
|
||
(fixtures / "a.pdf").write_bytes(b"a data")
|
||
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", data=b"a data")
|
||
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
|