cd5409be6e
Фикс-волна minor-находок ревью (T6.1, T6.3, T2.2-аналог); tasks.md отмечен, кроме make check-full. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
336 lines
13 KiB
Python
336 lines
13 KiB
Python
"""Приложение: `/healthcheck` (`uptime`, `version`, `nats`, `hotfolder`), lifespan (SMB-сессия, S3, NATS в фоне,
|
||
воркер). NATS и S3 не нужны: в тестах lifespan `NATS_ENABLED=false` либо методы `NatsGateway` подменены."""
|
||
|
||
import asyncio
|
||
import subprocess
|
||
import sys
|
||
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
|
||
from src.normalizer import OcrNormalizer
|
||
|
||
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", "subjects"}
|
||
assert body["version"] == get_settings().VERSION
|
||
assert body["subjects"] == {"image_request": "files.img_uploaded", "result": "files.tabular_parsed"}
|
||
datetime.fromisoformat(body["uptime"])
|
||
|
||
|
||
async def test_healthcheck_subjects_from_env_without_jwt(local_env: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setenv("OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT", "test.files.img_uploaded")
|
||
monkeypatch.setenv("OPENBAO__NATS__NATS_RESULT_SUBJECT", "test.files.tabular_parsed")
|
||
monkeypatch.setenv("OPENBAO__NATS__NATS_JWT", "secret-jwt-value")
|
||
get_settings.cache_clear()
|
||
app = create_app(get_settings())
|
||
|
||
async with _client(app) as client:
|
||
response = await client.get("/healthcheck")
|
||
|
||
assert response.json()["subjects"] == {
|
||
"image_request": "test.files.img_uploaded",
|
||
"result": "test.files.tabular_parsed",
|
||
}
|
||
assert "secret-jwt-value" not in response.text
|
||
|
||
|
||
async def test_healthcheck_nats_down_hides_jwt(local_env: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""NATS включён, но не подключён (`down`): значения JWT в теле `/healthcheck` нет."""
|
||
monkeypatch.setenv("NATS_ENABLED", "true")
|
||
monkeypatch.setenv("OPENBAO__NATS__NATS_JWT", "secret-jwt-value")
|
||
never = asyncio.Event()
|
||
|
||
async def fake_connect(self: NatsGateway) -> bool:
|
||
await never.wait()
|
||
return True
|
||
|
||
async def fake_close(self: NatsGateway) -> None:
|
||
return None
|
||
|
||
monkeypatch.setattr(NatsGateway, "connect", fake_connect)
|
||
monkeypatch.setattr(NatsGateway, "close", fake_close)
|
||
get_settings.cache_clear()
|
||
app = create_app(get_settings())
|
||
|
||
async with app.router.lifespan_context(app), _client(app) as client:
|
||
response = await client.get("/healthcheck")
|
||
|
||
assert response.json()["nats"] == "down"
|
||
assert "secret-jwt-value" not in response.text
|
||
|
||
|
||
async def test_startup_logs_nats_subjects(
|
||
local_env: Path, monkeypatch: pytest.MonkeyPatch, log_lines: list[str]
|
||
) -> None:
|
||
monkeypatch.setenv("OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT", "test.files.img_uploaded")
|
||
monkeypatch.setenv("OPENBAO__NATS__NATS_RESULT_SUBJECT", "test.files.tabular_parsed")
|
||
get_settings.cache_clear()
|
||
app = create_app(get_settings())
|
||
|
||
async with app.router.lifespan_context(app):
|
||
pass
|
||
|
||
line = next(line for line in log_lines if "Subject'ы NATS" in line)
|
||
assert "test.files.img_uploaded" in line and "test.files.tabular_parsed" in line
|
||
|
||
|
||
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": "внутренняя ошибка сервера"}
|
||
|
||
|
||
def test_create_app_fails_on_broken_rules(tmp_path: Path) -> None:
|
||
(tmp_path / "statement_lines.yaml").write_text("editions: []\n", encoding="utf-8")
|
||
settings = get_settings().model_copy(update={"SCAN_RULES_DIR": str(tmp_path), "SCAN_HOTFOLDER_PATH": str(tmp_path)})
|
||
with pytest.raises(ValueError, match="statement_lines.yaml"):
|
||
create_app(settings)
|
||
|
||
|
||
def test_create_app_uses_ocr_normalizer() -> None:
|
||
app = create_app(get_settings().model_copy(update={"SCAN_HOTFOLDER_PATH": "."}))
|
||
assert isinstance(app.state.normalizer, OcrNormalizer)
|
||
|
||
|
||
# --- Логгер: значения локальных переменных не попадают в лог (#18) ----------------------------------------------------
|
||
|
||
_LOG_EXCEPTION_SCRIPT = """
|
||
import importlib
|
||
import os
|
||
import sys
|
||
|
||
from loguru import logger
|
||
|
||
sys.path.insert(0, os.getcwd())
|
||
importlib.import_module("src.app")
|
||
|
||
|
||
def fail():
|
||
secret = "local-secret-value"
|
||
raise ValueError("boom" if secret else "")
|
||
|
||
|
||
try:
|
||
fail()
|
||
except ValueError:
|
||
logger.exception("ошибка")
|
||
logger.complete()
|
||
"""
|
||
|
||
|
||
def test_app_import_logs_exceptions_without_local_values(tmp_path: Path) -> None:
|
||
"""Импорт приложения настраивает loguru через setup_logger (diagnose=False): трейсбек есть, значений локальных
|
||
переменных в нём нет. Отдельный процесс — тесты перенастраивают обработчики loguru, здесь нужен чистый импорт.
|
||
Скрипт — файлом: без исходника (`python -c`) loguru не аннотирует строки и тест прошёл бы и с diagnose=True."""
|
||
script = tmp_path / "log_exception.py"
|
||
script.write_text(_LOG_EXCEPTION_SCRIPT)
|
||
result = subprocess.run(
|
||
[sys.executable, str(script)],
|
||
cwd=Path(__file__).resolve().parents[2],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=60,
|
||
check=True,
|
||
)
|
||
output = result.stdout + result.stderr
|
||
assert "ValueError: boom" in output
|
||
assert "local-secret-value" not in output
|