"""Быстрые тесты каркаса приложения: lifespan, /healthcheck, префикс API, клиент S3 (без БД и NATS). Аутентификация и supervisor NATS — tests/test_nats_core.py (подключение переехало в src/core/nats.py, рулинг R-3). """ import asyncio import importlib import threading from http import HTTPStatus from types import SimpleNamespace from typing import Any import httpx import pytest from loguru import logger from src.api.router import router from src.core.nats import NatsConnection from src.services import recognition_exchange # src/__init__.py реэкспортирует объект app под именем модуля, поэтому модуль берём через importlib app_module = importlib.import_module("src.app") @pytest.fixture async def client(): # Без lifespan: app.state.nats не установлен — healthcheck обязан это пережить transport = httpx.ASGITransport(app=app_module.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: yield c async def test_healthcheck_nats_off_when_disabled(client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(app_module.settings, "NATS_ENABLED", False) response = await client.get("/healthcheck") assert response.status_code == HTTPStatus.OK body = response.json() assert body["nats"] == "off" assert {"uptime", "version"} <= set(body) async def test_healthcheck_nats_down_when_enabled_without_connection( client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch ): monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True) response = await client.get("/healthcheck") assert response.status_code == HTTPStatus.OK assert response.json()["nats"] == "down" @pytest.mark.parametrize(("connected", "expected"), [(True, "ok"), (False, "down")]) async def test_healthcheck_nats_follows_connection_state( client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch, connected: bool, expected: str ): monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True) monkeypatch.setattr(app_module.app.state, "nats", SimpleNamespace(is_connected=connected), raising=False) response = await client.get("/healthcheck") assert response.status_code == HTTPStatus.OK assert response.json()["nats"] == expected def test_router_prefix(): assert router.prefix == "/api/v1" class _FakeNats: def __init__(self) -> None: self.started = False self.stopped = False self.is_connected = False async def start(self) -> None: self.started = True async def stop(self) -> None: self.stopped = True class _FakeS3Client: def __init__(self) -> None: self.session_closed = False async def close_session(self) -> None: self.session_closed = True @pytest.fixture def _no_smb(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(app_module.smbclient, "register_session", lambda *args, **kwargs: None) monkeypatch.setattr(app_module.smbclient, "reset_connection_cache", lambda: None) class _FakeEngine: """Подмена движка БД приложения: lifespan вызывает `dispose()` последним шагом остановки (финальное ревью M-3).""" def __init__(self, events: list[str] | None = None) -> None: self.disposed = 0 self.events = events async def dispose(self) -> None: self.disposed += 1 if self.events is not None: self.events.append("db") @pytest.fixture(autouse=True) def fake_engine(monkeypatch: pytest.MonkeyPatch) -> _FakeEngine: """Lifespan в быстрых тестах не трогает пул реального движка БД.""" engine = _FakeEngine() monkeypatch.setattr(app_module, "engine", engine) return engine @pytest.fixture(autouse=True) def exchange_loops(monkeypatch: pytest.MonkeyPatch) -> list[str]: """Циклы публикатора и вахтёра подменены ожиданием: быстрые тесты lifespan не ходят в БД.""" started: list[str] = [] async def publisher(session_factory: Any, nats: Any, **kwargs: Any) -> None: started.append("publisher") await asyncio.Event().wait() async def watchdog(session_factory: Any, **kwargs: Any) -> None: started.append("watchdog") await asyncio.Event().wait() monkeypatch.setattr(recognition_exchange, "run_publisher_loop", publisher) monkeypatch.setattr(recognition_exchange, "run_watchdog_loop", watchdog) return started @pytest.mark.usefixtures("_no_smb") async def test_lifespan_starts_when_s3_and_nats_unavailable(monkeypatch: pytest.MonkeyPatch): async def unavailable() -> None: return None monkeypatch.setattr(app_module, "_connect_s3", unavailable) monkeypatch.setattr(app_module, "_create_nats", lambda: None) reached_yield = False async with app_module.lifespan(app_module.app): reached_yield = True assert app_module.app.state.storage is None assert app_module.app.state.nats is None assert reached_yield @pytest.mark.usefixtures("_no_smb") async def test_lifespan_starts_and_stops_nats_and_closes_s3(monkeypatch: pytest.MonkeyPatch): fake_nats = _FakeNats() s3_client = _FakeS3Client() storage = SimpleNamespace(client=s3_client) async def connect_s3() -> SimpleNamespace: return storage monkeypatch.setattr(app_module, "_connect_s3", connect_s3) monkeypatch.setattr(app_module, "_create_nats", lambda: fake_nats) async with app_module.lifespan(app_module.app): assert app_module.app.state.storage is storage assert app_module.app.state.nats is fake_nats assert fake_nats.started assert not fake_nats.stopped assert not s3_client.session_closed assert fake_nats.stopped assert s3_client.session_closed class _NeverConnectingClient: """Клиент nats-py, который никогда не подключается: connect висит (как nats-py без сервера).""" is_connected = False is_closed = False async def connect(self, servers: str, **options: Any) -> None: await asyncio.Event().wait() async def close(self) -> None: self.is_closed = True @pytest.mark.usefixtures("_no_smb") async def test_lifespan_does_not_wait_for_nats_and_healthcheck_is_down(monkeypatch: pytest.MonkeyPatch): """Регрессия: недоступный NATS не задерживает старт; до первого подключения healthcheck показывает down.""" async def unavailable() -> None: return None nats = NatsConnection(app_module.settings.OPENBAO.NATS, client_factory=_NeverConnectingClient) monkeypatch.setattr(app_module, "_connect_s3", unavailable) monkeypatch.setattr(app_module, "_create_nats", lambda: nats) monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True) async with app_module.lifespan(app_module.app): transport = httpx.ASGITransport(app=app_module.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: response = await c.get("/healthcheck") assert response.json()["nats"] == "down" assert not nats.is_connected def test_create_nats_disabled_returns_none(monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(app_module.settings, "NATS_ENABLED", False) assert app_module._create_nats() is None def test_create_nats_enabled_returns_connection_for_results(monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True) nats = app_module._create_nats() assert isinstance(nats, NatsConnection) assert (nats.result_subject, nats.queue) == ("recognition.result", "arm-core") assert not nats.is_connected # создание не подключает: подключение — в start() async def test_result_subscription_uses_inbox_handler_with_dependencies(): """Подписка `recognition.result` получает `handle_message` с фабрикой сессий приложения и `NoopMapper`.""" subscribed: list[Any] = [] class _FakeConnection: async def subscribe_results(self, handler: Any) -> None: subscribed.append(handler) await app_module._subscribe_results(_FakeConnection()) (handler,) = subscribed assert handler.func is recognition_exchange.handle_message assert handler.args == () assert set(handler.keywords) == {"session_factory", "mapper"} assert handler.keywords["session_factory"] is app_module.async_session_maker assert isinstance(handler.keywords["mapper"], recognition_exchange.NoopMapper) def test_create_nats_subscribes_results_on_connect(monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True) nats = app_module._create_nats() assert nats is not None assert nats._on_connected is app_module._subscribe_results @pytest.mark.parametrize("secure", [True, False]) async def test_connect_s3_passes_secure_flag(monkeypatch: pytest.MonkeyPatch, secure: bool): created: dict[str, Any] = {} class _FakeMinio: def __init__(self, endpoint: str, **kwargs: Any) -> None: created.update(kwargs, endpoint=endpoint) async def bucket_exists(self, bucket: str) -> bool: return True monkeypatch.setattr(app_module, "Minio", _FakeMinio) monkeypatch.setattr(app_module.settings.OPENBAO.MINIO, "MINIO_SECURE", secure) storage = await app_module._connect_s3() assert storage is not None assert created["secure"] is secure class _RecordingNats(_FakeNats): """Запоминает, остановлены ли фоновые задачи обмена к моменту `stop()` соединения NATS.""" def __init__(self) -> None: super().__init__() self.exchange_done_at_stop: bool | None = None async def stop(self) -> None: exchange = app_module.app.state.exchange self.exchange_done_at_stop = all(task.done() for task in exchange.tasks) await super().stop() @pytest.mark.usefixtures("_no_smb") async def test_lifespan_starts_exchange_tasks_and_stops_them_before_nats( monkeypatch: pytest.MonkeyPatch, exchange_loops: list[str] ): """NATS_ENABLED=true: публикатор и вахтёр — две фоновые задачи; остановка — до `nats.stop()`.""" async def unavailable() -> None: return None fake_nats = _RecordingNats() monkeypatch.setattr(app_module, "_connect_s3", unavailable) monkeypatch.setattr(app_module, "_create_nats", lambda: fake_nats) async with app_module.lifespan(app_module.app): exchange = app_module.app.state.exchange assert exchange is not None assert len(exchange.tasks) == 2 # noqa: PLR2004 - публикатор и вахтёр await asyncio.sleep(0) assert sorted(exchange_loops) == ["publisher", "watchdog"] assert not any(task.done() for task in exchange.tasks) assert fake_nats.exchange_done_at_stop is True assert fake_nats.stopped @pytest.mark.usefixtures("_no_smb") async def test_lifespan_without_nats_starts_no_exchange_tasks( monkeypatch: pytest.MonkeyPatch, exchange_loops: list[str] ): """NATS_ENABLED=false (`_create_nats` → None): фоновые задачи обмена не запускаются.""" async def unavailable() -> None: return None monkeypatch.setattr(app_module, "_connect_s3", unavailable) monkeypatch.setattr(app_module.settings, "NATS_ENABLED", False) tasks_before = asyncio.all_tasks() async with app_module.lifespan(app_module.app): assert app_module.app.state.exchange is None await asyncio.sleep(0) assert exchange_loops == [] assert asyncio.all_tasks() - tasks_before == set() # --- SMB необязателен, остановка по независимым шагам (финальное ревью I-3, T4.5, M-3) ------------------------------- async def _unavailable_s3() -> None: return None def test_env_files_leave_smb_off(): """В `.env.test` (и, значит, в тестах) SMB выключен: пустой SMB_HOST.""" assert app_module.settings.OPENBAO.SMB.SMB_HOST == "" async def test_lifespan_skips_smb_when_host_empty(monkeypatch: pytest.MonkeyPatch, fake_engine: _FakeEngine): def forbidden(*args: Any, **kwargs: Any) -> None: raise AssertionError("SMB не настроен — вызова быть не должно") records: list[Any] = [] handler_id = logger.add(lambda message: records.append(message.record), level="DEBUG") monkeypatch.setattr(app_module.settings.OPENBAO.SMB, "SMB_HOST", "") monkeypatch.setattr(app_module.smbclient, "register_session", forbidden) monkeypatch.setattr(app_module.smbclient, "reset_connection_cache", forbidden) monkeypatch.setattr(app_module, "_connect_s3", _unavailable_s3) monkeypatch.setattr(app_module, "_create_nats", lambda: None) try: async with app_module.lifespan(app_module.app): pass finally: logger.remove(handler_id) assert [r for r in records if "SMB не настроен" in r["message"] and r["level"].name == "DEBUG"] assert not [r for r in records if r["level"].name == "ERROR"] assert fake_engine.disposed == 1 async def test_lifespan_registers_smb_in_worker_thread_when_configured(monkeypatch: pytest.MonkeyPatch): """Задан SMB_HOST: `register_session` и `reset_connection_cache` — вне потока цикла событий (`to_thread`).""" loop_thread = threading.get_ident() calls: list[tuple[str, Any, bool]] = [] def register_session(*args: Any, **kwargs: Any) -> None: calls.append(("register", args, threading.get_ident() != loop_thread)) def reset_connection_cache() -> None: calls.append(("reset", (), threading.get_ident() != loop_thread)) smb = app_module.settings.OPENBAO.SMB monkeypatch.setattr(smb, "SMB_HOST", "smb.local") monkeypatch.setattr(smb, "SMB_LOGIN", "user") monkeypatch.setattr(smb, "SMB_PASSWORD", "password") monkeypatch.setattr(app_module.smbclient, "register_session", register_session) monkeypatch.setattr(app_module.smbclient, "reset_connection_cache", reset_connection_cache) monkeypatch.setattr(app_module, "_connect_s3", _unavailable_s3) monkeypatch.setattr(app_module, "_create_nats", lambda: None) async with app_module.lifespan(app_module.app): assert calls == [("register", ("smb.local", "user", "password"), True)] assert calls[-1] == ("reset", (), True) class _FailingNats(_FakeNats): def __init__(self, events: list[str]) -> None: super().__init__() self.events = events async def stop(self) -> None: self.events.append("nats") raise RuntimeError("подставная ошибка остановки NATS") class _FailingS3Client: def __init__(self, events: list[str]) -> None: self.events = events async def close_session(self) -> None: self.events.append("s3") raise RuntimeError("подставная ошибка закрытия S3") async def test_shutdown_steps_are_independent_and_engine_disposed_last( monkeypatch: pytest.MonkeyPatch, exchange_loops: list[str] ): """Ошибки `nats.stop()` и закрытия S3 не отменяют сброс SMB и `engine.dispose()`; порядок шагов сохраняется.""" events: list[str] = [] monkeypatch.setattr(app_module, "engine", _FakeEngine(events)) monkeypatch.setattr(app_module.settings.OPENBAO.SMB, "SMB_HOST", "smb.local") monkeypatch.setattr(app_module.smbclient, "register_session", lambda *args, **kwargs: None) monkeypatch.setattr(app_module.smbclient, "reset_connection_cache", lambda: events.append("smb")) storage = SimpleNamespace(client=_FailingS3Client(events)) async def connect_s3() -> SimpleNamespace: return storage monkeypatch.setattr(app_module, "_connect_s3", connect_s3) monkeypatch.setattr(app_module, "_create_nats", lambda: _FailingNats(events)) records: list[Any] = [] handler_id = logger.add(lambda message: records.append(message.record), level="ERROR") try: async with app_module.lifespan(app_module.app): exchange = app_module.app.state.exchange assert exchange is not None finally: logger.remove(handler_id) assert all(task.done() for task in exchange.tasks) assert events == ["nats", "s3", "smb", "db"] failed_steps = sorted(r["extra"].get("step") for r in records) assert failed_steps == ["nats", "s3"]