3d31c1eac5
- OPENBAO__SMB__* необязательны (по умолчанию пусто); при пустом SMB_HOST сессия SMB не регистрируется (DEBUG «SMB не настроен»), в .env.example и .env.test SMB выключен — в логе dev-запуска больше нет трейсбэка SMB. - Задан SMB_HOST: register_session и reset_connection_cache вызываются через asyncio.to_thread и не блокируют цикл событий. - Остановка: задачи обмена -> NATS -> S3 -> SMB -> engine.dispose(); каждый шаг в своём try/except с логом step=..., ошибка одного шага не отменяет следующие. Пул соединений БД закрывается последним (M-3). - README: «что в логе при запуске», таблица настроек SMB, порядок остановки. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
436 lines
17 KiB
Python
436 lines
17 KiB
Python
"""Быстрые тесты каркаса приложения: 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"]
|