Срез 0 [5.3]: /healthcheck.nats и .hotfolder, воркер и NATS в lifespan, отметка задачи
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>
This commit is contained in:
Andrey Epifancev
parent
ea49abd03b
commit
09f0bb32e2
@ -1,12 +1,18 @@
|
||||
"""Точка входа FastAPI сервиса scan-gateway (адаптер ИС Сканирование). Без БД и без фронта.
|
||||
|
||||
Что здесь есть: настройки, регистрация SMB-сессии (только при `SCAN_HOTFOLDER_MODE=smb`), middleware (gzip, proxy
|
||||
headers), обработчики ошибок, `GET /healthcheck` (`uptime`, `version`).
|
||||
Чего здесь пока нет: NATS-воркера и S3-клиента (подключаются в lifespan задачами 5.2/5.3 change slice-0-skeleton),
|
||||
полей `nats`/`hotfolder` в healthcheck.
|
||||
- `create_app(settings)` строит горячую папку (`create_hotfolder`, без обращения к ФС/шаре) и `HealthProvider`;
|
||||
`GET /healthcheck` отдаёт `uptime`, `version`, `nats` (`ok`/`down`/`disabled`), `hotfolder` (`ok`/`down`).
|
||||
- lifespan: регистрация SMB-сессии (только `SCAN_HOTFOLDER_MODE=smb`), S3 (`S3Storage`), NATS (`NatsGateway`), воркер
|
||||
(`Worker` с `NullNormalizer`). Подключение к NATS блокирует до доступности сервера, поэтому идёт фоновой задачей:
|
||||
HTTP отвечает сразу (`nats = down`), подписка `recognition.image.request` появляется после подключения. Сбой этой
|
||||
задачи (например, `NatsConfigError`) — лог error с трассировкой и `nats = down`, HTTP-сервер жив.
|
||||
Остановка: отмена подключения, остановка воркера (задания в работе отменяются), закрытие NATS и S3.
|
||||
- Объекты lifespan доступны в `app.state`: `worker`, `nats`, `storage`, `health` (тесты, задачи 13–14).
|
||||
- `NATS_ENABLED=false` — подключения нет (`nats = disabled`), воркер запросов не получает (dev без NATS).
|
||||
"""
|
||||
|
||||
from contextlib import asynccontextmanager
|
||||
import asyncio
|
||||
from contextlib import asynccontextmanager, suppress
|
||||
from datetime import UTC, datetime
|
||||
|
||||
import smbclient
|
||||
@ -16,8 +22,14 @@ from starlette.middleware.gzip import GZipMiddleware
|
||||
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
|
||||
|
||||
from src.api.router import router
|
||||
from src.core.clients import S3Storage
|
||||
from src.core.config import Settings, get_settings
|
||||
from src.core.exception_handlers import register_exception_handlers
|
||||
from src.core.health import HealthProvider
|
||||
from src.core.nats import NatsGateway
|
||||
from src.core.worker import Worker
|
||||
from src.gateway import create_hotfolder
|
||||
from src.normalizer import NullNormalizer
|
||||
|
||||
|
||||
def _normalize_root_path(value: str) -> str:
|
||||
@ -29,10 +41,35 @@ def _normalize_root_path(value: str) -> str:
|
||||
return "/" + value.strip("/")
|
||||
|
||||
|
||||
async def run_nats(gateway: NatsGateway, worker: Worker) -> None:
|
||||
"""Фоновая задача: подключение к NATS (ждёт сервер сколь угодно долго) и подписка воркера на запросы.
|
||||
|
||||
Ошибка не роняет приложение: лог error с трассировкой; если соединение установлено, а подписка не удалась,
|
||||
соединение закрывается, чтобы `/healthcheck.nats = down` честно показывал, что запросы не принимаются.
|
||||
"""
|
||||
try:
|
||||
if not await gateway.connect():
|
||||
return
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception:
|
||||
logger.exception("NATS: подключение не выполнено, запросы не принимаются (nats=down) до перезапуска")
|
||||
return
|
||||
try:
|
||||
await gateway.subscribe_requests(worker.on_request)
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception:
|
||||
logger.exception("NATS: подписка на запросы не оформлена, соединение закрывается (nats=down)")
|
||||
await gateway.close()
|
||||
|
||||
|
||||
def create_app(settings: Settings | None = None) -> FastAPI:
|
||||
settings = settings or get_settings()
|
||||
version = settings.VERSION
|
||||
up_time = datetime.now(UTC)
|
||||
hotfolder = create_hotfolder(settings)
|
||||
health = HealthProvider(nats_enabled=settings.NATS_ENABLED, hotfolder=hotfolder)
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
@ -43,20 +80,38 @@ def create_app(settings: Settings | None = None) -> FastAPI:
|
||||
smbclient.register_session(smb.SMB_HOST, smb.SMB_LOGIN, smb.SMB_PASSWORD)
|
||||
smb_session = True
|
||||
except Exception:
|
||||
# учётные данные передаются в каждый вызов smbclient (SmbHotFolder), сессия восстановится при
|
||||
# доступной шаре; пока шара недоступна, /healthcheck.hotfolder = down
|
||||
logger.exception("Ошибка при регистрации сессии SMB")
|
||||
storage = S3Storage.from_settings(settings.OPENBAO.MINIO)
|
||||
gateway = NatsGateway(settings)
|
||||
worker = Worker(settings, storage=storage, publisher=gateway, hotfolder=hotfolder, normalizer=NullNormalizer())
|
||||
health.nats = gateway
|
||||
app.state.storage = storage
|
||||
app.state.nats = gateway
|
||||
app.state.worker = worker
|
||||
worker.start()
|
||||
nats_task = asyncio.create_task(run_nats(gateway, worker), name="nats-connect")
|
||||
try:
|
||||
logger.info("Приложение запущено")
|
||||
yield
|
||||
finally:
|
||||
logger.info("Приложение закрывается")
|
||||
nats_task.cancel()
|
||||
with suppress(asyncio.CancelledError):
|
||||
await nats_task
|
||||
await worker.stop()
|
||||
await gateway.close()
|
||||
await storage.close()
|
||||
if smb_session:
|
||||
smbclient.reset_connection_cache()
|
||||
|
||||
app = FastAPI(title="scan-gateway", version=version, root_path=settings.ROOT_PATH, lifespan=lifespan)
|
||||
app.state.health = health
|
||||
|
||||
@app.get("/healthcheck", status_code=200)
|
||||
async def healthcheck():
|
||||
return {"uptime": up_time, "version": version}
|
||||
return {"uptime": up_time, "version": version, **await health.snapshot()}
|
||||
|
||||
app.include_router(router)
|
||||
register_exception_handlers(app)
|
||||
|
||||
96
api/src/core/health.py
Normal file
96
api/src/core/health.py
Normal file
@ -0,0 +1,96 @@
|
||||
"""Состояние сервиса для `GET /healthcheck`: поля `nats` и `hotfolder` (D-8; spec image-recognition-gateway).
|
||||
|
||||
- `nats`: `disabled` при `NATS_ENABLED=false`; `ok`, если соединение установлено (`is_connected`), иначе `down` — в том
|
||||
числе пока идёт первое подключение, во время переподключения и после сбоя фоновой задачи подключения.
|
||||
- `hotfolder`: `HotFolder.healthy()` в потоке (`asyncio.to_thread`) под `wait_for` с коротким таймаутом и с кэшем на
|
||||
короткое время (T8.4: подключение к недоступной SMB-шаре ждёт до 60 с, проба не должна висеть и плодить потоки).
|
||||
Пока прежняя проверка не вернулась, новая не запускается — ответ `down`.
|
||||
В режиме smb сбой регистрации сессии при старте виден естественно: `SmbHotFolder.healthy()` обращается к шаре с теми
|
||||
же учётными данными и при недоступной шаре или неверном логине возвращает `False` (T2.3).
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from typing import Literal, Protocol
|
||||
|
||||
from loguru import logger
|
||||
|
||||
from src.gateway.hotfolder import HotFolder
|
||||
|
||||
__all__ = ["HOTFOLDER_CACHE_TTL", "HOTFOLDER_CHECK_TIMEOUT", "ConnectionState", "HealthProvider"]
|
||||
|
||||
HOTFOLDER_CHECK_TIMEOUT = 2.0 # секунд на `healthy()`; дольше — `down`
|
||||
HOTFOLDER_CACHE_TTL = 5.0 # секунд, в течение которых повторная проба отдаёт прежний результат
|
||||
|
||||
Status = Literal["ok", "down"]
|
||||
NatsStatus = Literal["ok", "down", "disabled"]
|
||||
|
||||
|
||||
class ConnectionState(Protocol):
|
||||
"""Источник состояния NATS (`NatsGateway`)."""
|
||||
|
||||
@property
|
||||
def is_connected(self) -> bool: ...
|
||||
|
||||
|
||||
class HealthProvider:
|
||||
"""Поля `nats`/`hotfolder` для `/healthcheck`. `nats` присваивается, когда соединение создано (lifespan)."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
nats_enabled: bool,
|
||||
hotfolder: HotFolder | None,
|
||||
check_timeout: float = HOTFOLDER_CHECK_TIMEOUT,
|
||||
cache_ttl: float = HOTFOLDER_CACHE_TTL,
|
||||
clock: Callable[[], float] = time.monotonic,
|
||||
) -> None:
|
||||
self.nats: ConnectionState | None = None
|
||||
self._nats_enabled = nats_enabled
|
||||
self._hotfolder = hotfolder
|
||||
self._check_timeout = check_timeout
|
||||
self._cache_ttl = cache_ttl
|
||||
self._clock = clock
|
||||
self._cached: Status | None = None
|
||||
self._cached_at = 0.0
|
||||
self._inflight: asyncio.Task[bool] | None = None
|
||||
|
||||
def nats_status(self) -> NatsStatus:
|
||||
if not self._nats_enabled:
|
||||
return "disabled"
|
||||
if self.nats is not None and self.nats.is_connected:
|
||||
return "ok"
|
||||
return "down"
|
||||
|
||||
async def hotfolder_status(self) -> Status:
|
||||
if self._hotfolder is None:
|
||||
return "down"
|
||||
if self._cached is not None and self._clock() - self._cached_at < self._cache_ttl:
|
||||
return self._cached
|
||||
if self._inflight is None or self._inflight.done():
|
||||
self._inflight = asyncio.create_task(asyncio.to_thread(self._hotfolder.healthy))
|
||||
self._inflight.add_done_callback(_consume_exception)
|
||||
status: Status
|
||||
try:
|
||||
healthy = await asyncio.wait_for(asyncio.shield(self._inflight), self._check_timeout)
|
||||
status = "ok" if healthy is True else "down"
|
||||
except TimeoutError:
|
||||
logger.warning(
|
||||
"Проверка горячей папки не уложилась в {timeout} с: hotfolder=down", timeout=self._check_timeout
|
||||
)
|
||||
status = "down"
|
||||
except Exception as exc:
|
||||
logger.warning("Проверка горячей папки упала: hotfolder=down. error_type={type}", type=type(exc).__name__)
|
||||
status = "down"
|
||||
self._cached, self._cached_at = status, self._clock()
|
||||
return status
|
||||
|
||||
async def snapshot(self) -> dict[str, str]:
|
||||
return {"nats": self.nats_status(), "hotfolder": await self.hotfolder_status()}
|
||||
|
||||
|
||||
def _consume_exception(task: "asyncio.Task[bool]") -> None:
|
||||
"""Исключение пробы, которую никто не дождался (таймаут), не должно попадать в лог asyncio как «не извлечённое»."""
|
||||
if not task.cancelled():
|
||||
task.exception()
|
||||
@ -1,18 +1,46 @@
|
||||
"""Приложение: `/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")
|
||||
|
||||
|
||||
async def test_healthcheck_returns_uptime_and_version_without_infrastructure(monkeypatch):
|
||||
@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 нет
|
||||
@ -21,13 +49,42 @@ async def test_healthcheck_returns_uptime_and_version_without_infrastructure(mon
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert set(body) == {"uptime", "version"}
|
||||
assert set(body) == {"uptime", "version", "nats", "hotfolder"}
|
||||
assert body["version"] == get_settings().VERSION
|
||||
datetime.fromisoformat(body["uptime"])
|
||||
|
||||
|
||||
async def test_lifespan_does_not_register_smb_session_in_local_mode(monkeypatch):
|
||||
monkeypatch.setenv("SCAN_HOTFOLDER_MODE", "local")
|
||||
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())
|
||||
@ -38,26 +95,114 @@ async def test_lifespan_does_not_register_smb_session_in_local_mode(monkeypatch)
|
||||
register.assert_not_called()
|
||||
|
||||
|
||||
async def test_lifespan_registers_smb_session_in_smb_mode(monkeypatch):
|
||||
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:
|
||||
assert (await client.get("/healthcheck")).status_code == 200
|
||||
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_unhandled_error_has_json_500(monkeypatch):
|
||||
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")
|
||||
|
||||
120
api/tests/unit/test_health.py
Normal file
120
api/tests/unit/test_health.py
Normal file
@ -0,0 +1,120 @@
|
||||
"""Состояние для `/healthcheck` (`src.core.health`): поля `nats` и `hotfolder` (spec «Старт в режиме локальных папок»,
|
||||
«Горячая папка недоступна», «Соединение с NATS без JetStream»; D-8; T8.4 — проверка не зависает, T2.3 — сбой SMB)."""
|
||||
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from src.core.health import HealthProvider
|
||||
from src.gateway.hotfolder import LocalHotFolder
|
||||
|
||||
pytestmark = pytest.mark.usefixtures("log_lines")
|
||||
|
||||
|
||||
class Connection:
|
||||
def __init__(self, connected: bool) -> None:
|
||||
self.is_connected = connected
|
||||
|
||||
|
||||
class CountingHotFolder:
|
||||
"""`healthy()` считает вызовы; может зависнуть до события или бросить исключение."""
|
||||
|
||||
def __init__(self, result: bool = True) -> None:
|
||||
self.result = result
|
||||
self.calls = 0
|
||||
self.release: threading.Event | None = None
|
||||
self.error: BaseException | None = None
|
||||
|
||||
def put(self, name: str, data: bytes) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
def poll(self, job_id: str) -> bytes | None:
|
||||
raise NotImplementedError
|
||||
|
||||
def healthy(self) -> bool:
|
||||
self.calls += 1
|
||||
if self.release is not None:
|
||||
self.release.wait(5)
|
||||
if self.error is not None:
|
||||
raise self.error
|
||||
return self.result
|
||||
|
||||
|
||||
class Clock:
|
||||
def __init__(self) -> None:
|
||||
self.now = 1000.0
|
||||
|
||||
def __call__(self) -> float:
|
||||
return self.now
|
||||
|
||||
|
||||
async def test_local_root_exists_is_ok(tmp_path: Path) -> None:
|
||||
health = HealthProvider(nats_enabled=False, hotfolder=LocalHotFolder(tmp_path))
|
||||
assert await health.snapshot() == {"nats": "disabled", "hotfolder": "ok"}
|
||||
|
||||
|
||||
async def test_missing_root_is_down(tmp_path: Path) -> None:
|
||||
health = HealthProvider(nats_enabled=False, hotfolder=LocalHotFolder(tmp_path / "missing"))
|
||||
assert await health.hotfolder_status() == "down"
|
||||
|
||||
|
||||
async def test_no_hotfolder_is_down() -> None:
|
||||
assert await HealthProvider(nats_enabled=False, hotfolder=None).hotfolder_status() == "down"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("enabled", "connection", "expected"),
|
||||
[
|
||||
(False, None, "disabled"),
|
||||
(False, Connection(True), "disabled"),
|
||||
(True, None, "down"),
|
||||
(True, Connection(False), "down"),
|
||||
(True, Connection(True), "ok"),
|
||||
],
|
||||
)
|
||||
def test_nats_status(enabled: bool, connection: Connection | None, expected: str) -> None:
|
||||
health = HealthProvider(nats_enabled=enabled, hotfolder=None)
|
||||
health.nats = connection
|
||||
assert health.nats_status() == expected
|
||||
|
||||
|
||||
async def test_hotfolder_result_is_cached_for_ttl() -> None:
|
||||
hotfolder = CountingHotFolder(result=True)
|
||||
clock = Clock()
|
||||
health = HealthProvider(nats_enabled=False, hotfolder=hotfolder, cache_ttl=5, clock=clock)
|
||||
assert await health.hotfolder_status() == "ok"
|
||||
hotfolder.result = False
|
||||
clock.now += 4.9
|
||||
assert await health.hotfolder_status() == "ok"
|
||||
assert hotfolder.calls == 1
|
||||
clock.now += 0.2
|
||||
assert await health.hotfolder_status() == "down"
|
||||
assert hotfolder.calls == 2
|
||||
|
||||
|
||||
async def test_hanging_check_is_down_within_timeout_and_not_repeated() -> None:
|
||||
hotfolder = CountingHotFolder()
|
||||
hotfolder.release = threading.Event()
|
||||
clock = Clock()
|
||||
health = HealthProvider(nats_enabled=False, hotfolder=hotfolder, check_timeout=0.05, cache_ttl=5, clock=clock)
|
||||
try:
|
||||
assert await health.hotfolder_status() == "down"
|
||||
clock.now += 10 # кэш истёк, но прежняя проверка ещё висит — новый поток не запускается
|
||||
assert await health.hotfolder_status() == "down"
|
||||
assert hotfolder.calls == 1
|
||||
finally:
|
||||
hotfolder.release.set()
|
||||
|
||||
|
||||
async def test_check_raising_is_down() -> None:
|
||||
hotfolder = CountingHotFolder()
|
||||
hotfolder.error = OSError("smb")
|
||||
health = HealthProvider(nats_enabled=False, hotfolder=hotfolder)
|
||||
assert await health.hotfolder_status() == "down"
|
||||
|
||||
|
||||
async def test_snapshot_with_connected_nats(tmp_path: Path) -> None:
|
||||
health = HealthProvider(nats_enabled=True, hotfolder=LocalHotFolder(tmp_path))
|
||||
health.nats = Connection(True)
|
||||
assert await health.snapshot() == {"nats": "ok", "hotfolder": "ok"}
|
||||
@ -64,7 +64,7 @@
|
||||
|
||||
### Task 12: [5.3] `src/core/worker.py`
|
||||
|
||||
- [ ] `src/core/worker.py`: `on_request(msg)`: валидация → LRU `job_id` → ack `processing` → очередь; семафор `SCAN_MAX_INFLIGHT`; `handle(job)`: S3 get → `hotfolder.put` → опрос до дедлайна → `raw/{job_id}.json` → `Normalizer` → результат/конверт → publish; таблица ошибок D-7; `/healthcheck.nats`, `.hotfolder`; проверка: unit-тесты `handle` с фейками и `LocalHotFolder` + `process_once` эмулятора — сценарии spec (ack раньше файла, повторный запрос → один файл, успех → NOT_IMPLEMENTED с `raw_s3_key`, INVALID_DOCUMENT, TIMEOUT с укороченными таймаутами, RECOGNITION_FAILED при отсутствии каталога, SCHEMA_MISMATCH без ack, FILE_READ_ERROR)
|
||||
- [x] `src/core/worker.py`: `on_request(msg)`: валидация → LRU `job_id` → ack `processing` → очередь; семафор `SCAN_MAX_INFLIGHT`; `handle(job)`: S3 get → `hotfolder.put` → опрос до дедлайна → `raw/{job_id}.json` → `Normalizer` → результат/конверт → publish; таблица ошибок D-7; `/healthcheck.nats`, `.hotfolder`; проверка: unit-тесты `handle` с фейками и `LocalHotFolder` + `process_once` эмулятора — сценарии spec (ack раньше файла, повторный запрос → один файл, успех → NOT_IMPLEMENTED с `raw_s3_key`, INVALID_DOCUMENT, TIMEOUT с укороченными таймаутами, RECOGNITION_FAILED при отсутствии каталога, SCHEMA_MISMATCH без ack, FILE_READ_ERROR)
|
||||
|
||||
### Task 13: [5.4] Тест параллелизма
|
||||
|
||||
|
||||
@ -50,7 +50,7 @@
|
||||
- [x] 5.2 `src/core/clients.py` (`get_bytes`, `put_json`, `put_bytes`), `src/core/nats.py` (подключение с/без nkeys,
|
||||
автопереподключение, `subscribe_requests(queue="scan-gateway", cb)`, `publish` с `Nats-Msg-Id` и повторами);
|
||||
проверка: unit с фейками, интеграционный тест подключения к compose-NATS без JetStream
|
||||
- [ ] 5.3 `src/core/worker.py`: `on_request(msg)`: валидация → LRU `job_id` → ack `processing` → очередь; семафор
|
||||
- [x] 5.3 `src/core/worker.py`: `on_request(msg)`: валидация → LRU `job_id` → ack `processing` → очередь; семафор
|
||||
`SCAN_MAX_INFLIGHT`; `handle(job)`: S3 get → `hotfolder.put` → опрос до дедлайна → `raw/{job_id}.json` → `Normalizer` →
|
||||
результат/конверт → publish; таблица ошибок D-7; `/healthcheck.nats`, `.hotfolder`; проверка: unit-тесты `handle` с
|
||||
фейками и `LocalHotFolder` + `process_once` эмулятора — сценарии spec (ack раньше файла, повторный запрос → один файл,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user