a65ff8cf40
- F-D: фейк rename бросает SMBOSError(STATUS_OBJECT_NAME_COLLISION) (и обычный FileExistsError), rmdir — STATUS_DIRECTORY_NOT_EMPTY; тесты validate_name, сирот .part/.done, access-denied scandir, подкаталога в Output, .part при существующем назначении, учётных данных во всех вызовах очистки; докстринг SmbHotFolder.put больше не противоречит сам себе (T12.5) - F-E: тест ошибки очистки проверяет вызов очистки и ровно одну строку WARNING - T9.4: докстринг протокола Normalizer; T13.4: устаревшие ссылки на «Task 14»; мёртвый noqa BLE001 в worker.py Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
87 lines
3.5 KiB
Python
87 lines
3.5 KiB
Python
"""Фикстуры unit-тестов: окружение воркера шлюза `make_env` (обвязка — `tests/worker_helpers.py`)."""
|
||
|
||
import json
|
||
from collections.abc import AsyncIterator, Awaitable, Callable
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from src.core.config import get_settings
|
||
from src.core.nats import NatsGateway
|
||
from src.core.worker import Worker
|
||
from src.gateway.hotfolder import LocalHotFolder
|
||
from src.normalizer import NullNormalizer
|
||
from tests.worker_helpers import PDF_BYTES, SUCCESS_ANSWER, Env, EnvFactory, FakeNatsClient, FakeStorage, SpyHotFolder
|
||
|
||
|
||
@pytest.fixture
|
||
async def make_env(tmp_path: Path) -> AsyncIterator[EnvFactory]:
|
||
"""Фабрика окружений воркера на `tmp_path`: настройки — `get_settings()` с короткими опросом и тайм-аутом
|
||
(`**overrides` — поверх), подмены `normalizer`, `clock`, `sleep` (пауза между опросами горячей папки).
|
||
Воркер запущен (`start()`); после теста все созданные воркеры останавливаются.
|
||
"""
|
||
created: list[Env] = []
|
||
|
||
async def build(
|
||
*,
|
||
normalizer: Any = None,
|
||
root_exists: bool = True,
|
||
clock: Callable[[], datetime] | None = None,
|
||
sleep: Callable[[float], Awaitable[None]] | None = None,
|
||
**overrides: Any,
|
||
) -> Env:
|
||
root = tmp_path / f"hotfolder-{len(created)}"
|
||
if root_exists:
|
||
root.mkdir()
|
||
fixtures = tmp_path / "fixtures"
|
||
fixtures.mkdir(exist_ok=True)
|
||
(fixtures / "ok_doc.json").write_text(json.dumps(SUCCESS_ANSWER, ensure_ascii=False), encoding="utf-8")
|
||
(fixtures / "ok_doc.pdf").write_bytes(PDF_BYTES) # пара ответа по sha256 содержимого (эмулятор)
|
||
settings = get_settings().model_copy(
|
||
update={
|
||
"SCAN_HOTFOLDER_PATH": str(root),
|
||
"SCAN_POLL_SECONDS": 0.01,
|
||
"SCAN_TIMEOUT_SECONDS": 5,
|
||
"SCAN_SETTLE_SECONDS": 0, # R-23: набор ответа готов сразу, тесты не ждут стабилизации
|
||
"PUBLISH_RETRIES": 1,
|
||
**overrides,
|
||
}
|
||
)
|
||
events: list[tuple[str, str]] = []
|
||
storage = FakeStorage(events)
|
||
client = FakeNatsClient(events)
|
||
gateway = NatsGateway(settings, client_factory=lambda: client, retry_delay=0)
|
||
await gateway.connect()
|
||
local = LocalHotFolder(
|
||
root,
|
||
settings.SCAN_INPUT_DIR,
|
||
settings.SCAN_OUTPUT_DIR,
|
||
settings.SCAN_TMP_DIR,
|
||
settle_seconds=settings.SCAN_SETTLE_SECONDS,
|
||
cleanup=settings.SCAN_CLEANUP,
|
||
)
|
||
hotfolder = SpyHotFolder(local, events)
|
||
kwargs: dict[str, Any] = {}
|
||
if clock is not None:
|
||
kwargs["clock"] = clock
|
||
if sleep is not None:
|
||
kwargs["sleep"] = sleep
|
||
worker = Worker(
|
||
settings,
|
||
storage=storage,
|
||
publisher=gateway,
|
||
hotfolder=hotfolder,
|
||
normalizer=normalizer or NullNormalizer(),
|
||
**kwargs,
|
||
)
|
||
worker.start()
|
||
env = Env(settings, root, fixtures, events, storage, client, gateway, hotfolder, worker)
|
||
created.append(env)
|
||
return env
|
||
|
||
yield build
|
||
for env in created:
|
||
await env.worker.stop()
|