Срез 0 [4.2]: SmbHotFolder поверх smbclient (src/gateway/smb.py), фабрика create_hotfolder, unit-тесты, отметка задачи
SmbHotFolder: UNC-корень \\host\SMB_PATH (или полный UNC в SMB_PATH), put — `.<name>.part` + smbclient.replace, poll — open_file(rb), ENOENT (в т.ч. нет каталога) -> None, healthy — listdir корня без исключений. create_hotfolder(settings): local -> LocalHotFolder, smb -> SmbHotFolder с ленивым импортом модуля (R-6); отсутствие импорта src.gateway.smb в режиме local проверяется в отдельном интерпретаторе. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
ad5cb7f0f0
commit
1695573b35
@ -0,0 +1,30 @@
|
||||
"""Горячие папки ИС Сканирование: фабрика по настройкам (R-6).
|
||||
|
||||
`SCAN_HOTFOLDER_MODE=local` — `LocalHotFolder` (модуль `src.gateway.smb` не импортируется), `smb` — `SmbHotFolder`
|
||||
из `OPENBAO.SMB` (импорт модуля — только здесь, лениво).
|
||||
"""
|
||||
|
||||
from src.core.config import Settings
|
||||
from src.gateway.hotfolder import HotFolder, LocalHotFolder
|
||||
|
||||
__all__ = ["create_hotfolder"]
|
||||
|
||||
|
||||
def create_hotfolder(settings: Settings) -> HotFolder:
|
||||
"""Горячая папка для режима `SCAN_HOTFOLDER_MODE`. К файловой системе и шаре не обращается."""
|
||||
mode = settings.SCAN_HOTFOLDER_MODE
|
||||
if mode == "local":
|
||||
return LocalHotFolder(settings.SCAN_HOTFOLDER_PATH, settings.SCAN_INPUT_DIR, settings.SCAN_OUTPUT_DIR)
|
||||
if mode == "smb":
|
||||
from src.gateway.smb import SmbHotFolder # noqa: PLC0415 — ленивый импорт: в режиме local smb не нужен (R-6)
|
||||
|
||||
smb = settings.OPENBAO.SMB
|
||||
return SmbHotFolder(
|
||||
smb.SMB_HOST,
|
||||
smb.SMB_PATH,
|
||||
settings.SCAN_INPUT_DIR,
|
||||
settings.SCAN_OUTPUT_DIR,
|
||||
login=smb.SMB_LOGIN,
|
||||
password=smb.SMB_PASSWORD,
|
||||
)
|
||||
raise ValueError(f"неизвестный SCAN_HOTFOLDER_MODE: {mode!r}")
|
||||
115
api/src/gateway/smb.py
Normal file
115
api/src/gateway/smb.py
Normal file
@ -0,0 +1,115 @@
|
||||
r"""Горячая папка на SMB-шаре платформы (`SCAN_HOTFOLDER_MODE=smb`, D-1, R-5): `SmbHotFolder` поверх `smbclient`.
|
||||
|
||||
Модуль импортируется лениво — только фабрикой `create_hotfolder` в режиме `smb` (R-6). SMB-сессия регистрируется
|
||||
в lifespan приложения (`app.py`); логин/пароль, если заданы, передаются ещё и в каждый вызов `smbclient`, чтобы
|
||||
сессия поднималась заново по требованию (после сбоя регистрации при старте или сброса соединения).
|
||||
|
||||
Пути — UNC: `\\<host>\<SMB_PATH>\<input_dir>\<имя>` и `\\<host>\<SMB_PATH>\<output_dir>\<job_id>\<job_id>.json`.
|
||||
Ошибки smbprotocol — `SMBOSError` (подкласс `OSError`, но НЕ `FileNotFoundError`): «нет файла/каталога» распознаётся
|
||||
по `errno == ENOENT`.
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
import errno
|
||||
import ntpath
|
||||
from typing import Any
|
||||
|
||||
import smbclient
|
||||
|
||||
from src.gateway.hotfolder import validate_job_id, validate_name
|
||||
|
||||
|
||||
def _has_errno(err: OSError, code: int, exc_type: type[OSError]) -> bool:
|
||||
"""`SMBOSError` несёт `errno`, но не превращается в подкласс `OSError` по нему — проверяем и то, и другое."""
|
||||
return isinstance(err, exc_type) or err.errno == code
|
||||
|
||||
|
||||
def build_unc_root(host: str, path: str) -> str:
|
||||
r"""UNC-корень горячей папки.
|
||||
|
||||
`path` — путь на сервере `host` (`share/scan`, разделители `/` или `\`) или уже полный UNC-путь
|
||||
(`\\server\share\dir` или `//server/share/dir`, как `SMB_PATH` в старом коде) — тогда `host` не используется.
|
||||
Нужны хотя бы сервер и имя шары, иначе `ValueError`.
|
||||
"""
|
||||
if path.startswith(("\\\\", "//")):
|
||||
unc = ntpath.normpath(path)
|
||||
else:
|
||||
server, share_path = host.strip("\\/ "), path.strip("\\/ ")
|
||||
if not server or not share_path:
|
||||
raise ValueError(f"неполный адрес SMB: host={host!r}, path={path!r}")
|
||||
unc = ntpath.normpath(f"\\\\{server}\\{share_path}")
|
||||
if len([part for part in unc.split("\\") if part]) < 2: # noqa: PLR2004 — сервер и шара
|
||||
raise ValueError(f"в пути SMB нет сервера и шары: {path!r}")
|
||||
return unc.rstrip("\\")
|
||||
|
||||
|
||||
class SmbHotFolder:
|
||||
"""Горячая папка ИС Сканирование на SMB-шаре. Методы синхронные (блокирующий SMB), семантика как у
|
||||
`LocalHotFolder`: те же правила имён, атомарная запись `.<name>.part` + replace, `Input` создаётся без родителей,
|
||||
корень не создаётся."""
|
||||
|
||||
def __init__( # noqa: PLR0913 — адрес и учётные данные D-1 + имена каталогов, как у LocalHotFolder
|
||||
self,
|
||||
host: str,
|
||||
path: str,
|
||||
input_dir: str = "Input",
|
||||
output_dir: str = "Output",
|
||||
*,
|
||||
login: str = "",
|
||||
password: str = "",
|
||||
) -> None:
|
||||
self.root: str = build_unc_root(host, path)
|
||||
self.input_dir: str = ntpath.join(self.root, input_dir)
|
||||
self.output_dir: str = ntpath.join(self.root, output_dir)
|
||||
self._credentials: dict[str, Any] = {"username": login, "password": password} if login else {}
|
||||
|
||||
def answer_path(self, job_id: str) -> str:
|
||||
r"""UNC-путь ответа ИС: `<output_dir>\<job_id>\<job_id>.json`."""
|
||||
validate_job_id(job_id)
|
||||
return ntpath.join(self.output_dir, job_id, f"{job_id}.json")
|
||||
|
||||
def put(self, name: str, data: bytes) -> None:
|
||||
"""Записать `.<name>.part` во входящий каталог и переименовать в `name` (`smbclient.replace`: существующий
|
||||
файл заменяется). Имя проверяется до любого обращения к шаре. При сбое временный файл удаляется (ошибка
|
||||
удаления не маскирует исходную). Нет корня — `SMBOSError` (ENOENT) от `mkdir`.
|
||||
"""
|
||||
validate_name(name)
|
||||
self._ensure_input_dir()
|
||||
part = ntpath.join(self.input_dir, f".{name}.part")
|
||||
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)
|
||||
except BaseException:
|
||||
with contextlib.suppress(Exception):
|
||||
smbclient.remove(part, **self._credentials)
|
||||
raise
|
||||
|
||||
def poll(self, job_id: str) -> bytes | None:
|
||||
r"""Байты `<output>\<job_id>\<job_id>.json` или `None`, пока его нет (нет файла, каталога задания или
|
||||
`Output`). Остальные ошибки SMB поднимаются."""
|
||||
path = self.answer_path(job_id)
|
||||
try:
|
||||
with smbclient.open_file(path, mode="rb", **self._credentials) as fd:
|
||||
data: bytes = fd.read()
|
||||
except OSError as err:
|
||||
if _has_errno(err, errno.ENOENT, FileNotFoundError):
|
||||
return None
|
||||
raise
|
||||
return data
|
||||
|
||||
def healthy(self) -> bool:
|
||||
"""Корень доступен и является каталогом (`smbclient.listdir` проходит). Не бросает исключений."""
|
||||
try:
|
||||
smbclient.listdir(self.root, **self._credentials)
|
||||
except Exception:
|
||||
return False
|
||||
return True
|
||||
|
||||
def _ensure_input_dir(self) -> None:
|
||||
"""`mkdir` входящего каталога; «уже существует» (EEXIST) — не ошибка."""
|
||||
try:
|
||||
smbclient.mkdir(self.input_dir, **self._credentials)
|
||||
except OSError as err:
|
||||
if not _has_errno(err, errno.EEXIST, FileExistsError):
|
||||
raise
|
||||
114
api/tests/unit/test_hotfolder_factory.py
Normal file
114
api/tests/unit/test_hotfolder_factory.py
Normal file
@ -0,0 +1,114 @@
|
||||
"""Фабрика горячей папки `create_hotfolder(settings)` (R-6, D-1, slice-0-skeleton): режим из `SCAN_HOTFOLDER_MODE`,
|
||||
модуль `src.gateway.smb` импортируется только в режиме `smb`."""
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from src.core.config import Settings, SMBSettings, get_settings
|
||||
from src.gateway import create_hotfolder
|
||||
from src.gateway.hotfolder import LocalHotFolder
|
||||
|
||||
API_DIR = Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
def _local_settings(root: Path) -> Settings:
|
||||
return get_settings().model_copy(
|
||||
update={"SCAN_HOTFOLDER_MODE": "local", "SCAN_HOTFOLDER_PATH": str(root), "SCAN_INPUT_DIR": "In"}
|
||||
)
|
||||
|
||||
|
||||
def _smb_settings() -> Settings:
|
||||
settings = get_settings()
|
||||
smb = SMBSettings(SMB_HOST="smb.example.local", SMB_PATH="share/scan", SMB_LOGIN="user", SMB_PASSWORD="secret")
|
||||
return settings.model_copy(
|
||||
update={
|
||||
"SCAN_HOTFOLDER_MODE": "smb",
|
||||
"SCAN_OUTPUT_DIR": "Out",
|
||||
"OPENBAO": settings.OPENBAO.model_copy(update={"SMB": smb}),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_local_mode_builds_local_hotfolder(tmp_path: Path) -> None:
|
||||
hf = create_hotfolder(_local_settings(tmp_path))
|
||||
assert isinstance(hf, LocalHotFolder)
|
||||
assert hf.root == tmp_path
|
||||
assert hf.input_dir == tmp_path / "In"
|
||||
assert hf.output_dir == tmp_path / "Output"
|
||||
assert hf.healthy() is True
|
||||
|
||||
|
||||
def test_smb_mode_builds_smb_hotfolder() -> None:
|
||||
from src.gateway.smb import SmbHotFolder # noqa: PLC0415 — импорт смб-модуля только в этом тесте
|
||||
|
||||
hf = create_hotfolder(_smb_settings())
|
||||
assert isinstance(hf, SmbHotFolder)
|
||||
assert hf.root == "\\\\smb.example.local\\share\\scan"
|
||||
assert hf.input_dir == "\\\\smb.example.local\\share\\scan\\Input"
|
||||
assert hf.output_dir == "\\\\smb.example.local\\share\\scan\\Out"
|
||||
|
||||
|
||||
def test_smb_mode_does_not_touch_share(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
from src.gateway import smb # noqa: PLC0415
|
||||
|
||||
def forbidden(*args: object, **kwargs: object) -> None:
|
||||
raise AssertionError("фабрика не должна обращаться к SMB")
|
||||
|
||||
for name in ("open_file", "listdir", "mkdir", "replace", "remove", "register_session"):
|
||||
monkeypatch.setattr(smb.smbclient, name, forbidden)
|
||||
create_hotfolder(_smb_settings())
|
||||
|
||||
|
||||
def test_unknown_mode_is_rejected(tmp_path: Path) -> None:
|
||||
settings = _local_settings(tmp_path).model_copy(update={"SCAN_HOTFOLDER_MODE": "ftp"})
|
||||
with pytest.raises(ValueError, match="SCAN_HOTFOLDER_MODE"):
|
||||
create_hotfolder(settings)
|
||||
|
||||
|
||||
def test_local_mode_does_not_import_smb_module(tmp_path: Path) -> None:
|
||||
"""Отдельный интерпретатор: в процессе pytest модуль мог быть импортирован другими тестами."""
|
||||
code = "\n".join(
|
||||
[
|
||||
"import sys",
|
||||
"from src.core.config import get_settings",
|
||||
"from src.gateway import create_hotfolder",
|
||||
f"settings = get_settings().model_copy(update={{'SCAN_HOTFOLDER_PATH': {str(tmp_path)!r}}})",
|
||||
"hf = create_hotfolder(settings)",
|
||||
"assert settings.SCAN_HOTFOLDER_MODE == 'local'",
|
||||
"assert type(hf).__name__ == 'LocalHotFolder', type(hf)",
|
||||
"loaded = sorted(m for m in sys.modules if m.startswith('src.gateway'))",
|
||||
"assert 'src.gateway.smb' not in sys.modules, loaded",
|
||||
"print(loaded)",
|
||||
]
|
||||
)
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", code], cwd=API_DIR, capture_output=True, text=True, check=False, timeout=60
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert "src.gateway.hotfolder" in result.stdout
|
||||
|
||||
|
||||
def test_smb_mode_imports_smb_module_lazily() -> None:
|
||||
"""Контроль предыдущего теста: в режиме smb тот же сценарий импортирует `src.gateway.smb`."""
|
||||
code = "\n".join(
|
||||
[
|
||||
"import sys",
|
||||
"from src.core.config import SMBSettings, get_settings",
|
||||
"from src.gateway import create_hotfolder",
|
||||
"assert 'src.gateway.smb' not in sys.modules",
|
||||
"s = get_settings()",
|
||||
"smb = SMBSettings(SMB_HOST='h', SMB_PATH='share', SMB_LOGIN='u', SMB_PASSWORD='p')",
|
||||
"openbao = s.OPENBAO.model_copy(update={'SMB': smb})",
|
||||
"s = s.model_copy(update={'SCAN_HOTFOLDER_MODE': 'smb', 'OPENBAO': openbao})",
|
||||
"hf = create_hotfolder(s)",
|
||||
"assert type(hf).__name__ == 'SmbHotFolder', type(hf)",
|
||||
"assert 'src.gateway.smb' in sys.modules",
|
||||
]
|
||||
)
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", code], cwd=API_DIR, capture_output=True, text=True, check=False, timeout=60
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
331
api/tests/unit/test_smb_hotfolder.py
Normal file
331
api/tests/unit/test_smb_hotfolder.py
Normal file
@ -0,0 +1,331 @@
|
||||
"""Горячая папка на SMB-шаре (spec image-recognition-gateway, D-1, R-5, slice-0-skeleton).
|
||||
|
||||
Модуль `smbclient` подменён фейком: UNC-пути `\\\\host\\share\\...` отображаются на `tmp_path`, ошибки — настоящие
|
||||
`SMBOSError` (как у smbprotocol), все вызовы записываются. Тесты проверяют сами вызовы и UNC-пути.
|
||||
"""
|
||||
|
||||
import os
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from typing import IO, Any
|
||||
|
||||
import pytest
|
||||
from smbprotocol.exceptions import NtStatus, SMBException, SMBOSError
|
||||
|
||||
from src.gateway import smb
|
||||
from src.gateway.hotfolder import HotFolder, inbox_name
|
||||
from src.gateway.smb import SmbHotFolder
|
||||
|
||||
HOST = "smb.example.local"
|
||||
ROOT = "\\\\smb.example.local\\share\\scan"
|
||||
INPUT = ROOT + "\\Input"
|
||||
OUTPUT = ROOT + "\\Output"
|
||||
|
||||
|
||||
class FakeSmbClient:
|
||||
"""Подмена модуля `smbclient` (только используемые функции) поверх локального каталога."""
|
||||
|
||||
def __init__(self, base: Path) -> None:
|
||||
self.base = base
|
||||
self.calls: list[tuple[str, tuple[str, ...], dict[str, Any]]] = []
|
||||
self.fail: dict[str, BaseException] = {} # имя функции -> исключение, которое она бросит
|
||||
|
||||
def _record(self, func: str, args: tuple[str, ...], kwargs: dict[str, Any]) -> None:
|
||||
self.calls.append((func, args, kwargs))
|
||||
if func in self.fail:
|
||||
raise self.fail[func]
|
||||
|
||||
def _local(self, unc: str) -> Path:
|
||||
assert unc.startswith("\\\\"), f"не UNC-путь: {unc!r}"
|
||||
assert "/" not in unc, f"прямой слеш в UNC-пути: {unc!r}"
|
||||
return self.base.joinpath(*[part for part in unc.split("\\") if part])
|
||||
|
||||
def open_file(self, path: str, mode: str = "r", **kwargs: Any) -> IO[Any]:
|
||||
self._record("open_file", (path, mode), kwargs)
|
||||
local = self._local(path)
|
||||
if not local.parent.is_dir():
|
||||
raise SMBOSError(NtStatus.STATUS_OBJECT_PATH_NOT_FOUND, path)
|
||||
if "r" in mode and not local.is_file():
|
||||
raise SMBOSError(NtStatus.STATUS_OBJECT_NAME_NOT_FOUND, path)
|
||||
return open(local, mode) # noqa: SIM115 — возвращаем файловый объект, как smbclient.open_file
|
||||
|
||||
def mkdir(self, path: str, **kwargs: Any) -> None:
|
||||
self._record("mkdir", (path,), kwargs)
|
||||
local = self._local(path)
|
||||
if local.exists():
|
||||
raise SMBOSError(NtStatus.STATUS_OBJECT_NAME_COLLISION, path)
|
||||
if not local.parent.is_dir():
|
||||
raise SMBOSError(NtStatus.STATUS_OBJECT_PATH_NOT_FOUND, path)
|
||||
local.mkdir()
|
||||
|
||||
def replace(self, src: str, dst: str, **kwargs: Any) -> None:
|
||||
self._record("replace", (src, dst), kwargs)
|
||||
local_src = self._local(src)
|
||||
if not local_src.exists():
|
||||
raise SMBOSError(NtStatus.STATUS_OBJECT_NAME_NOT_FOUND, src)
|
||||
os.replace(local_src, self._local(dst))
|
||||
|
||||
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()
|
||||
|
||||
def listdir(self, path: str, **kwargs: Any) -> list[str]:
|
||||
self._record("listdir", (path,), kwargs)
|
||||
local = self._local(path)
|
||||
if not local.exists():
|
||||
raise SMBOSError(NtStatus.STATUS_OBJECT_NAME_NOT_FOUND, path)
|
||||
if not local.is_dir():
|
||||
raise SMBOSError(NtStatus.STATUS_NOT_A_DIRECTORY, path)
|
||||
return [p.name for p in local.iterdir()]
|
||||
|
||||
def funcs(self) -> list[str]:
|
||||
return [func for func, _, _ in self.calls]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> FakeSmbClient:
|
||||
fake = FakeSmbClient(tmp_path / "smb")
|
||||
monkeypatch.setattr(smb, "smbclient", fake)
|
||||
return fake
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def share_root(fake: FakeSmbClient) -> Path:
|
||||
"""Существующий корень горячей папки `\\\\smb.example.local\\share\\scan`."""
|
||||
root = fake.base / HOST / "share" / "scan"
|
||||
root.mkdir(parents=True)
|
||||
return root
|
||||
|
||||
|
||||
def _job_id() -> str:
|
||||
return str(uuid.uuid4())
|
||||
|
||||
|
||||
def _hotfolder(**kwargs: Any) -> SmbHotFolder:
|
||||
return SmbHotFolder(HOST, "share/scan", **kwargs)
|
||||
|
||||
|
||||
def _write_answer(share_root: Path, job_id: str, data: bytes) -> None:
|
||||
folder = share_root / "Output" / job_id
|
||||
folder.mkdir(parents=True)
|
||||
(folder / f"{job_id}.json").write_bytes(data)
|
||||
|
||||
|
||||
class TestPaths:
|
||||
def test_default_layout(self) -> None:
|
||||
hf = _hotfolder()
|
||||
job_id = _job_id()
|
||||
assert hf.root == ROOT
|
||||
assert hf.input_dir == INPUT
|
||||
assert hf.output_dir == OUTPUT
|
||||
assert hf.answer_path(job_id) == f"{OUTPUT}\\{job_id}\\{job_id}.json"
|
||||
|
||||
@pytest.mark.parametrize("path", ["share/scan", "share\\scan", "/share/scan/", "\\share\\scan\\"])
|
||||
def test_relative_share_path_is_joined_with_host(self, path: str) -> None:
|
||||
assert SmbHotFolder(HOST, path).root == ROOT
|
||||
|
||||
@pytest.mark.parametrize("path", ["\\\\srv\\share\\scan", "//srv/share/scan", "\\\\srv\\share\\scan\\"])
|
||||
def test_full_unc_path_is_used_as_is(self, path: str) -> None:
|
||||
assert SmbHotFolder(HOST, path).root == "\\\\srv\\share\\scan"
|
||||
|
||||
def test_share_root_only(self) -> None:
|
||||
assert SmbHotFolder(HOST, "share").root == "\\\\smb.example.local\\share"
|
||||
|
||||
def test_custom_dir_names(self) -> None:
|
||||
hf = SmbHotFolder(HOST, "share/scan", input_dir="Входящие", output_dir="Исходящие")
|
||||
assert hf.input_dir == ROOT + "\\Входящие"
|
||||
assert hf.output_dir == ROOT + "\\Исходящие"
|
||||
assert hf.answer_path("abc") == ROOT + "\\Исходящие\\abc\\abc.json"
|
||||
|
||||
@pytest.mark.parametrize(("host", "path"), [(HOST, ""), (HOST, "/"), ("", "share/scan"), ("", "\\\\srv")])
|
||||
def test_rejects_incomplete_location(self, host: str, path: str) -> None:
|
||||
with pytest.raises(ValueError, match="SMB"):
|
||||
SmbHotFolder(host, path)
|
||||
|
||||
@pytest.mark.parametrize("job_id", ["", ".", "..", "../x", "a/b", "a\\b", "a:b", "a\x00b"])
|
||||
def test_answer_path_rejects_unsafe_job_id(self, job_id: str) -> None:
|
||||
with pytest.raises(ValueError, match="job_id"):
|
||||
_hotfolder().answer_path(job_id)
|
||||
|
||||
def test_constructor_does_not_touch_share(self, fake: FakeSmbClient) -> None:
|
||||
_hotfolder(login="user", password="secret")
|
||||
assert fake.calls == []
|
||||
|
||||
def test_implements_protocol(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
hf: HotFolder = _hotfolder()
|
||||
assert hf.healthy() is True
|
||||
|
||||
|
||||
class TestPut:
|
||||
def test_writes_part_then_replaces(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
name = inbox_name(_job_id(), "aep_5.PDF")
|
||||
data = bytes(range(256)) * 10
|
||||
_hotfolder().put(name, data)
|
||||
part = f"{INPUT}\\.{name}.part"
|
||||
assert fake.calls == [
|
||||
("mkdir", (INPUT,), {}),
|
||||
("open_file", (part, "wb"), {}),
|
||||
("replace", (part, f"{INPUT}\\{name}"), {}),
|
||||
]
|
||||
assert (share_root / "Input" / name).read_bytes() == data
|
||||
assert [p.name for p in (share_root / "Input").iterdir()] == [name]
|
||||
|
||||
def test_existing_input_dir_is_fine(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
(share_root / "Input").mkdir()
|
||||
_hotfolder().put("a.pdf", b"1")
|
||||
assert (share_root / "Input" / "a.pdf").read_bytes() == b"1"
|
||||
|
||||
def test_missing_root_raises_and_writes_nothing(self, fake: FakeSmbClient) -> None:
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
_hotfolder().put("a.pdf", b"1")
|
||||
assert isinstance(excinfo.value, SMBOSError)
|
||||
assert fake.funcs() == ["mkdir"]
|
||||
assert not fake.base.exists()
|
||||
|
||||
def test_mkdir_other_error_propagates(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
fake.fail["mkdir"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
|
||||
with pytest.raises(SMBOSError):
|
||||
_hotfolder().put("a.pdf", b"1")
|
||||
assert fake.funcs() == ["mkdir"]
|
||||
|
||||
def test_overwrites_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"]
|
||||
|
||||
def test_empty_data_is_written(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
_hotfolder().put("empty.pdf", b"")
|
||||
assert (share_root / "Input" / "empty.pdf").read_bytes() == b""
|
||||
|
||||
def test_spaces_and_cyrillic_are_accepted(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
name = inbox_name(_job_id(), "Отчёт за 2024 год (копия).pdf")
|
||||
_hotfolder().put(name, b"x")
|
||||
assert (share_root / "Input" / name).read_bytes() == b"x"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"name",
|
||||
[
|
||||
"",
|
||||
"../escape.pdf",
|
||||
"sub/a.pdf",
|
||||
"sub\\a.pdf",
|
||||
"/abs.pdf",
|
||||
".hidden.pdf",
|
||||
".a.pdf.part",
|
||||
"..",
|
||||
".",
|
||||
"a\x00b",
|
||||
"a:b.pdf",
|
||||
"a<b.pdf",
|
||||
"a>b.pdf",
|
||||
'a"b.pdf',
|
||||
"a|b.pdf",
|
||||
"a?b.pdf",
|
||||
"a*b.pdf",
|
||||
"a\x01b.pdf",
|
||||
"a\nb.pdf",
|
||||
],
|
||||
)
|
||||
def test_rejects_invalid_names_before_any_call(self, fake: FakeSmbClient, share_root: Path, name: str) -> None:
|
||||
with pytest.raises(ValueError, match="имя файла"):
|
||||
_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)
|
||||
with pytest.raises(OSError):
|
||||
_hotfolder().put("a.pdf", b"data")
|
||||
part = f"{INPUT}\\.a.pdf.part"
|
||||
assert fake.calls[-1] == ("remove", (part,), {})
|
||||
assert list((share_root / "Input").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["remove"] = SMBException("соединение потеряно")
|
||||
with pytest.raises(SMBOSError):
|
||||
_hotfolder().put("a.pdf", b"data")
|
||||
assert fake.funcs() == ["mkdir", "open_file", "replace", "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"}] * 3
|
||||
|
||||
|
||||
class TestPoll:
|
||||
def test_none_until_answer_exists(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
hf = _hotfolder()
|
||||
job_id = _job_id()
|
||||
assert hf.poll(job_id) is None # нет Output: STATUS_OBJECT_PATH_NOT_FOUND
|
||||
(share_root / "Output").mkdir()
|
||||
assert hf.poll(job_id) is None # нет Output/<job_id>
|
||||
(share_root / "Output" / job_id).mkdir()
|
||||
assert hf.poll(job_id) is None # нет <job_id>.json: STATUS_OBJECT_NAME_NOT_FOUND
|
||||
|
||||
def test_returns_bytes_when_answer_appears(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
hf = _hotfolder()
|
||||
job_id = _job_id()
|
||||
payload = b'{"file_uid": "x", "data_ocr": []}'
|
||||
_write_answer(share_root, job_id, payload)
|
||||
assert hf.poll(job_id) == payload
|
||||
assert fake.calls == [("open_file", (f"{OUTPUT}\\{job_id}\\{job_id}.json", "rb"), {})]
|
||||
|
||||
def test_reads_only_own_answer(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
job_id, other = _job_id(), _job_id()
|
||||
_write_answer(share_root, other, b"{}")
|
||||
(share_root / "Output" / other / "extra.json").write_bytes(b"[]")
|
||||
hf = _hotfolder()
|
||||
assert hf.poll(job_id) is None
|
||||
assert hf.poll(other) == b"{}"
|
||||
assert fake.funcs() == ["open_file", "open_file"]
|
||||
|
||||
def test_missing_root_is_none(self, fake: FakeSmbClient) -> None:
|
||||
assert _hotfolder().poll(_job_id()) is None
|
||||
|
||||
def test_plain_file_not_found_is_none(self, fake: FakeSmbClient) -> None:
|
||||
fake.fail["open_file"] = FileNotFoundError("нет файла")
|
||||
assert _hotfolder().poll(_job_id()) is None
|
||||
|
||||
def test_other_errors_propagate(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
fake.fail["open_file"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, OUTPUT)
|
||||
with pytest.raises(SMBOSError):
|
||||
_hotfolder().poll(_job_id())
|
||||
|
||||
def test_invalid_job_id_rejected_before_any_call(self, fake: FakeSmbClient) -> None:
|
||||
with pytest.raises(ValueError, match="job_id"):
|
||||
_hotfolder().poll("../x")
|
||||
assert fake.calls == []
|
||||
|
||||
def test_passes_credentials(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
_hotfolder(login="user", password="secret").poll(_job_id())
|
||||
assert fake.calls[0][2] == {"username": "user", "password": "secret"}
|
||||
|
||||
|
||||
class TestHealthy:
|
||||
def test_true_for_existing_root(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
assert _hotfolder().healthy() is True
|
||||
assert fake.calls == [("listdir", (ROOT,), {})]
|
||||
|
||||
def test_false_for_missing_root(self, fake: FakeSmbClient) -> None:
|
||||
assert _hotfolder().healthy() is False
|
||||
|
||||
def test_false_for_file_root(self, fake: FakeSmbClient) -> None:
|
||||
file_root = fake.base / HOST / "share" / "scan"
|
||||
file_root.parent.mkdir(parents=True)
|
||||
file_root.write_bytes(b"x")
|
||||
assert _hotfolder().healthy() is False
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"error", [SMBException("сервер недоступен"), ValueError("плохой путь"), ConnectionError("сброс"), OSError()]
|
||||
)
|
||||
def test_never_raises(self, fake: FakeSmbClient, share_root: Path, error: Exception) -> None:
|
||||
fake.fail["listdir"] = error
|
||||
assert _hotfolder().healthy() is False
|
||||
|
||||
def test_passes_credentials(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
_hotfolder(login="user", password="secret").healthy()
|
||||
assert fake.calls == [("listdir", (ROOT,), {"username": "user", "password": "secret"})]
|
||||
@ -48,7 +48,7 @@
|
||||
|
||||
### Task 8: [4.2] `src/gateway/smb.py`
|
||||
|
||||
- [ ] `src/gateway/smb.py`: `SmbHotFolder` поверх `smbclient` (UNC `\\host\path\Input`), те же методы; unit-тест с подменённым `smbclient` (вызовы `open_file`, `listdir`); проверка: тест зелёный, в режиме `local` модуль не импортируется
|
||||
- [x] `src/gateway/smb.py`: `SmbHotFolder` поверх `smbclient` (UNC `\\host\path\Input`), те же методы; unit-тест с подменённым `smbclient` (вызовы `open_file`, `listdir`); проверка: тест зелёный, в режиме `local` модуль не импортируется
|
||||
|
||||
### Task 9: [4.3] `src/scan_emulator/__init__.py` + `__main__.py`
|
||||
|
||||
|
||||
@ -35,7 +35,7 @@
|
||||
|
||||
- [x] 4.1 `src/gateway/hotfolder.py`: протокол `HotFolder`, `LocalHotFolder` (put/poll/healthy); `tests/unit/test_local_hotfolder.py`
|
||||
на `tmp_path`; проверка: put пишет `<uuid>_<name>`, poll возвращает `None` до появления ответа и байты после
|
||||
- [ ] 4.2 `src/gateway/smb.py`: `SmbHotFolder` поверх `smbclient` (UNC `\\host\path\Input`), те же методы; unit-тест с
|
||||
- [x] 4.2 `src/gateway/smb.py`: `SmbHotFolder` поверх `smbclient` (UNC `\\host\path\Input`), те же методы; unit-тест с
|
||||
подменённым `smbclient` (вызовы `open_file`, `listdir`); проверка: тест зелёный, в режиме `local` модуль не импортируется
|
||||
- [ ] 4.3 `src/scan_emulator/__init__.py` + `__main__.py`: функция `process_once(root, fixtures, delay)` и цикл; поиск
|
||||
фикстуры по имени/префиксу, подстановка `file_uid`, ошибка `INVALID_DOCUMENT`, переименование в `.done`;
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user