1695573b35
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>
115 lines
4.8 KiB
Python
115 lines
4.8 KiB
Python
"""Фабрика горячей папки `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
|