Срез 0 [5.4]: обвязка тестов воркера — в tests/worker_helpers.py и tests/unit/conftest.py

Фейки S3/NATS, шпион горячей папки, Env, эмулятор ИС и eventually вынесены из test_worker.py
без изменения логики, чтобы ими пользовался отдельный тест параллелизма. make_env получил
подмену sleep (пауза между опросами горячей папки).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

api/tests/unit/conftest.py

@@ -0,0 +1,76 @@
+"""Фикстуры 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 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")
+        settings = get_settings().model_copy(
+            update={
+                "SCAN_HOTFOLDER_PATH": str(root),
+                "SCAN_POLL_SECONDS": 0.01,
+                "SCAN_TIMEOUT_SECONDS": 5,
+                "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()
+        hotfolder = SpyHotFolder(LocalHotFolder(root, settings.SCAN_INPUT_DIR, settings.SCAN_OUTPUT_DIR), 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()

api/tests/unit/test_worker.py

@@ -1,181 +1,54 @@
 """Воркер шлюза (`src.core.worker`, задача 5.3): сценарии spec image-recognition-gateway, D-2, D-3, D-7, R-14..R-20.
 
-Без Docker и сети: S3 — словарь в памяти (ошибки — классы `src.core.clients`), NATS — настоящий `NatsGateway` поверх
+Обвязка (фейки S3/NATS, шпион горячей папки, эмулятор ИС, журнал событий) — `tests/worker_helpers.py`, фикстура
-фейкового клиента nats-py (тело, `Nats-Msg-Id`, повторы и `PublishFailed` — как в проде), горячая папка — настоящий
+`make_env` — `tests/unit/conftest.py`. Параллелизм и остановка при очереди на слот — `test_worker_concurrency.py`.
-`LocalHotFolder` на `tmp_path` под шпионом, ИС Сканирование — `process_once` эмулятора (задержка 0) в фоновой задаче.
-Общий журнал событий (`events`) упорядочивает публикации, чтения S3 и запись во входящий каталог.
-Каждый опубликованный конверт проверяется JSON Schema `recognition-result-message.schema.json`.
 """
 
 import asyncio
 import json
 import shutil
 import uuid
-from collections.abc import AsyncIterator, Awaitable, Callable
+from collections.abc import Callable
-from contextlib import asynccontextmanager, suppress
-from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any
 
 import pytest
 from jsonschema import Draft202012Validator, FormatChecker
-from nats import errors as nats_errors
 
 from src.contract import (
-    NATS_MSG_ID_HEADER,
-    RESULT_SUBJECT,
     ErrorCode,
     MessageError,
     ResultStatus,
     StatementDocument,
 )
-from src.core.clients import JSON_CONTENT_TYPE, OCTET_STREAM, S3ObjectNotFoundError, S3WriteError
+from src.core.clients import JSON_CONTENT_TYPE, OCTET_STREAM
-from src.core.config import Settings, get_settings
+from src.core.worker import MAX_ERROR_MESSAGE_CHARS
-from src.core.nats import NatsGateway
+from src.normalizer import NOT_IMPLEMENTED_MESSAGE, NormalizeOutcome
-from src.core.worker import MAX_ERROR_MESSAGE_CHARS, Worker
+from tests.worker_helpers import (
-from src.gateway.hotfolder import LocalHotFolder
+    API_DIR,
-from src.normalizer import NOT_IMPLEMENTED_MESSAGE, NormalizeOutcome, NullNormalizer
+    BUCKET,
-from src.scan_emulator import process_once
+    CONTRACTS_DIR,
+    PDF_BYTES,
+    SUCCESS_ANSWER,
+    EnvFactory,
+    emulator,
+    eventually,
+)
 
-API_DIR = Path(__file__).resolve().parents[2]
-CONTRACTS_DIR = API_DIR / "contracts"
 # `<корень scan-gateway>/../fixtures/ocr` (workspace и worktree `.wt/<срез>/`), как в тестах эмулятора
 REAL_FIXTURES = API_DIR.parent.parent / "fixtures" / "ocr"
-BUCKET = "arm-ka"
-PDF_BYTES = b"%PDF-1.4\n% scan-gateway test\n\x00\x01\x02binary"
-
-MESSAGE_VALIDATOR = Draft202012Validator(
-    json.loads((CONTRACTS_DIR / "recognition-result-message.schema.json").read_text(encoding="utf-8")),
-    format_checker=FormatChecker(),
-)
 RESULT_VALIDATOR = Draft202012Validator(
     json.loads((CONTRACTS_DIR / "recognition-result.schema.json").read_text(encoding="utf-8")),
     format_checker=FormatChecker(),
 )
 STATEMENT_EXAMPLE = json.loads((CONTRACTS_DIR / "examples" / "result-statement-minimal.json").read_text("utf-8"))
 
-# Синтетический успешный ответ ИС (R-10): форма `fixtures/ocr/aep_5.json`, `error = null`, непустой `data_ocr`
-SUCCESS_ANSWER: dict[str, Any] = {
-    "file_uid": "ok_doc",
-    "error": None,
-    "data_ocr": {
-        "label": {"form_name": "Бухгалтерский баланс", "confidence": 98.0},
-        "organization": {"form_name": "ООО Ромашка", "confidence": 97.5},
-        "inn": {"form_name": "7700000000", "confidence": 99.0},
-        "units": {"form_name": "тыс. руб.", "confidence": 95.0},
-        "date": {"form_name": "30.09.2025", "confidence": 96.0},
-        "table_structure": [{"line_code": "1600", "line_name": "БАЛАНС", "values": ["5678", "-12.3"]}],
-    },
-}
-
 
 # Логи воркера перехватываются (`log_lines` из conftest): вывод pytest чистый
 pytestmark = pytest.mark.usefixtures("log_lines")
 
 
-# --- фейки --------------------------------------------------------------------------------------------------------
-class FakeStorage:
-    """S3 в памяти с интерфейсом `S3Storage`: get_bytes / put_bytes / put_json и ошибками `src.core.clients`."""
-
-    def __init__(self, events: list[tuple[str, str]], bucket: str = BUCKET) -> None:
-        self.events = events
-        self.bucket = bucket
-        self.objects: dict[tuple[str, str], bytes] = {}
-        self.content_types: dict[tuple[str, str], str] = {}
-        self.get_calls: list[tuple[str, str]] = []
-        self.fail_put_prefixes: tuple[str, ...] = ()
-
-    async def get_bytes(self, key: str, *, bucket: str | None = None) -> bytes:
-        bucket_name = bucket or self.bucket
-        self.get_calls.append((bucket_name, key))
-        self.events.append(("s3_get", key))
-        try:
-            return self.objects[(bucket_name, key)]
-        except KeyError:
-            raise S3ObjectNotFoundError(
-                f"S3: объект {bucket_name}/{key} не прочитан (NoSuchKey)", bucket=bucket_name, key=key
-            ) from None
-
-    async def put_bytes(self, key: str, data: bytes, content_type: str = OCTET_STREAM) -> None:
-        if self.fail_put_prefixes and key.startswith(self.fail_put_prefixes):
-            raise S3WriteError(
-                f"S3: объект {self.bucket}/{key} не записан (ConnectionError)", bucket=self.bucket, key=key
-            )
-        self.objects[(self.bucket, key)] = data
-        self.content_types[(self.bucket, key)] = content_type
-        self.events.append(("s3_put", key))
-
-    async def put_json(self, key: str, obj: Any) -> None:
-        await self.put_bytes(key, json.dumps(obj, ensure_ascii=False).encode("utf-8"), JSON_CONTENT_TYPE)
-
-
-class FakeNatsClient:
-    """Подмена `nats.aio.client.Client` для настоящего `NatsGateway`: записывает публикации в общий журнал."""
-
-    def __init__(self, events: list[tuple[str, str]]) -> None:
-        self.events = events
-        self.is_connected = False
-        self.is_closed = False
-        self.published: list[tuple[str, bytes, dict[str, str]]] = []
-        self.fail = False
-        self.block: asyncio.Event | None = None
-
-    async def connect(self, *args: Any, **kwargs: Any) -> None:
-        self.is_connected = True
-
-    async def publish(self, subject: str, payload: bytes = b"", headers: dict[str, str] | None = None) -> None:
-        if self.block is not None:
-            await self.block.wait()
-        if self.fail:
-            raise nats_errors.ConnectionClosedError
-        assert headers is not None
-        self.published.append((subject, payload, headers))
-        self.events.append(("publish", headers[NATS_MSG_ID_HEADER]))
-
-    async def flush(self, timeout: float = 10) -> None:
-        return None
-
-    async def drain(self) -> None:
-        self.is_connected = False
-        self.is_closed = True
-
-    async def close(self) -> None:
-        self.is_connected = False
-        self.is_closed = True
-
-
-class SpyHotFolder:
-    """`LocalHotFolder` под шпионом: журнал записей во входящий каталог, сценарии сбоев `poll`."""
-
-    def __init__(self, inner: LocalHotFolder, events: list[tuple[str, str]]) -> None:
-        self.inner = inner
-        self.events = events
-        self.puts: list[str] = []
-        self.polls = 0
-        self.poll_errors: list[BaseException] = []
-        self.poll_always: BaseException | None = None
-        self.poll_script: list[bytes | None] = []
-
-    def put(self, name: str, data: bytes) -> None:
-        self.inner.put(name, data)
-        self.puts.append(name)
-        self.events.append(("put", name))
-
-    def poll(self, job_id: str) -> bytes | None:
-        self.polls += 1
-        if self.poll_always is not None:
-            raise self.poll_always
-        if self.poll_errors:
-            raise self.poll_errors.pop(0)
-        if self.poll_script:
-            return self.poll_script.pop(0)
-        return self.inner.poll(job_id)
-
-    def healthy(self) -> bool:
-        return self.inner.healthy()
-
-
 class RaisingNormalizer:
     def normalize(self, raw: object) -> NormalizeOutcome:
         raise RuntimeError("нормализатор сломан")
@@ -193,185 +66,6 @@ class StubNormalizer:
         return self.outcome
 
 
-# --- окружение теста ----------------------------------------------------------------------------------------------
-@dataclass
-class Env:
-    settings: Settings
-    root: Path
-    fixtures: Path
-    events: list[tuple[str, str]]
-    storage: FakeStorage
-    client: FakeNatsClient
-    gateway: NatsGateway
-    hotfolder: SpyHotFolder
-    worker: Worker
-    local: LocalHotFolder = field(init=False)
-
-    def __post_init__(self) -> None:
-        self.local = self.hotfolder.inner
-
-    # запрос -------------------------------------------------------------------------------------------------------
-    def request(  # noqa: PLR0913 - параметры запроса для сценариев
-        self,
-        *,
-        original_name: str = "ok_doc.pdf",
-        deadline_s: float = 60,
-        attempt: int = 1,
-        bucket: str = BUCKET,
-        data: bytes | None = PDF_BYTES,
-        job_id: str | None = None,
-    ) -> dict[str, Any]:
-        job = job_id or str(uuid.uuid4())
-        file_id = str(uuid.uuid4())
-        key = f"uploads/{file_id}/{job}/source.bin"
-        if data is not None:
-            self.storage.objects[(bucket, key)] = data
-        deadline = datetime.now(UTC) + timedelta(seconds=deadline_s)
-        return {
-            "schema_version": "1.0",
-            "job_id": job,
-            "file_id": file_id,
-            "pipeline": "image",
-            "file_type": "PDF",
-            "original_name": original_name,
-            "s3": {"bucket": bucket, "key": key},
-            "reply_deadline": deadline.isoformat().replace("+00:00", "Z"),
-            "attempt": attempt,
-        }
-
-    async def deliver(self, body: dict[str, Any] | bytes) -> None:
-        data = body if isinstance(body, bytes) else json.dumps(body, ensure_ascii=False).encode("utf-8")
-        await self.worker.on_request(FakeMsg(data))
-
-    async def run(self, body: dict[str, Any], timeout: float = 5) -> dict[str, Any]:
-        """Доставить запрос, дождаться окончания всех заданий, вернуть единственный итоговый конверт задания."""
-        await self.deliver(body)
-        await asyncio.wait_for(self.worker.wait_idle(), timeout)
-        results = self.results(body["job_id"])
-        assert len(results) == 1, results
-        return results[0][1]
-
-    # публикации ---------------------------------------------------------------------------------------------------
-    def published(self) -> list[tuple[str, dict[str, Any]]]:
-        """(Nats-Msg-Id, тело) всех публикаций; каждое тело проверено схемой конверта."""
-        out = []
-        for subject, payload, headers in self.client.published:
-            assert subject == RESULT_SUBJECT
-            body = json.loads(payload.decode("utf-8"))
-            MESSAGE_VALIDATOR.validate(body)
-            assert body["service"]["name"] == "scan-gateway"
-            assert body["schema_version"] == "1.0"
-            out.append((headers[NATS_MSG_ID_HEADER], body))
-        return out
-
-    def acks(self, job_id: str) -> list[str]:
-        return [mid for mid, body in self.published() if body["job_id"] == job_id and body["status"] == "processing"]
-
-    def results(self, job_id: str) -> list[tuple[str, dict[str, Any]]]:
-        return [
-            (mid, body) for mid, body in self.published() if body["job_id"] == job_id and body["status"] != "processing"
-        ]
-
-    def inbox(self) -> list[str]:
-        if not self.local.input_dir.is_dir():
-            return []
-        return sorted(p.name for p in self.local.input_dir.iterdir() if not p.name.startswith("."))
-
-    def answer(self, job_id: str, content: bytes) -> None:
-        path = self.local.answer_path(job_id)
-        path.parent.mkdir(parents=True, exist_ok=True)
-        path.write_bytes(content)
-
-
-class FakeMsg:
-    def __init__(self, data: bytes) -> None:
-        self.data = data
-        self.subject = "recognition.image.request"
-        self.headers: dict[str, str] | None = None
-
-
-EnvFactory = Callable[..., Awaitable[Env]]
-
-
-@pytest.fixture
-async def make_env(tmp_path: Path) -> AsyncIterator[EnvFactory]:
-    created: list[Env] = []
-
-    async def build(
-        *,
-        normalizer: Any = None,
-        root_exists: bool = True,
-        clock: Callable[[], datetime] | 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")
-        settings = get_settings().model_copy(
-            update={
-                "SCAN_HOTFOLDER_PATH": str(root),
-                "SCAN_POLL_SECONDS": 0.01,
-                "SCAN_TIMEOUT_SECONDS": 5,
-                "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()
-        hotfolder = SpyHotFolder(LocalHotFolder(root, settings.SCAN_INPUT_DIR, settings.SCAN_OUTPUT_DIR), events)
-        kwargs: dict[str, Any] = {}
-        if clock is not None:
-            kwargs["clock"] = clock
-        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()
-
-
-@asynccontextmanager
-async def emulator(env: Env, fixtures: Path | None = None) -> AsyncIterator[None]:
-    """ИС Сканирование: `process_once` (задержка 0) каждые 5 мс, пока открыт контекст."""
-
-    async def loop() -> None:
-        while True:
-            process_once(env.root, fixtures or env.fixtures, 0)
-            await asyncio.sleep(0.005)
-
-    task = asyncio.create_task(loop())
-    try:
-        yield
-    finally:
-        task.cancel()
-        with suppress(asyncio.CancelledError):
-            await task
-
-
-async def eventually(predicate: Callable[[], bool], timeout: float = 3.0) -> None:
-    deadline = asyncio.get_running_loop().time() + timeout
-    while not predicate():
-        if asyncio.get_running_loop().time() > deadline:
-            raise AssertionError("условие не выполнилось вовремя")
-        await asyncio.sleep(0.005)
-
-
 def assert_error(body: dict[str, Any], code: ErrorCode, *, raw: bool) -> None:
     assert body["status"] == "error"
     assert body["error"]["code"] == code.value

api/tests/worker_helpers.py

@@ -0,0 +1,282 @@
+"""Общая обвязка unit-тестов воркера шлюза (`test_worker.py`, `test_worker_concurrency.py`); фикстура `make_env` —
+в `tests/unit/conftest.py`.
+
+Без Docker и сети: S3 — словарь в памяти (ошибки — классы `src.core.clients`), NATS — настоящий `NatsGateway` поверх
+фейкового клиента nats-py (тело, `Nats-Msg-Id`, повторы и `PublishFailed` — как в проде), горячая папка — настоящий
+`LocalHotFolder` на `tmp_path` под шпионом, ИС Сканирование — `process_once` эмулятора (задержка 0) в фоновой задаче.
+Общий журнал событий (`events`) упорядочивает публикации, чтения S3 и запись во входящий каталог.
+Каждый опубликованный конверт проверяется JSON Schema `recognition-result-message.schema.json`.
+"""
+
+import asyncio
+import json
+import uuid
+from collections.abc import AsyncIterator, Awaitable, Callable
+from contextlib import asynccontextmanager, suppress
+from dataclasses import dataclass, field
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+from jsonschema import Draft202012Validator, FormatChecker
+from nats import errors as nats_errors
+
+from src.contract import NATS_MSG_ID_HEADER, RESULT_SUBJECT
+from src.core.clients import JSON_CONTENT_TYPE, OCTET_STREAM, S3ObjectNotFoundError, S3WriteError
+from src.core.config import Settings
+from src.core.nats import NatsGateway
+from src.core.worker import Worker
+from src.gateway.hotfolder import LocalHotFolder
+from src.scan_emulator import process_once
+
+API_DIR = Path(__file__).resolve().parents[1]
+CONTRACTS_DIR = API_DIR / "contracts"
+BUCKET = "arm-ka"
+PDF_BYTES = b"%PDF-1.4\n% scan-gateway test\n\x00\x01\x02binary"
+
+MESSAGE_VALIDATOR = Draft202012Validator(
+    json.loads((CONTRACTS_DIR / "recognition-result-message.schema.json").read_text(encoding="utf-8")),
+    format_checker=FormatChecker(),
+)
+
+# Синтетический успешный ответ ИС (R-10): форма `fixtures/ocr/aep_5.json`, `error = null`, непустой `data_ocr`
+SUCCESS_ANSWER: dict[str, Any] = {
+    "file_uid": "ok_doc",
+    "error": None,
+    "data_ocr": {
+        "label": {"form_name": "Бухгалтерский баланс", "confidence": 98.0},
+        "organization": {"form_name": "ООО Ромашка", "confidence": 97.5},
+        "inn": {"form_name": "7700000000", "confidence": 99.0},
+        "units": {"form_name": "тыс. руб.", "confidence": 95.0},
+        "date": {"form_name": "30.09.2025", "confidence": 96.0},
+        "table_structure": [{"line_code": "1600", "line_name": "БАЛАНС", "values": ["5678", "-12.3"]}],
+    },
+}
+
+
+# --- фейки --------------------------------------------------------------------------------------------------------
+class FakeStorage:
+    """S3 в памяти с интерфейсом `S3Storage`: get_bytes / put_bytes / put_json и ошибками `src.core.clients`."""
+
+    def __init__(self, events: list[tuple[str, str]], bucket: str = BUCKET) -> None:
+        self.events = events
+        self.bucket = bucket
+        self.objects: dict[tuple[str, str], bytes] = {}
+        self.content_types: dict[tuple[str, str], str] = {}
+        self.get_calls: list[tuple[str, str]] = []
+        self.fail_put_prefixes: tuple[str, ...] = ()
+
+    async def get_bytes(self, key: str, *, bucket: str | None = None) -> bytes:
+        bucket_name = bucket or self.bucket
+        self.get_calls.append((bucket_name, key))
+        self.events.append(("s3_get", key))
+        try:
+            return self.objects[(bucket_name, key)]
+        except KeyError:
+            raise S3ObjectNotFoundError(
+                f"S3: объект {bucket_name}/{key} не прочитан (NoSuchKey)", bucket=bucket_name, key=key
+            ) from None
+
+    async def put_bytes(self, key: str, data: bytes, content_type: str = OCTET_STREAM) -> None:
+        if self.fail_put_prefixes and key.startswith(self.fail_put_prefixes):
+            raise S3WriteError(
+                f"S3: объект {self.bucket}/{key} не записан (ConnectionError)", bucket=self.bucket, key=key
+            )
+        self.objects[(self.bucket, key)] = data
+        self.content_types[(self.bucket, key)] = content_type
+        self.events.append(("s3_put", key))
+
+    async def put_json(self, key: str, obj: Any) -> None:
+        await self.put_bytes(key, json.dumps(obj, ensure_ascii=False).encode("utf-8"), JSON_CONTENT_TYPE)
+
+
+class FakeNatsClient:
+    """Подмена `nats.aio.client.Client` для настоящего `NatsGateway`: записывает публикации в общий журнал."""
+
+    def __init__(self, events: list[tuple[str, str]]) -> None:
+        self.events = events
+        self.is_connected = False
+        self.is_closed = False
+        self.published: list[tuple[str, bytes, dict[str, str]]] = []
+        self.fail = False
+        self.block: asyncio.Event | None = None
+
+    async def connect(self, *args: Any, **kwargs: Any) -> None:
+        self.is_connected = True
+
+    async def publish(self, subject: str, payload: bytes = b"", headers: dict[str, str] | None = None) -> None:
+        if self.block is not None:
+            await self.block.wait()
+        if self.fail:
+            raise nats_errors.ConnectionClosedError
+        assert headers is not None
+        self.published.append((subject, payload, headers))
+        self.events.append(("publish", headers[NATS_MSG_ID_HEADER]))
+
+    async def flush(self, timeout: float = 10) -> None:
+        return None
+
+    async def drain(self) -> None:
+        self.is_connected = False
+        self.is_closed = True
+
+    async def close(self) -> None:
+        self.is_connected = False
+        self.is_closed = True
+
+
+class SpyHotFolder:
+    """`LocalHotFolder` под шпионом: журнал записей во входящий каталог, сценарии сбоев `poll`."""
+
+    def __init__(self, inner: LocalHotFolder, events: list[tuple[str, str]]) -> None:
+        self.inner = inner
+        self.events = events
+        self.puts: list[str] = []
+        self.polls = 0
+        self.poll_errors: list[BaseException] = []
+        self.poll_always: BaseException | None = None
+        self.poll_script: list[bytes | None] = []
+
+    def put(self, name: str, data: bytes) -> None:
+        self.inner.put(name, data)
+        self.puts.append(name)
+        self.events.append(("put", name))
+
+    def poll(self, job_id: str) -> bytes | None:
+        self.polls += 1
+        if self.poll_always is not None:
+            raise self.poll_always
+        if self.poll_errors:
+            raise self.poll_errors.pop(0)
+        if self.poll_script:
+            return self.poll_script.pop(0)
+        return self.inner.poll(job_id)
+
+    def healthy(self) -> bool:
+        return self.inner.healthy()
+
+
+# --- окружение теста ----------------------------------------------------------------------------------------------
+@dataclass
+class Env:
+    settings: Settings
+    root: Path
+    fixtures: Path
+    events: list[tuple[str, str]]
+    storage: FakeStorage
+    client: FakeNatsClient
+    gateway: NatsGateway
+    hotfolder: SpyHotFolder
+    worker: Worker
+    local: LocalHotFolder = field(init=False)
+
+    def __post_init__(self) -> None:
+        self.local = self.hotfolder.inner
+
+    # запрос -------------------------------------------------------------------------------------------------------
+    def request(  # noqa: PLR0913 - параметры запроса для сценариев
+        self,
+        *,
+        original_name: str = "ok_doc.pdf",
+        deadline_s: float = 60,
+        attempt: int = 1,
+        bucket: str = BUCKET,
+        data: bytes | None = PDF_BYTES,
+        job_id: str | None = None,
+    ) -> dict[str, Any]:
+        job = job_id or str(uuid.uuid4())
+        file_id = str(uuid.uuid4())
+        key = f"uploads/{file_id}/{job}/source.bin"
+        if data is not None:
+            self.storage.objects[(bucket, key)] = data
+        deadline = datetime.now(UTC) + timedelta(seconds=deadline_s)
+        return {
+            "schema_version": "1.0",
+            "job_id": job,
+            "file_id": file_id,
+            "pipeline": "image",
+            "file_type": "PDF",
+            "original_name": original_name,
+            "s3": {"bucket": bucket, "key": key},
+            "reply_deadline": deadline.isoformat().replace("+00:00", "Z"),
+            "attempt": attempt,
+        }
+
+    async def deliver(self, body: dict[str, Any] | bytes) -> None:
+        data = body if isinstance(body, bytes) else json.dumps(body, ensure_ascii=False).encode("utf-8")
+        await self.worker.on_request(FakeMsg(data))
+
+    async def run(self, body: dict[str, Any], timeout: float = 5) -> dict[str, Any]:
+        """Доставить запрос, дождаться окончания всех заданий, вернуть единственный итоговый конверт задания."""
+        await self.deliver(body)
+        await asyncio.wait_for(self.worker.wait_idle(), timeout)
+        results = self.results(body["job_id"])
+        assert len(results) == 1, results
+        return results[0][1]
+
+    # публикации ---------------------------------------------------------------------------------------------------
+    def published(self) -> list[tuple[str, dict[str, Any]]]:
+        """(Nats-Msg-Id, тело) всех публикаций; каждое тело проверено схемой конверта."""
+        out = []
+        for subject, payload, headers in self.client.published:
+            assert subject == RESULT_SUBJECT
+            body = json.loads(payload.decode("utf-8"))
+            MESSAGE_VALIDATOR.validate(body)
+            assert body["service"]["name"] == "scan-gateway"
+            assert body["schema_version"] == "1.0"
+            out.append((headers[NATS_MSG_ID_HEADER], body))
+        return out
+
+    def acks(self, job_id: str) -> list[str]:
+        return [mid for mid, body in self.published() if body["job_id"] == job_id and body["status"] == "processing"]
+
+    def results(self, job_id: str) -> list[tuple[str, dict[str, Any]]]:
+        return [
+            (mid, body) for mid, body in self.published() if body["job_id"] == job_id and body["status"] != "processing"
+        ]
+
+    def inbox(self) -> list[str]:
+        if not self.local.input_dir.is_dir():
+            return []
+        return sorted(p.name for p in self.local.input_dir.iterdir() if not p.name.startswith("."))
+
+    def answer(self, job_id: str, content: bytes) -> None:
+        path = self.local.answer_path(job_id)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(content)
+
+
+class FakeMsg:
+    def __init__(self, data: bytes) -> None:
+        self.data = data
+        self.subject = "recognition.image.request"
+        self.headers: dict[str, str] | None = None
+
+
+EnvFactory = Callable[..., Awaitable[Env]]
+
+
+@asynccontextmanager
+async def emulator(env: Env, fixtures: Path | None = None) -> AsyncIterator[None]:
+    """ИС Сканирование: `process_once` (задержка 0) каждые 5 мс, пока открыт контекст."""
+
+    async def loop() -> None:
+        while True:
+            process_once(env.root, fixtures or env.fixtures, 0)
+            await asyncio.sleep(0.005)
+
+    task = asyncio.create_task(loop())
+    try:
+        yield
+    finally:
+        task.cancel()
+        with suppress(asyncio.CancelledError):
+            await task
+
+
+async def eventually(predicate: Callable[[], bool], timeout: float = 3.0) -> None:
+    deadline = asyncio.get_running_loop().time() + timeout
+    while not predicate():
+        if asyncio.get_running_loop().time() > deadline:
+            raise AssertionError("условие не выполнилось вовремя")
+        await asyncio.sleep(0.005)