96 lines
3.3 KiB
Python
96 lines
3.3 KiB
Python
"""Быстрые тесты каркаса приложения: аутентификация NATS, /healthcheck, префикс API (без БД и NATS)."""
|
|
|
|
import importlib
|
|
import os
|
|
from http import HTTPStatus
|
|
from types import SimpleNamespace
|
|
|
|
import httpx
|
|
import nkeys
|
|
import pytest
|
|
|
|
from src.api.router import router
|
|
from src.core.config import NATSSettings
|
|
|
|
# src/__init__.py реэкспортирует объект app под именем модуля, поэтому модуль берём через importlib
|
|
app_module = importlib.import_module("src.app")
|
|
|
|
|
|
def _nats_settings(seed: str = "", jwt: str = "") -> NATSSettings:
|
|
return NATSSettings(NATS_URL="nats://localhost:4222", NATS_SEED=seed, NATS_JWT=jwt)
|
|
|
|
|
|
def _valid_seed() -> str:
|
|
return nkeys.encode_seed(os.urandom(32), nkeys.PREFIX_BYTE_USER).decode("utf-8")
|
|
|
|
|
|
def test_build_nats_auth_empty_credentials_gives_no_kwargs():
|
|
assert app_module._build_nats_auth(_nats_settings()) == {}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("seed", "jwt"),
|
|
[("SU-only-seed", ""), ("", "only-jwt")],
|
|
)
|
|
def test_build_nats_auth_partial_credentials_is_config_error(seed: str, jwt: str):
|
|
with pytest.raises(RuntimeError, match="NATS_SEED"):
|
|
app_module._build_nats_auth(_nats_settings(seed, jwt))
|
|
|
|
|
|
def test_build_nats_auth_valid_credentials_give_callbacks():
|
|
seed = _valid_seed()
|
|
auth = app_module._build_nats_auth(_nats_settings(seed, "header.payload.signature"))
|
|
|
|
assert set(auth) == {"user_jwt_cb", "signature_cb"}
|
|
assert auth["user_jwt_cb"]() == b"header.payload.signature"
|
|
signature = auth["signature_cb"](b"nonce")
|
|
assert isinstance(signature, bytes)
|
|
assert signature
|
|
|
|
|
|
@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"
|