132 lines
5.6 KiB
Python
132 lines
5.6 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",
|
||
"SCAN_SETTLE_SECONDS": 1.5,
|
||
}
|
||
)
|
||
|
||
|
||
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",
|
||
"SCAN_SETTLE_SECONDS": 3,
|
||
"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.tmp_dir == tmp_path / "Tmp" # служебный каталог эмулятора, шлюз его не использует
|
||
assert hf.settle_seconds == 1.5
|
||
assert hf.healthy() is True
|
||
|
||
|
||
def test_cleanup_flag_from_settings(tmp_path: Path) -> None:
|
||
settings_without = _local_settings(tmp_path)
|
||
settings_with = _local_settings(tmp_path).model_copy(update={"SCAN_CLEANUP": True})
|
||
assert create_hotfolder(settings_without).cleanup_enabled is False
|
||
assert create_hotfolder(settings_with).cleanup_enabled 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"
|
||
assert not hasattr(hf, "tmp_dir") # у шлюза служебного каталога нет (change hotfolder-direct-write)
|
||
assert hf.settle_seconds == 3
|
||
|
||
|
||
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", "scandir", "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
|