09f0bb32e2
src/core/health.py: HealthProvider — nats ok/down/disabled по is_connected; hotfolder — healthy() в потоке под wait_for (2 с) с кэшем 5 с, зависшая проба не запускается повторно (T8.4). Сбой регистрации SMB-сессии виден естественно: SmbHotFolder.healthy() обращается к шаре с теми же учётными данными (T2.3). app.py: в lifespan S3Storage, NatsGateway, Worker(NullNormalizer); подключение к NATS и подписка — фоновой задачей (connect ждёт сервер), её сбой — logger.exception и nats=down без падения HTTP; остановка отменяет подключение, останавливает воркер, закрывает NATS и S3. Объекты lifespan — в app.state. test_app.py: тесты lifespan переведены на NATS_ENABLED=false и tmp-корень (иначе фоновое подключение шло бы в compose NATS), проверка полей healthcheck расширена nats/hotfolder. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
218 lines
8.8 KiB
Python
218 lines
8.8 KiB
Python
"""Приложение: `/healthcheck` (`uptime`, `version`, `nats`, `hotfolder`), lifespan (SMB-сессия, S3, NATS в фоне,
|
||
воркер). NATS и S3 не нужны: в тестах lifespan `NATS_ENABLED=false` либо методы `NatsGateway` подменены."""
|
||
|
||
import asyncio
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from unittest.mock import MagicMock
|
||
|
||
import httpx
|
||
import pytest
|
||
import smbclient
|
||
|
||
from src.app import create_app
|
||
from src.core.config import get_settings
|
||
from src.core.nats import NatsGateway
|
||
from src.core.worker import Worker
|
||
|
||
pytestmark = pytest.mark.usefixtures("log_lines")
|
||
|
||
|
||
def _client(app) -> httpx.AsyncClient:
|
||
return httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test")
|
||
|
||
|
||
@pytest.fixture
|
||
def local_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||
"""Режим local с существующим корнем и без NATS: lifespan не ходит в сеть."""
|
||
monkeypatch.setenv("SCAN_HOTFOLDER_MODE", "local")
|
||
monkeypatch.setenv("SCAN_HOTFOLDER_PATH", str(tmp_path))
|
||
monkeypatch.setenv("NATS_ENABLED", "false")
|
||
return tmp_path
|
||
|
||
|
||
async def eventually(predicate, timeout: float = 2.0) -> None:
|
||
loop = asyncio.get_running_loop()
|
||
deadline = loop.time() + timeout
|
||
while not predicate():
|
||
assert loop.time() < deadline, "условие не выполнилось вовремя"
|
||
await asyncio.sleep(0.005)
|
||
|
||
|
||
async def test_healthcheck_without_lifespan_has_all_fields(local_env: Path) -> None:
|
||
app = create_app(get_settings())
|
||
|
||
# lifespan не запускается: подключений к NATS/S3 нет
|
||
async with _client(app) as client:
|
||
response = await client.get("/healthcheck")
|
||
|
||
assert response.status_code == 200
|
||
body = response.json()
|
||
assert set(body) == {"uptime", "version", "nats", "hotfolder"}
|
||
assert body["version"] == get_settings().VERSION
|
||
datetime.fromisoformat(body["uptime"])
|
||
|
||
|
||
async def test_start_in_local_mode(local_env: Path) -> None:
|
||
"""Spec «Старт в режиме локальных папок»: 200 с uptime, version, nats, hotfolder = ok; OPENBAO__SMB__* не нужны."""
|
||
app = create_app(get_settings())
|
||
async with app.router.lifespan_context(app), _client(app) as client:
|
||
response = await client.get("/healthcheck")
|
||
assert response.status_code == 200
|
||
body = response.json()
|
||
assert body["hotfolder"] == "ok"
|
||
assert body["nats"] == "disabled"
|
||
assert body["version"] == get_settings().VERSION
|
||
|
||
|
||
async def test_missing_hotfolder_root_is_down(local_env: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setenv("SCAN_HOTFOLDER_PATH", str(local_env / "missing"))
|
||
app = create_app(get_settings())
|
||
async with app.router.lifespan_context(app), _client(app) as client:
|
||
assert (await client.get("/healthcheck")).json()["hotfolder"] == "down"
|
||
|
||
|
||
async def test_lifespan_starts_and_stops_worker(local_env: Path) -> None:
|
||
app = create_app(get_settings())
|
||
async with app.router.lifespan_context(app):
|
||
worker = app.state.worker
|
||
assert isinstance(worker, Worker)
|
||
assert isinstance(app.state.nats, NatsGateway)
|
||
# после остановки воркер запросов не принимает
|
||
await worker.on_request(type("Msg", (), {"data": b"{}"})())
|
||
assert worker.active_jobs == 0
|
||
|
||
|
||
async def test_lifespan_does_not_register_smb_session_in_local_mode(local_env: Path, monkeypatch) -> None:
|
||
register = MagicMock()
|
||
monkeypatch.setattr(smbclient, "register_session", register)
|
||
app = create_app(get_settings())
|
||
|
||
async with app.router.lifespan_context(app), _client(app) as client:
|
||
assert (await client.get("/healthcheck")).status_code == 200
|
||
|
||
register.assert_not_called()
|
||
|
||
|
||
def _smb_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setenv("SCAN_HOTFOLDER_MODE", "smb")
|
||
monkeypatch.setenv("OPENBAO__SMB__SMB_HOST", "smb.example")
|
||
monkeypatch.setenv("OPENBAO__SMB__SMB_PATH", "share")
|
||
monkeypatch.setenv("OPENBAO__SMB__SMB_LOGIN", "user")
|
||
monkeypatch.setenv("OPENBAO__SMB__SMB_PASSWORD", "secret")
|
||
monkeypatch.setenv("NATS_ENABLED", "false")
|
||
|
||
|
||
async def test_lifespan_registers_smb_session_in_smb_mode(monkeypatch) -> None:
|
||
_smb_env(monkeypatch)
|
||
register = MagicMock()
|
||
reset = MagicMock()
|
||
monkeypatch.setattr(smbclient, "register_session", register)
|
||
monkeypatch.setattr(smbclient, "reset_connection_cache", reset)
|
||
monkeypatch.setattr(smbclient, "listdir", MagicMock(return_value=["Input", "Output"]))
|
||
app = create_app(get_settings())
|
||
|
||
async with app.router.lifespan_context(app), _client(app) as client:
|
||
response = await client.get("/healthcheck")
|
||
assert response.status_code == 200
|
||
assert response.json()["hotfolder"] == "ok"
|
||
|
||
register.assert_called_once_with("smb.example", "user", "secret")
|
||
reset.assert_called_once()
|
||
|
||
|
||
async def test_smb_session_failure_shows_hotfolder_down(monkeypatch, log_lines: list[str]) -> None:
|
||
"""T2.3: регистрация SMB-сессии при старте не удалась (шара недоступна) — `hotfolder = down`, сервис жив."""
|
||
_smb_env(monkeypatch)
|
||
monkeypatch.setattr(smbclient, "register_session", MagicMock(side_effect=OSError("сервер недоступен")))
|
||
monkeypatch.setattr(smbclient, "listdir", MagicMock(side_effect=OSError("сервер недоступен")))
|
||
app = create_app(get_settings())
|
||
|
||
async with app.router.lifespan_context(app), _client(app) as client:
|
||
response = await client.get("/healthcheck")
|
||
|
||
assert response.status_code == 200
|
||
assert response.json()["hotfolder"] == "down"
|
||
assert any("SMB" in line for line in log_lines if line.startswith("ERROR|"))
|
||
|
||
|
||
async def test_nats_connects_in_background_and_subscribes_worker(local_env: Path, monkeypatch) -> None:
|
||
monkeypatch.setenv("NATS_ENABLED", "true")
|
||
subscribed: list[Any] = []
|
||
connected = asyncio.Event()
|
||
|
||
async def fake_connect(self: NatsGateway) -> bool:
|
||
await connected.wait() # подключение «висит», а HTTP уже отвечает
|
||
return True
|
||
|
||
async def fake_subscribe(self: NatsGateway, cb: Any, queue: str = "scan-gateway") -> None:
|
||
subscribed.append(cb)
|
||
|
||
async def fake_close(self: NatsGateway) -> None:
|
||
return None
|
||
|
||
monkeypatch.setattr(NatsGateway, "connect", fake_connect)
|
||
monkeypatch.setattr(NatsGateway, "subscribe_requests", fake_subscribe)
|
||
monkeypatch.setattr(NatsGateway, "close", fake_close)
|
||
app = create_app(get_settings())
|
||
|
||
async with app.router.lifespan_context(app), _client(app) as client:
|
||
assert (await client.get("/healthcheck")).json()["nats"] == "down"
|
||
connected.set()
|
||
await eventually(lambda: bool(subscribed))
|
||
assert subscribed == [app.state.worker.on_request]
|
||
|
||
|
||
async def test_nats_connect_hanging_is_cancelled_on_shutdown(local_env: Path, monkeypatch) -> None:
|
||
monkeypatch.setenv("NATS_ENABLED", "true")
|
||
cancelled = asyncio.Event()
|
||
|
||
async def fake_connect(self: NatsGateway) -> bool:
|
||
try:
|
||
await asyncio.Event().wait()
|
||
except asyncio.CancelledError:
|
||
cancelled.set()
|
||
raise
|
||
return True
|
||
|
||
async def fake_close(self: NatsGateway) -> None:
|
||
return None
|
||
|
||
monkeypatch.setattr(NatsGateway, "connect", fake_connect)
|
||
monkeypatch.setattr(NatsGateway, "close", fake_close)
|
||
app = create_app(get_settings())
|
||
async with app.router.lifespan_context(app):
|
||
await asyncio.sleep(0.01)
|
||
assert cancelled.is_set()
|
||
|
||
|
||
async def test_nats_config_error_is_loud_and_shows_down(local_env: Path, monkeypatch, log_lines: list[str]) -> None:
|
||
"""Неполные учётные данные NATS (`NatsConfigError`): лог error с трассировкой, `nats = down`, HTTP жив."""
|
||
monkeypatch.setenv("NATS_ENABLED", "true")
|
||
monkeypatch.setenv("OPENBAO__NATS__NATS_SEED", "SUAINCOMPLETESEED")
|
||
monkeypatch.setenv("OPENBAO__NATS__NATS_JWT", "")
|
||
app = create_app(get_settings())
|
||
|
||
async with app.router.lifespan_context(app), _client(app) as client:
|
||
await eventually(lambda: any(line.startswith("ERROR|") and "NATS" in line for line in log_lines))
|
||
response = await client.get("/healthcheck")
|
||
|
||
assert response.status_code == 200
|
||
assert response.json()["nats"] == "down"
|
||
|
||
|
||
async def test_unhandled_error_has_json_500(local_env: Path) -> None:
|
||
app = create_app(get_settings())
|
||
|
||
@app.get("/boom")
|
||
async def boom():
|
||
raise RuntimeError("x")
|
||
|
||
transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||
response = await client.get("/boom")
|
||
|
||
assert response.status_code == 500
|
||
assert response.json() == {"detail": "внутренняя ошибка сервера"}
|