feat(hotfolder): SMB — rename без замены и cleanup под флагом (D-1, D-2)
This commit is contained in:
Andrey Epifancev
parent
5c2d89bef3
commit
c00823c673
@ -4,9 +4,10 @@ r"""Горячая папка на SMB-шаре платформы (`SCAN_HOTFOL
|
||||
в lifespan приложения (`app.py`); логин/пароль, если заданы, передаются ещё и в каждый вызов `smbclient`, чтобы
|
||||
сессия поднималась заново по требованию (после сбоя регистрации при старте или сброса соединения).
|
||||
|
||||
Пути — UNC: временный файл `\\<host>\<SMB_PATH>\<tmp_dir>\.<имя>.part` переименовывается во входящий
|
||||
`\\<host>\<SMB_PATH>\<input_dir>\<имя>` (одна шара — rename атомарен, OQ-79); ответ — все `*.json` каталога
|
||||
`\\<host>\<SMB_PATH>\<output_dir>\<job_id>\` (OQ-78), готовность — по стабилизации набора (`SettleTracker`).
|
||||
Пути — UNC: временный файл `\\<host>\<SMB_PATH>\<tmp_dir>\.<имя>.part` переименовывается (rename без замены)
|
||||
во входящий `\\<host>\<SMB_PATH>\<input_dir>\<имя>` (одна шара — rename атомарен, OQ-79); ответ — все `*.json`
|
||||
каталога `\\<host>\<SMB_PATH>\<output_dir>\<job_id>\` (OQ-78), готовность — по стабилизации набора
|
||||
(`SettleTracker`). Очистка (D-2) удаляет ответы, входящий файл и временные файлы при `SCAN_CLEANUP=true`.
|
||||
Ошибки smbprotocol — `SMBOSError` (подкласс `OSError`, но НЕ `FileNotFoundError`): «нет файла/каталога» распознаётся
|
||||
по `errno == ENOENT`.
|
||||
"""
|
||||
@ -65,8 +66,9 @@ def build_unc_root(host: str, path: str) -> str:
|
||||
|
||||
class SmbHotFolder:
|
||||
"""Горячая папка ИС Сканирование на SMB-шаре. Методы синхронные (блокирующий SMB), семантика как у
|
||||
`LocalHotFolder`: те же правила имён, атомарная запись `Tmp\\.<name>.part` + replace в `Input`, `Tmp` и `Input`
|
||||
создаются без родителей, корень не создаётся, готовность набора ответа — по стабилизации (`SettleTracker`)."""
|
||||
`LocalHotFolder`: те же правила имён, атомарная запись `Tmp\\.<name>.part` + rename без замены в `Input`,
|
||||
`Tmp` и `Input` создаются без родителей, корень не создаётся, готовность набора ответа — по стабилизации
|
||||
(`SettleTracker`). Очистка (D-2) под флагом `SCAN_CLEANUP`."""
|
||||
|
||||
def __init__( # noqa: PLR0913 — адрес и учётные данные D-1 + имена каталогов и готовность, как у LocalHotFolder
|
||||
self,
|
||||
@ -100,9 +102,10 @@ class SmbHotFolder:
|
||||
return ntpath.join(self.answer_dir(job_id), f"{job_id}.json")
|
||||
|
||||
def put(self, name: str, data: bytes) -> None:
|
||||
"""Записать `.<name>.part` в служебный каталог и переименовать в `<input>\\<name>` (`smbclient.replace`:
|
||||
существующий файл заменяется). Имя проверяется до любого обращения к шаре. При сбое временный файл удаляется
|
||||
(ошибка удаления не маскирует исходную). Нет корня — `SMBOSError` (ENOENT) от первого `mkdir`.
|
||||
"""Записать `.<name>.part` в служебный каталог, затем `smbclient.rename` в `<input>\\<name>` без замены:
|
||||
назначение есть — файл считается доставленным, `.part` остаётся (удаляется только при включённой очистке).
|
||||
Имя проверяется до любого обращения к шаре. При сбое временный файл удаляется (ошибка удаления не маскирует
|
||||
исходную). Нет корня — `SMBOSError` (ENOENT) от первого `mkdir`.
|
||||
"""
|
||||
validate_name(name)
|
||||
self._ensure_dir(self.tmp_dir)
|
||||
@ -111,10 +114,15 @@ class SmbHotFolder:
|
||||
try:
|
||||
with smbclient.open_file(part, mode="wb", **self._credentials) as fd:
|
||||
fd.write(data)
|
||||
smbclient.replace(part, ntpath.join(self.input_dir, name), **self._credentials)
|
||||
smbclient.rename(part, ntpath.join(self.input_dir, name), **self._credentials)
|
||||
except OSError as err:
|
||||
self._drop_part(part)
|
||||
ntstatus = getattr(err, "ntstatus", None)
|
||||
if _has_errno(err, errno.EEXIST, FileExistsError) or ntstatus == NtStatus.STATUS_OBJECT_NAME_COLLISION:
|
||||
return
|
||||
raise
|
||||
except BaseException:
|
||||
with contextlib.suppress(Exception):
|
||||
smbclient.remove(part, **self._credentials)
|
||||
self._drop_part(part)
|
||||
raise
|
||||
|
||||
def poll(self, job_id: str) -> list[AnswerFile] | None:
|
||||
@ -156,9 +164,42 @@ class SmbHotFolder:
|
||||
"""Число заданий с ненастоявшимся набором (для тестов и диагностики)."""
|
||||
return len(self._settle)
|
||||
|
||||
def _drop_part(self, path: str) -> None:
|
||||
"""Удалить временный файл при ошибке, только если очистка включена (D-2)."""
|
||||
if self.cleanup_enabled:
|
||||
with contextlib.suppress(Exception):
|
||||
smbclient.remove(path, **self._credentials)
|
||||
|
||||
def cleanup(self, job_id: str, name: str) -> bool:
|
||||
"""Очистка SMB — заглушка на Task 11 (логика в Task 12)."""
|
||||
return False
|
||||
"""Удалить следы задания: ответ из `output`, входящий файл, временные файлы. Возвращает `False`, если
|
||||
очистка выключена; `True` при успешной очистке. Ошибки удаления (кроме ENOENT) поднимаются."""
|
||||
if not self.cleanup_enabled:
|
||||
return False
|
||||
validate_name(name)
|
||||
folder = self.answer_dir(job_id)
|
||||
try:
|
||||
entries = list(smbclient.scandir(folder, **self._credentials))
|
||||
except OSError as err:
|
||||
is_enoent = _has_errno(err, errno.ENOENT, FileNotFoundError)
|
||||
is_enotdir = _has_errno(err, errno.ENOTDIR, NotADirectoryError)
|
||||
if not is_enoent and not is_enotdir:
|
||||
raise
|
||||
else:
|
||||
for entry in entries:
|
||||
if entry.is_file():
|
||||
smbclient.remove(ntpath.join(folder, entry.name), **self._credentials)
|
||||
smbclient.rmdir(folder, **self._credentials)
|
||||
for path in (
|
||||
ntpath.join(self.input_dir, name),
|
||||
ntpath.join(self.input_dir, f"{name}.done"),
|
||||
ntpath.join(self.tmp_dir, f".{name}.part"),
|
||||
):
|
||||
try:
|
||||
smbclient.remove(path, **self._credentials)
|
||||
except OSError as err:
|
||||
if not _has_errno(err, errno.ENOENT, FileNotFoundError):
|
||||
raise
|
||||
return True
|
||||
|
||||
def healthy(self) -> bool:
|
||||
"""Корень доступен и является каталогом (`smbclient.listdir` проходит). Не бросает исключений."""
|
||||
|
||||
@ -80,12 +80,23 @@ class FakeSmbClient:
|
||||
raise SMBOSError(NtStatus.STATUS_OBJECT_NAME_NOT_FOUND, src)
|
||||
os.replace(local_src, self._local(dst))
|
||||
|
||||
def rename(self, src: str, dst: str, **kwargs: Any) -> None:
|
||||
self._record("rename", (src, dst), kwargs)
|
||||
target = self._local(dst)
|
||||
if target.exists():
|
||||
raise FileExistsError(dst)
|
||||
os.rename(self._local(src), target)
|
||||
|
||||
def remove(self, path: str, **kwargs: Any) -> None:
|
||||
self._record("remove", (path,), kwargs)
|
||||
local = self._local(path)
|
||||
if not local.is_file():
|
||||
raise SMBOSError(NtStatus.STATUS_OBJECT_NAME_NOT_FOUND, path)
|
||||
local.unlink()
|
||||
try:
|
||||
os.unlink(self._local(path))
|
||||
except FileNotFoundError:
|
||||
raise SMBOSError(NtStatus.STATUS_OBJECT_NAME_NOT_FOUND, path) from None
|
||||
|
||||
def rmdir(self, path: str, **kwargs: Any) -> None:
|
||||
self._record("rmdir", (path,), kwargs)
|
||||
os.rmdir(self._local(path))
|
||||
|
||||
def scandir(self, path: str, **kwargs: Any) -> list[SimpleNamespace]:
|
||||
"""Как `smbclient.scandir`: элементы с `name`, `is_file()` и `smb_info.end_of_file` (размер из листинга)."""
|
||||
@ -206,7 +217,7 @@ class TestPaths:
|
||||
|
||||
|
||||
class TestPut:
|
||||
def test_writes_part_in_tmp_then_replaces_into_input(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
def test_writes_part_in_tmp_then_renames_into_input(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
name = inbox_name(_job_id(), "aep_5.PDF")
|
||||
data = bytes(range(256)) * 10
|
||||
_hotfolder().put(name, data)
|
||||
@ -215,7 +226,7 @@ class TestPut:
|
||||
("mkdir", (TMP,), {}),
|
||||
("mkdir", (INPUT,), {}),
|
||||
("open_file", (part, "wb"), {}),
|
||||
("replace", (part, f"{INPUT}\\{name}"), {}),
|
||||
("rename", (part, f"{INPUT}\\{name}"), {}),
|
||||
]
|
||||
assert (share_root / "Input" / name).read_bytes() == data
|
||||
assert [p.name for p in (share_root / "Input").iterdir()] == [name]
|
||||
@ -225,14 +236,14 @@ class TestPut:
|
||||
"""OQ-79: пока пишется временный файл, во входящем каталоге его нет — он в `Tmp`."""
|
||||
name = "a.pdf"
|
||||
seen: dict[str, list[str]] = {}
|
||||
real_replace = fake.replace
|
||||
real_rename = fake.rename
|
||||
|
||||
def spy(src: str, dst: str, **kwargs: Any) -> None:
|
||||
seen["input"] = sorted(p.name for p in (share_root / "Input").iterdir())
|
||||
seen["tmp"] = sorted(p.name for p in (share_root / "Tmp").iterdir())
|
||||
real_replace(src, dst, **kwargs)
|
||||
real_rename(src, dst, **kwargs)
|
||||
|
||||
fake.replace = spy # type: ignore[method-assign]
|
||||
fake.rename = spy # type: ignore[method-assign]
|
||||
_hotfolder().put(name, b"data")
|
||||
assert seen == {"input": [], "tmp": [".a.pdf.part"]}
|
||||
assert [p.name for p in (share_root / "Input").iterdir()] == [name]
|
||||
@ -283,7 +294,7 @@ class TestPut:
|
||||
fake.fail["mkdir"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
|
||||
_hotfolder().put("a.pdf", b"1")
|
||||
assert (share_root / "Input" / "a.pdf").read_bytes() == b"1"
|
||||
assert fake.funcs() == ["mkdir", "isdir", "mkdir", "isdir", "open_file", "replace"]
|
||||
assert fake.funcs() == ["mkdir", "isdir", "mkdir", "isdir", "open_file", "rename"]
|
||||
assert fake.calls[3] == ("isdir", (INPUT,), {})
|
||||
|
||||
def test_mkdir_denied_and_no_input_raises_original(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
@ -304,12 +315,12 @@ class TestPut:
|
||||
assert excinfo.value is denied
|
||||
assert fake.funcs() == ["mkdir", "isdir"]
|
||||
|
||||
def test_overwrites_existing_file(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
def test_no_replace_on_existing_file(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
hf = _hotfolder()
|
||||
hf.put("a.pdf", b"old")
|
||||
hf.put("a.pdf", b"new")
|
||||
assert (share_root / "Input" / "a.pdf").read_bytes() == b"new"
|
||||
assert [p.name for p in (share_root / "Input").iterdir()] == ["a.pdf"]
|
||||
assert (share_root / "Input" / "a.pdf").read_bytes() == b"old"
|
||||
assert (share_root / "Tmp" / ".a.pdf.part").read_bytes() == b"new"
|
||||
|
||||
def test_empty_data_is_written(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
_hotfolder().put("empty.pdf", b"")
|
||||
@ -349,27 +360,85 @@ class TestPut:
|
||||
_hotfolder().put(name, b"x")
|
||||
assert fake.calls == []
|
||||
|
||||
def test_cleans_part_file_on_replace_failure(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
fake.fail["replace"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
|
||||
def test_cleans_part_file_on_rename_failure(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
fake.fail["rename"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
|
||||
with pytest.raises(OSError):
|
||||
_hotfolder().put("a.pdf", b"data")
|
||||
_hotfolder(cleanup=True).put("a.pdf", b"data")
|
||||
part = f"{TMP}\\.a.pdf.part"
|
||||
assert fake.calls[-1] == ("remove", (part,), {})
|
||||
assert list((share_root / "Input").iterdir()) == []
|
||||
assert list((share_root / "Tmp").iterdir()) == []
|
||||
|
||||
def test_cleanup_error_does_not_mask_original(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
fake.fail["replace"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
|
||||
fake.fail["rename"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
|
||||
fake.fail["remove"] = SMBException("соединение потеряно")
|
||||
with pytest.raises(SMBOSError):
|
||||
_hotfolder().put("a.pdf", b"data")
|
||||
assert fake.funcs() == ["mkdir", "mkdir", "open_file", "replace", "remove"]
|
||||
_hotfolder(cleanup=True).put("a.pdf", b"data")
|
||||
assert fake.funcs() == ["mkdir", "mkdir", "open_file", "rename", "remove"]
|
||||
|
||||
def test_passes_credentials_to_every_call(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
_hotfolder(login="user", password="secret").put("a.pdf", b"1")
|
||||
assert [kwargs for _, _, kwargs in fake.calls] == [{"username": "user", "password": "secret"}] * 4
|
||||
|
||||
|
||||
class TestPutWithoutReplace:
|
||||
def test_uses_rename_not_replace(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
hf = _hotfolder()
|
||||
hf.put("job.pdf", b"x")
|
||||
names = [call[0] for call in fake.calls]
|
||||
assert "rename" in names and "replace" not in names
|
||||
assert fake._local(INPUT + "\\job.pdf").read_bytes() == b"x"
|
||||
assert not fake._local(TMP + "\\.job.pdf.part").exists()
|
||||
|
||||
def test_existing_destination_kept_and_part_left(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
hf = _hotfolder()
|
||||
hf.put("job.pdf", b"first")
|
||||
hf.put("job.pdf", b"second")
|
||||
assert fake._local(INPUT + "\\job.pdf").read_bytes() == b"first"
|
||||
assert fake._local(TMP + "\\.job.pdf.part").read_bytes() == b"second"
|
||||
|
||||
|
||||
class TestCleanup:
|
||||
def test_disabled_is_noop(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
hf = _hotfolder()
|
||||
hf.put("job.pdf", b"x")
|
||||
before = len(fake.calls)
|
||||
assert hf.cleanup("job", "job.pdf") is False
|
||||
assert len(fake.calls) == before
|
||||
|
||||
def test_enabled_removes_everything(self, fake: FakeSmbClient) -> None:
|
||||
share_root = fake.base / HOST / "share" / "scan"
|
||||
share_root.mkdir(parents=True)
|
||||
clock = FakeClock()
|
||||
hf = SmbHotFolder(HOST, "share\\scan", settle_seconds=0, clock=clock, cleanup=True)
|
||||
job_id = str(uuid.uuid4())
|
||||
hf.put(f"{job_id}.pdf", b"x")
|
||||
answer = fake._local(OUTPUT + f"\\{job_id}")
|
||||
answer.mkdir(parents=True)
|
||||
(answer / f"{job_id}.json").write_bytes(b"{}")
|
||||
(answer / "readme.txt").write_bytes(b"keep")
|
||||
assert hf.cleanup(job_id, f"{job_id}.pdf") is True
|
||||
assert not answer.exists() and not fake._local(INPUT + f"\\{job_id}.pdf").exists()
|
||||
assert [c[0] for c in fake.calls].count("rmdir") == 1
|
||||
|
||||
def test_missing_is_fine(self, fake: FakeSmbClient) -> None:
|
||||
share_root = fake.base / HOST / "share" / "scan"
|
||||
share_root.mkdir(parents=True)
|
||||
clock = FakeClock()
|
||||
hf = SmbHotFolder(HOST, "share\\scan", settle_seconds=0, clock=clock, cleanup=True)
|
||||
assert hf.cleanup(str(uuid.uuid4()), "nothing.pdf") is True
|
||||
|
||||
def test_access_denied_propagates(self, fake: FakeSmbClient) -> None:
|
||||
share_root = fake.base / HOST / "share" / "scan"
|
||||
share_root.mkdir(parents=True)
|
||||
clock = FakeClock()
|
||||
hf = SmbHotFolder(HOST, "share\\scan", settle_seconds=0, clock=clock, cleanup=True)
|
||||
hf.put("job.pdf", b"x")
|
||||
fake.fail["remove"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT + "\\job.pdf")
|
||||
with pytest.raises(SMBOSError):
|
||||
hf.cleanup("job", "job.pdf")
|
||||
|
||||
|
||||
class TestPoll:
|
||||
def test_none_until_answer_dir_has_json(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
hf = _hotfolder()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user