5fb046fced
T8.1: inbox_name заменяет `: < > " | ? *` и управляющие символы на `_`, убирает завершающие точки/пробелы; ValueError только если имени не осталось. validate_name/put остаются строгими. T8.2: SmbHotFolder.poll возвращает None на STATUS_SHARING_VIOLATION/STATUS_DELETE_PENDING (ИС ещё пишет ответ), монопольное открытие сохранено. T8.3: при ошибке mkdir, кроме EEXIST/ENOENT, put продолжает, если smbclient.path.isdir(Input); иначе исходная ошибка. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
278 lines
11 KiB
Python
278 lines
11 KiB
Python
"""Горячая папка в обычном каталоге (spec image-recognition-gateway, D-1, R-4/R-5, slice-0-skeleton).
|
||
|
||
Быстрые тесты на `tmp_path`: атомарная запись входящего файла, опрос ответа ИС, проверка здоровья, валидация имён.
|
||
"""
|
||
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from src.gateway.hotfolder import HotFolder, LocalHotFolder, inbox_name, validate_name
|
||
|
||
|
||
def _job_id() -> str:
|
||
return str(uuid.uuid4())
|
||
|
||
|
||
def _write_answer(hf: LocalHotFolder, job_id: str, data: bytes) -> Path:
|
||
path = hf.answer_path(job_id)
|
||
path.parent.mkdir(parents=True)
|
||
path.write_bytes(data)
|
||
return path
|
||
|
||
|
||
class TestInboxName:
|
||
def test_joins_job_id_and_name(self) -> None:
|
||
job_id = _job_id()
|
||
assert inbox_name(job_id, "aep_5.PDF") == f"{job_id}_aep_5.PDF"
|
||
|
||
@pytest.mark.parametrize(
|
||
("original", "expected"),
|
||
[
|
||
("../../etc/passwd", "passwd"),
|
||
("dir/sub/report.pdf", "report.pdf"),
|
||
("C:\\scans\\report.pdf", "report.pdf"),
|
||
("/abs/report.pdf", "report.pdf"),
|
||
("..\\..\\report.pdf", "report.pdf"),
|
||
],
|
||
)
|
||
def test_sanitizes_to_basename(self, original: str, expected: str) -> None:
|
||
assert inbox_name("job", original) == f"job_{expected}"
|
||
|
||
@pytest.mark.parametrize("original", ["", "/", "..", ".", "a/..", "dir/", "\\", "...", ". .", " ", "dir/. "])
|
||
def test_rejects_unusable_names(self, original: str) -> None:
|
||
with pytest.raises(ValueError, match="имя файла"):
|
||
inbox_name("job", original)
|
||
|
||
@pytest.mark.parametrize(
|
||
("original", "expected"),
|
||
[
|
||
("a:b?.pdf", "a_b_.pdf"),
|
||
("Отчёт: 2024.pdf", "Отчёт_ 2024.pdf"),
|
||
('Баланс "итог".pdf', "Баланс _итог_.pdf"),
|
||
("форма?.pdf", "форма_.pdf"),
|
||
("a|b<c>*.pdf", "a_b_c__.pdf"),
|
||
("C:report.pdf", "C_report.pdf"),
|
||
("a\x01b.pdf", "a_b.pdf"),
|
||
("bad\x00name.pdf", "bad_name.pdf"),
|
||
("tab\tname.pdf", "tab_name.pdf"),
|
||
],
|
||
)
|
||
def test_replaces_chars_illegal_on_windows(self, original: str, expected: str) -> None:
|
||
"""Имя от пользователя законно на Linux/S3/macOS: заменяем, а не отвергаем (контракт: только minLength 1)."""
|
||
assert inbox_name("job", original) == f"job_{expected}"
|
||
|
||
@pytest.mark.parametrize(
|
||
("original", "expected"),
|
||
[("report.pdf.", "report.pdf"), ("report.pdf . ", "report.pdf"), ("report .pdf", "report .pdf")],
|
||
)
|
||
def test_strips_trailing_dots_and_spaces(self, original: str, expected: str) -> None:
|
||
assert inbox_name("job", original) == f"job_{expected}"
|
||
|
||
def test_keeps_spaces_and_cyrillic(self) -> None:
|
||
job_id = _job_id()
|
||
assert inbox_name(job_id, "Отчёт за 2024 год (копия).PDF") == f"{job_id}_Отчёт за 2024 год (копия).PDF"
|
||
|
||
@pytest.mark.parametrize(
|
||
"original",
|
||
[
|
||
"a:b?.pdf",
|
||
'Баланс "итог".pdf',
|
||
"Отчёт за 2024 год (копия).PDF",
|
||
".hidden.pdf",
|
||
"..pdf",
|
||
"x\x00\x1f<>|*?.pdf. ",
|
||
"C:\\dir\\..\\скан.png",
|
||
" leading.pdf",
|
||
],
|
||
)
|
||
def test_result_always_passes_validate_name_and_put(self, tmp_path: Path, original: str) -> None:
|
||
hf = LocalHotFolder(tmp_path)
|
||
name = inbox_name(_job_id(), original)
|
||
validate_name(name)
|
||
hf.put(name, b"x")
|
||
assert (hf.input_dir / name).read_bytes() == b"x"
|
||
|
||
def test_result_is_accepted_by_put(self, tmp_path: Path) -> None:
|
||
hf = LocalHotFolder(tmp_path)
|
||
name = inbox_name(_job_id(), "../../scan.pdf")
|
||
hf.put(name, b"x")
|
||
assert (hf.input_dir / name).read_bytes() == b"x"
|
||
|
||
|
||
class TestPaths:
|
||
def test_default_layout(self, tmp_path: Path) -> None:
|
||
hf = LocalHotFolder(tmp_path)
|
||
job_id = _job_id()
|
||
assert hf.root == tmp_path
|
||
assert hf.input_dir == tmp_path / "Input"
|
||
assert hf.output_dir == tmp_path / "Output"
|
||
assert hf.answer_path(job_id) == tmp_path / "Output" / job_id / f"{job_id}.json"
|
||
|
||
def test_custom_dir_names_and_str_root(self, tmp_path: Path) -> None:
|
||
hf = LocalHotFolder(str(tmp_path), input_dir="Входящие", output_dir="Исходящие")
|
||
assert hf.root == tmp_path
|
||
assert hf.input_dir == tmp_path / "Входящие"
|
||
assert hf.output_dir == tmp_path / "Исходящие"
|
||
assert hf.answer_path("abc") == tmp_path / "Исходящие" / "abc" / "abc.json"
|
||
|
||
@pytest.mark.parametrize("job_id", ["", ".", "..", "../x", "a/b", "a\\b", "a:b", "a?b", "a*b", "a\x00b", "a\nb"])
|
||
def test_answer_path_rejects_unsafe_job_id(self, tmp_path: Path, job_id: str) -> None:
|
||
with pytest.raises(ValueError, match="job_id"):
|
||
LocalHotFolder(tmp_path).answer_path(job_id)
|
||
|
||
def test_implements_protocol(self, tmp_path: Path) -> None:
|
||
hf: HotFolder = LocalHotFolder(tmp_path)
|
||
assert hf.healthy() is True
|
||
|
||
|
||
class TestPut:
|
||
def test_writes_bytes_under_inbox_name(self, tmp_path: Path) -> None:
|
||
hf = LocalHotFolder(tmp_path)
|
||
name = inbox_name(_job_id(), "aep_5.pdf")
|
||
data = bytes(range(256)) * 10
|
||
hf.put(name, data)
|
||
assert (tmp_path / "Input" / name).read_bytes() == data
|
||
|
||
def test_leaves_no_part_files(self, tmp_path: Path) -> None:
|
||
hf = LocalHotFolder(tmp_path)
|
||
name = inbox_name(_job_id(), "aep_5.pdf")
|
||
hf.put(name, b"data")
|
||
assert [p.name for p in hf.input_dir.iterdir()] == [name]
|
||
|
||
def test_creates_input_dir_only(self, tmp_path: Path) -> None:
|
||
hf = LocalHotFolder(tmp_path)
|
||
assert not hf.input_dir.exists()
|
||
hf.put("a.pdf", b"1")
|
||
assert hf.input_dir.is_dir()
|
||
assert not hf.output_dir.exists()
|
||
|
||
def test_existing_input_dir_is_fine(self, tmp_path: Path) -> None:
|
||
(tmp_path / "Input").mkdir()
|
||
LocalHotFolder(tmp_path).put("a.pdf", b"1")
|
||
assert (tmp_path / "Input" / "a.pdf").read_bytes() == b"1"
|
||
|
||
def test_missing_root_raises_and_creates_nothing(self, tmp_path: Path) -> None:
|
||
root = tmp_path / "absent"
|
||
with pytest.raises(FileNotFoundError):
|
||
LocalHotFolder(root).put("a.pdf", b"1")
|
||
assert not root.exists()
|
||
assert list(tmp_path.iterdir()) == []
|
||
|
||
def test_overwrites_existing_file(self, tmp_path: Path) -> None:
|
||
hf = LocalHotFolder(tmp_path)
|
||
hf.put("a.pdf", b"old")
|
||
hf.put("a.pdf", b"new")
|
||
assert (hf.input_dir / "a.pdf").read_bytes() == b"new"
|
||
assert [p.name for p in hf.input_dir.iterdir()] == ["a.pdf"]
|
||
|
||
def test_empty_data_is_written(self, tmp_path: Path) -> None:
|
||
hf = LocalHotFolder(tmp_path)
|
||
hf.put("empty.pdf", b"")
|
||
assert (hf.input_dir / "empty.pdf").read_bytes() == b""
|
||
|
||
@pytest.mark.parametrize(
|
||
"name",
|
||
["", "../escape.pdf", "sub/a.pdf", "sub\\a.pdf", "/abs.pdf", ".hidden.pdf", ".a.pdf.part", "..", ".", "a\x00b"],
|
||
)
|
||
def test_rejects_invalid_names(self, tmp_path: Path, name: str) -> None:
|
||
hf = LocalHotFolder(tmp_path)
|
||
with pytest.raises(ValueError, match="имя файла"):
|
||
hf.put(name, b"x")
|
||
assert not hf.input_dir.exists()
|
||
assert not (tmp_path.parent / "escape.pdf").exists()
|
||
|
||
@pytest.mark.parametrize("name", ["a:b.pdf", "a<b.pdf", "a>b.pdf", 'a"b.pdf', "a|b.pdf", "a?b.pdf", "a*b.pdf"])
|
||
def test_rejects_names_illegal_on_windows(self, tmp_path: Path, name: str) -> None:
|
||
hf = LocalHotFolder(tmp_path)
|
||
with pytest.raises(ValueError, match="имя файла"):
|
||
hf.put(name, b"x")
|
||
assert not hf.input_dir.exists()
|
||
|
||
@pytest.mark.parametrize("name", ["a\x01b.pdf", "a\tb.pdf", "a\nb.pdf", "a\x1fb.pdf"])
|
||
def test_rejects_control_chars(self, tmp_path: Path, name: str) -> None:
|
||
with pytest.raises(ValueError, match="имя файла"):
|
||
LocalHotFolder(tmp_path).put(name, b"x")
|
||
|
||
def test_accepts_spaces_cyrillic_and_inbox_names(self, tmp_path: Path) -> None:
|
||
hf = LocalHotFolder(tmp_path)
|
||
names = [inbox_name(_job_id(), "Отчёт за 2024 год (копия).PDF"), inbox_name(_job_id(), "aep_8.pdf"), "a b.pdf"]
|
||
for name in names:
|
||
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:
|
||
hf = LocalHotFolder(tmp_path)
|
||
|
||
def boom(src: object, dst: object) -> None:
|
||
raise OSError("диск недоступен")
|
||
|
||
monkeypatch.setattr("src.gateway.hotfolder.os.replace", boom)
|
||
with pytest.raises(OSError, match="диск недоступен"):
|
||
hf.put("a.pdf", b"data")
|
||
assert list(hf.input_dir.iterdir()) == []
|
||
|
||
|
||
class TestPoll:
|
||
def test_none_until_answer_exists(self, tmp_path: Path) -> None:
|
||
hf = LocalHotFolder(tmp_path)
|
||
job_id = _job_id()
|
||
assert hf.poll(job_id) is None
|
||
(tmp_path / "Output").mkdir()
|
||
assert hf.poll(job_id) is None
|
||
(tmp_path / "Output" / job_id).mkdir()
|
||
assert hf.poll(job_id) is None
|
||
|
||
def test_returns_bytes_when_answer_appears(self, tmp_path: Path) -> None:
|
||
hf = LocalHotFolder(tmp_path)
|
||
job_id = _job_id()
|
||
payload = b'{"file_uid": "x", "data_ocr": []}'
|
||
assert hf.poll(job_id) is None
|
||
_write_answer(hf, job_id, payload)
|
||
assert hf.poll(job_id) == payload
|
||
|
||
def test_other_job_is_unaffected(self, tmp_path: Path) -> None:
|
||
hf = LocalHotFolder(tmp_path)
|
||
job_id, other = _job_id(), _job_id()
|
||
_write_answer(hf, other, b"{}")
|
||
assert hf.poll(job_id) is None
|
||
assert hf.poll(other) == b"{}"
|
||
|
||
def test_missing_root_is_none(self, tmp_path: Path) -> None:
|
||
assert LocalHotFolder(tmp_path / "absent").poll(_job_id()) is None
|
||
|
||
def test_file_removed_between_exists_and_read_is_none(
|
||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
hf = LocalHotFolder(tmp_path)
|
||
job_id = _job_id()
|
||
path = _write_answer(hf, job_id, b"{}")
|
||
|
||
def vanished(self: Path) -> bytes:
|
||
path.unlink()
|
||
raise FileNotFoundError(str(self))
|
||
|
||
monkeypatch.setattr(Path, "read_bytes", vanished)
|
||
assert hf.poll(job_id) is None
|
||
|
||
|
||
class TestHealthy:
|
||
def test_true_for_existing_dir(self, tmp_path: Path) -> None:
|
||
assert LocalHotFolder(tmp_path).healthy() is True
|
||
|
||
def test_false_for_missing_root(self, tmp_path: Path) -> None:
|
||
assert LocalHotFolder(tmp_path / "absent").healthy() is False
|
||
|
||
def test_false_for_file_root(self, tmp_path: Path) -> None:
|
||
file_root = tmp_path / "file"
|
||
file_root.write_bytes(b"x")
|
||
assert LocalHotFolder(file_root).healthy() is False
|
||
|
||
def test_never_raises(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||
def boom(self: Path) -> bool:
|
||
raise PermissionError("нет доступа")
|
||
|
||
monkeypatch.setattr(Path, "is_dir", boom)
|
||
assert LocalHotFolder(tmp_path).healthy() is False
|