"""Unit-тесты lifespan и `/healthcheck` приложения без NATS и MinIO (design D-6, R-17; carry-over T8.7). `NatsConnection`, `_connect_s3` и `RecognitionWorker` модуля `src.app` подменяются фейками: проверяется, что при `NATS_ENABLED=true` воркер запускается и подписывается через `on_connected`, порядок остановки (воркер → NATS → S3) и значения `nats = ok | down | off`. """ import importlib import json from collections.abc import AsyncIterator, Awaitable, Callable, Iterator from contextlib import asynccontextmanager from pathlib import Path from types import SimpleNamespace from typing import Any import httpx import pytest from loguru import logger from pydantic import SecretStr from src.contract import NATS_MSG_ID_HEADER, RESULT_SUBJECT from src.core.clients import S3Storage from src.core.worker import RecognitionWorker EXAMPLES_DIR = Path(__file__).resolve().parents[2] / "contracts" / "examples" REQUEST = json.loads((EXAMPLES_DIR / "request-tabular.json").read_text(encoding="utf-8")) JOB_ID = REQUEST["job_id"] app_module = importlib.import_module("src.app") # модуль, а не объект `app`, который реэкспортирует пакет `src` @asynccontextmanager async def running_app(module: Any) -> AsyncIterator[httpx.AsyncClient]: """Lifespan приложения + HTTP-клиент поверх ASGI (без сети).""" application = module.app async with application.router.lifespan_context(application): transport = httpx.ASGITransport(app=application) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: yield client class FakeStorage: def __init__(self, events: list[str]) -> None: self.events = events self.client = SimpleNamespace(close_session=self._close) self.objects = {(REQUEST["s3"]["bucket"], REQUEST["s3"]["key"]): b"xlsx"} async def _close(self) -> None: self.events.append("s3.close") async def get_bytes(self, bucket: str, key: str) -> bytes: return self.objects[(bucket, key)] async def put_json(self, bucket: str, key: str, obj: Any) -> None: raise AssertionError("байты «xlsx» — не книга: парсер отдаёт ошибку без файла результата") class FakeNatsConnection: """Повторяет интерфейс `NatsConnection`, нужный lifespan и воркеру.""" instances: list["FakeNatsConnection"] = [] connect_on_start = True events: list[str] = [] # общий журнал порядка запуска и остановки (задаётся фикстурой) def __init__( self, settings: Any, *, on_connected: Callable[["FakeNatsConnection"], Awaitable[None]] | None = None, publish_retries: int = 3, ) -> None: self.settings = settings self.on_connected = on_connected self.publish_retries = publish_retries self.is_connected = False self.handlers: list[tuple[str, Callable[[Any], Awaitable[None]]]] = [] self.published: list[tuple[str, dict[str, Any], str]] = [] FakeNatsConnection.instances.append(self) async def start(self) -> None: self.events.append("nats.start") if self.connect_on_start: self.is_connected = True if self.on_connected is not None: await self.on_connected(self) async def stop(self) -> None: self.events.append("nats.stop") self.is_connected = False async def subscribe_requests(self, cb: Callable[[Any], Awaitable[None]]) -> None: self.handlers.append(("files.tabular_uploaded", cb)) async def publish(self, subject: str, payload: bytes, msg_id: str) -> None: self.published.append((subject, json.loads(payload), msg_id)) @pytest.fixture def events() -> list[str]: return [] @pytest.fixture def patched_app(monkeypatch: pytest.MonkeyPatch, events: list[str]) -> Iterator[Any]: """`src.app` с фейковыми S3, NATS и записывающим порядок остановки воркером.""" FakeNatsConnection.instances = [] FakeNatsConnection.connect_on_start = True FakeNatsConnection.events = events storage = FakeStorage(events) async def fake_connect_s3() -> FakeStorage: return storage class RecordingWorker(RecognitionWorker): def start(self) -> None: events.append("worker.start") super().start() async def stop(self, timeout: float = 10.0) -> None: events.append("worker.stop") await super().stop(timeout) monkeypatch.setattr(app_module, "_connect_s3", fake_connect_s3) monkeypatch.setattr(app_module, "NatsConnection", FakeNatsConnection) monkeypatch.setattr(app_module, "RecognitionWorker", RecordingWorker) yield app_module def _expected_subjects(module: Any) -> dict[str, str]: nats = module.settings.OPENBAO.NATS return {"tabular_request": nats.NATS_TABULAR_REQUEST_SUBJECT, "result": nats.NATS_RESULT_SUBJECT} async def test_nats_disabled_starts_nothing_and_healthcheck_reports_off( patched_app: Any, monkeypatch: pytest.MonkeyPatch, events: list[str] ) -> None: monkeypatch.setattr(patched_app.settings, "NATS_ENABLED", False) async with running_app(patched_app) as client: response = await client.get("/healthcheck") assert patched_app.app.state.worker is None assert patched_app.app.state.nats is None assert response.status_code == 200 # noqa: PLR2004 - HTTP 200 body = response.json() assert set(body) == {"uptime", "version", "nats", "subjects"} assert body["subjects"] == _expected_subjects(patched_app) assert body["nats"] == "off" assert body["version"] == patched_app.settings.VERSION assert FakeNatsConnection.instances == [] assert events == ["s3.close"] async def test_minimal_environment_starts_and_healthcheck_answers( patched_app: Any, monkeypatch: pytest.MonkeyPatch, minimal_settings: Callable[..., Any] ) -> None: """Старт с минимальным окружением: только MinIO, адрес NATS и DEBUG (без seed/jwt, БД и SMB).""" monkeypatch.setattr(patched_app, "settings", minimal_settings()) assert patched_app.settings.NATS_ENABLED is True async with running_app(patched_app) as client: response = await client.get("/healthcheck") assert response.status_code == 200 # noqa: PLR2004 - HTTP 200 assert set(response.json()) == {"uptime", "version", "nats", "subjects"} async def test_nats_enabled_subscribes_worker_and_reports_ok( patched_app: Any, monkeypatch: pytest.MonkeyPatch, events: list[str] ) -> None: monkeypatch.setattr(patched_app.settings, "NATS_ENABLED", True) async with running_app(patched_app) as client: assert (await client.get("/healthcheck")).json()["nats"] == "ok" (connection,) = FakeNatsConnection.instances assert connection.publish_retries == patched_app.settings.PUBLISH_RETRIES assert connection.handlers, "воркер должен подписаться через on_connected" worker = patched_app.app.state.worker assert isinstance(worker, RecognitionWorker) # Запрос, доставленный подписке, проходит через воркер с парсером по умолчанию (`get_parser(settings)` → # OsvParser): байты «xlsx» — не книга, ack и результат UNSUPPORTED_FORMAT в files.tabular_parsed _subject, handler = connection.handlers[0] message = SimpleNamespace( data=json.dumps(REQUEST).encode(), headers={NATS_MSG_ID_HEADER: f"{JOB_ID}:1"}, subject=_subject ) await handler(message) await worker.join() assert [(subject, msg_id) for subject, _body, msg_id in connection.published] == [ (RESULT_SUBJECT, f"ack:{JOB_ID}:1"), (RESULT_SUBJECT, f"result:{JOB_ID}:1"), ] assert connection.published[1][1]["error"]["code"] == "UNSUPPORTED_FORMAT" assert connection.published[1][1]["service"]["name"] == "tabular-parser" # Старт: воркер готов до подключения (on_connected сразу подписывает worker.on_request). # Остановка: воркер дорабатывает очередь, пока NATS ещё жив, затем NATS, затем S3 assert events == ["worker.start", "nats.start", "worker.stop", "nats.stop", "s3.close"] async def test_nats_enabled_but_not_connected_reports_down(patched_app: Any, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(patched_app.settings, "NATS_ENABLED", True) FakeNatsConnection.connect_on_start = False async with running_app(patched_app) as client: response = await client.get("/healthcheck") assert response.json()["nats"] == "down" assert response.json()["subjects"] == _expected_subjects(patched_app) assert "jwt" not in response.text.lower() (connection,) = FakeNatsConnection.instances assert connection.handlers == [] # подписка — только после подключения assert patched_app.app.state.worker is not None # --- subject'ы из настроек, JWT вне ответа, лог старта, передача result_subject воркеру ------------------------------ CUSTOM_SUBJECTS = { "NATS_TABULAR_REQUEST_SUBJECT": "test.files.tabular_uploaded", "NATS_RESULT_SUBJECT": "test.files.tabular_parsed", } SECRET_JWT = "eyJ0eXAiOiJKV1QiLCJhbGciOiJlZDI1NTE5LW5rZXkifQ.healthcheck-secret.sig" @pytest.fixture def custom_nats_settings(patched_app: Any, monkeypatch: pytest.MonkeyPatch) -> Any: """Подменяет `settings.OPENBAO.NATS` копией с нестандартными subject'ами и непустым JWT.""" nats = patched_app.settings.OPENBAO.NATS.model_copy(update={**CUSTOM_SUBJECTS, "NATS_JWT": SecretStr(SECRET_JWT)}) monkeypatch.setattr(patched_app.settings.OPENBAO, "NATS", nats) return nats @pytest.mark.parametrize(("nats_enabled", "state"), [(False, "off"), (True, "down")]) async def test_healthcheck_reports_configured_subjects_and_hides_jwt( patched_app: Any, custom_nats_settings: Any, monkeypatch: pytest.MonkeyPatch, nats_enabled: bool, state: str, ) -> None: monkeypatch.setattr(patched_app.settings, "NATS_ENABLED", nats_enabled) FakeNatsConnection.connect_on_start = False # state=down: NATS включён, но связи нет async with running_app(patched_app) as client: response = await client.get("/healthcheck") body = response.json() assert body["nats"] == state assert body["subjects"] == { "tabular_request": "test.files.tabular_uploaded", "result": "test.files.tabular_parsed", } assert SECRET_JWT not in response.text assert SECRET_JWT.split(".")[1] not in response.text async def test_startup_logs_configured_subjects( patched_app: Any, custom_nats_settings: Any, monkeypatch: pytest.MonkeyPatch, log_records: list[Any] ) -> None: monkeypatch.setattr(patched_app.settings, "NATS_ENABLED", False) async with running_app(patched_app): pass messages = [record["message"] for record in log_records if "Subject'ы NATS" in record["message"]] assert len(messages) == 1 assert "test.files.tabular_uploaded" in messages[0] assert "test.files.tabular_parsed" in messages[0] assert SECRET_JWT not in messages[0] async def test_build_worker_passes_configured_result_subject( patched_app: Any, custom_nats_settings: Any, monkeypatch: pytest.MonkeyPatch ) -> None: """Воркер публикует ack и результат в настроенный `NATS_RESULT_SUBJECT`, а не в константу контракта.""" nats, worker = patched_app._build_worker(FakeStorage([])) assert nats.settings.NATS_RESULT_SUBJECT == "test.files.tabular_parsed" assert worker._result_subject == "test.files.tabular_parsed" # noqa: SLF001 - проверяем проводку настроек # --- проверка бакета при старте (_connect_s3) ------------------------------------------------------------------------ @pytest.fixture def log_records() -> Iterator[list[Any]]: records: list[Any] = [] handler_id = logger.add(lambda message: records.append(message.record), level="DEBUG") yield records logger.remove(handler_id) class FakeMinio: """Подмена `miniopy_async.Minio`: `bucket_exists` отдаёт заданный ответ или бросает исключение.""" answer: bool | Exception = True closed = 0 def __init__(self, *_args: Any, **_kwargs: Any) -> None: pass async def bucket_exists(self, bucket: str) -> bool: if isinstance(FakeMinio.answer, Exception): raise FakeMinio.answer return FakeMinio.answer async def close_session(self) -> None: FakeMinio.closed += 1 @pytest.mark.parametrize( ("answer", "level", "fragment"), [ pytest.param(True, "INFO", "Бакет проверен", id="bucket-exists"), pytest.param(False, "WARNING", "Бакет не найден", id="bucket-missing"), pytest.param(ConnectionError("MinIO недоступен"), "ERROR", "Не удалось проверить", id="minio-down"), ], ) async def test_connect_s3_logs_bucket_check_and_keeps_client( monkeypatch: pytest.MonkeyPatch, log_records: list[Any], answer: Any, level: str, fragment: str ) -> None: monkeypatch.setattr(FakeMinio, "answer", answer) monkeypatch.setattr(FakeMinio, "closed", 0) monkeypatch.setattr(app_module, "Minio", FakeMinio) storage = await app_module._connect_s3() # Клиент остаётся в любом случае: MinIO может подняться позже, воркер прочитает файл без рестарта assert isinstance(storage, S3Storage) assert storage.bucket == app_module.settings.OPENBAO.MINIO.MINIO_BUCKET assert FakeMinio.closed == 0 messages = [record["message"] for record in log_records if record["level"].name == level] assert any(fragment in message for message in messages), messages