73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
from datetime import datetime
|
|
from unittest.mock import MagicMock
|
|
|
|
import httpx
|
|
import smbclient
|
|
|
|
from src.app import create_app
|
|
from src.core.config import get_settings
|
|
|
|
|
|
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):
|
|
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"}
|
|
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")
|
|
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()
|
|
|
|
|
|
async def test_lifespan_registers_smb_session_in_smb_mode(monkeypatch):
|
|
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")
|
|
register = MagicMock()
|
|
reset = MagicMock()
|
|
monkeypatch.setattr(smbclient, "register_session", register)
|
|
monkeypatch.setattr(smbclient, "reset_connection_cache", reset)
|
|
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_called_once_with("smb.example", "user", "secret")
|
|
reset.assert_called_once()
|
|
|
|
|
|
async def test_unhandled_error_has_json_500(monkeypatch):
|
|
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": "внутренняя ошибка сервера"}
|