e8d14edfbb
`*` вместе с credentials спецификация CORS запрещает (Starlette тогда отражает Origin запроса); фронт передаёт токен заголовком Authorization, cookie не нужны. README «Конфигурация» уточнён. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
651 lines
26 KiB
Python
651 lines
26 KiB
Python
"""Быстрые тесты каркаса приложения: lifespan, /healthcheck, префикс API, клиент S3 (без БД и NATS).
|
||
|
||
Аутентификация и supervisor NATS — tests/test_nats_core.py (подключение переехало в src/core/nats.py, рулинг R-3).
|
||
"""
|
||
|
||
import asyncio
|
||
import importlib
|
||
import threading
|
||
from http import HTTPStatus
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
from fastapi import Request
|
||
from fastapi.routing import APIRoute
|
||
from loguru import logger
|
||
|
||
from src.api.router import router
|
||
from src.core.config import Settings
|
||
from src.core.nats import NatsConnection
|
||
from src.services import recognition_exchange
|
||
|
||
# src/__init__.py реэкспортирует объект app под именем модуля, поэтому модуль берём через importlib
|
||
app_module = importlib.import_module("src.app")
|
||
|
||
|
||
@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"
|
||
|
||
|
||
class _FakeNats:
|
||
def __init__(self) -> None:
|
||
self.started = False
|
||
self.stopped = False
|
||
self.is_connected = False
|
||
|
||
async def start(self) -> None:
|
||
self.started = True
|
||
|
||
async def stop(self) -> None:
|
||
self.stopped = True
|
||
|
||
|
||
class _FakeS3Client:
|
||
def __init__(self) -> None:
|
||
self.session_closed = False
|
||
|
||
async def close_session(self) -> None:
|
||
self.session_closed = True
|
||
|
||
|
||
@pytest.fixture
|
||
def _no_smb(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setattr(app_module.smbclient, "register_session", lambda *args, **kwargs: None)
|
||
monkeypatch.setattr(app_module.smbclient, "reset_connection_cache", lambda: None)
|
||
|
||
|
||
class _FakeEngine:
|
||
"""Подмена движка БД приложения: lifespan вызывает `dispose()` последним шагом остановки (финальное ревью M-3)."""
|
||
|
||
def __init__(self, events: list[str] | None = None) -> None:
|
||
self.disposed = 0
|
||
self.events = events
|
||
|
||
async def dispose(self) -> None:
|
||
self.disposed += 1
|
||
if self.events is not None:
|
||
self.events.append("db")
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def fake_engine(monkeypatch: pytest.MonkeyPatch) -> _FakeEngine:
|
||
"""Lifespan в быстрых тестах не трогает пул реального движка БД."""
|
||
engine = _FakeEngine()
|
||
monkeypatch.setattr(app_module, "engine", engine)
|
||
return engine
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def exchange_loops(monkeypatch: pytest.MonkeyPatch) -> list[str]:
|
||
"""Циклы публикатора и вахтёра подменены ожиданием: быстрые тесты lifespan не ходят в БД."""
|
||
started: list[str] = []
|
||
|
||
async def publisher(session_factory: Any, nats: Any, **kwargs: Any) -> None:
|
||
started.append("publisher")
|
||
await asyncio.Event().wait()
|
||
|
||
async def watchdog(session_factory: Any, **kwargs: Any) -> None:
|
||
started.append("watchdog")
|
||
await asyncio.Event().wait()
|
||
|
||
monkeypatch.setattr(recognition_exchange, "run_publisher_loop", publisher)
|
||
monkeypatch.setattr(recognition_exchange, "run_watchdog_loop", watchdog)
|
||
return started
|
||
|
||
|
||
@pytest.mark.usefixtures("_no_smb")
|
||
async def test_lifespan_starts_when_s3_and_nats_unavailable(monkeypatch: pytest.MonkeyPatch):
|
||
async def unavailable() -> None:
|
||
return None
|
||
|
||
monkeypatch.setattr(app_module, "_connect_s3", unavailable)
|
||
monkeypatch.setattr(app_module, "_create_nats", lambda: None)
|
||
reached_yield = False
|
||
|
||
async with app_module.lifespan(app_module.app):
|
||
reached_yield = True
|
||
assert app_module.app.state.storage is None
|
||
assert app_module.app.state.nats is None
|
||
|
||
assert reached_yield
|
||
|
||
|
||
@pytest.mark.usefixtures("_no_smb")
|
||
async def test_lifespan_starts_and_stops_nats_and_closes_s3(monkeypatch: pytest.MonkeyPatch):
|
||
fake_nats = _FakeNats()
|
||
s3_client = _FakeS3Client()
|
||
storage = SimpleNamespace(client=s3_client)
|
||
|
||
async def connect_s3() -> SimpleNamespace:
|
||
return storage
|
||
|
||
monkeypatch.setattr(app_module, "_connect_s3", connect_s3)
|
||
monkeypatch.setattr(app_module, "_create_nats", lambda: fake_nats)
|
||
|
||
async with app_module.lifespan(app_module.app):
|
||
assert app_module.app.state.storage is storage
|
||
assert app_module.app.state.nats is fake_nats
|
||
assert fake_nats.started
|
||
assert not fake_nats.stopped
|
||
assert not s3_client.session_closed
|
||
|
||
assert fake_nats.stopped
|
||
assert s3_client.session_closed
|
||
|
||
|
||
class _NeverConnectingClient:
|
||
"""Клиент nats-py, который никогда не подключается: connect висит (как nats-py без сервера)."""
|
||
|
||
is_connected = False
|
||
is_closed = False
|
||
|
||
async def connect(self, servers: str, **options: Any) -> None:
|
||
await asyncio.Event().wait()
|
||
|
||
async def close(self) -> None:
|
||
self.is_closed = True
|
||
|
||
|
||
@pytest.mark.usefixtures("_no_smb")
|
||
async def test_lifespan_does_not_wait_for_nats_and_healthcheck_is_down(monkeypatch: pytest.MonkeyPatch):
|
||
"""Регрессия: недоступный NATS не задерживает старт; до первого подключения healthcheck показывает down."""
|
||
|
||
async def unavailable() -> None:
|
||
return None
|
||
|
||
nats = NatsConnection(app_module.settings.OPENBAO.NATS, client_factory=_NeverConnectingClient)
|
||
monkeypatch.setattr(app_module, "_connect_s3", unavailable)
|
||
monkeypatch.setattr(app_module, "_create_nats", lambda: nats)
|
||
monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True)
|
||
|
||
async with app_module.lifespan(app_module.app):
|
||
transport = httpx.ASGITransport(app=app_module.app)
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
||
response = await c.get("/healthcheck")
|
||
assert response.json()["nats"] == "down"
|
||
|
||
assert not nats.is_connected
|
||
|
||
|
||
def test_create_nats_disabled_returns_none(monkeypatch: pytest.MonkeyPatch):
|
||
monkeypatch.setattr(app_module.settings, "NATS_ENABLED", False)
|
||
assert app_module._create_nats() is None
|
||
|
||
|
||
def test_create_nats_enabled_returns_connection_for_results(monkeypatch: pytest.MonkeyPatch):
|
||
monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True)
|
||
nats = app_module._create_nats()
|
||
assert isinstance(nats, NatsConnection)
|
||
assert (nats.result_subject, nats.queue) == ("recognition.result", "arm-core")
|
||
assert not nats.is_connected # создание не подключает: подключение — в start()
|
||
|
||
|
||
async def test_result_subscription_uses_inbox_handler_with_dependencies():
|
||
"""Подписка `recognition.result` получает `handle_message` с фабрикой сессий приложения и `NoopMapper`."""
|
||
subscribed: list[Any] = []
|
||
|
||
class _FakeConnection:
|
||
async def subscribe_results(self, handler: Any) -> None:
|
||
subscribed.append(handler)
|
||
|
||
await app_module._subscribe_results(_FakeConnection())
|
||
|
||
(handler,) = subscribed
|
||
assert handler.func is recognition_exchange.handle_message
|
||
assert handler.args == ()
|
||
assert set(handler.keywords) == {"session_factory", "mapper"}
|
||
assert handler.keywords["session_factory"] is app_module.async_session_maker
|
||
assert isinstance(handler.keywords["mapper"], recognition_exchange.NoopMapper)
|
||
|
||
|
||
def test_create_nats_subscribes_results_on_connect(monkeypatch: pytest.MonkeyPatch):
|
||
monkeypatch.setattr(app_module.settings, "NATS_ENABLED", True)
|
||
nats = app_module._create_nats()
|
||
assert nats is not None
|
||
assert nats._on_connected is app_module._subscribe_results
|
||
|
||
|
||
@pytest.mark.parametrize("secure", [True, False])
|
||
async def test_connect_s3_passes_secure_flag(monkeypatch: pytest.MonkeyPatch, secure: bool):
|
||
created: dict[str, Any] = {}
|
||
|
||
class _FakeMinio:
|
||
def __init__(self, endpoint: str, **kwargs: Any) -> None:
|
||
created.update(kwargs, endpoint=endpoint)
|
||
|
||
async def bucket_exists(self, bucket: str) -> bool:
|
||
return True
|
||
|
||
monkeypatch.setattr(app_module, "Minio", _FakeMinio)
|
||
monkeypatch.setattr(app_module.settings.OPENBAO.MINIO, "MINIO_SECURE", secure)
|
||
|
||
storage = await app_module._connect_s3()
|
||
|
||
assert storage is not None
|
||
assert created["secure"] is secure
|
||
|
||
|
||
class _RecordingNats(_FakeNats):
|
||
"""Запоминает, остановлены ли фоновые задачи обмена к моменту `stop()` соединения NATS."""
|
||
|
||
def __init__(self) -> None:
|
||
super().__init__()
|
||
self.exchange_done_at_stop: bool | None = None
|
||
|
||
async def stop(self) -> None:
|
||
exchange = app_module.app.state.exchange
|
||
self.exchange_done_at_stop = all(task.done() for task in exchange.tasks)
|
||
await super().stop()
|
||
|
||
|
||
@pytest.mark.usefixtures("_no_smb")
|
||
async def test_lifespan_starts_exchange_tasks_and_stops_them_before_nats(
|
||
monkeypatch: pytest.MonkeyPatch, exchange_loops: list[str]
|
||
):
|
||
"""NATS_ENABLED=true: публикатор и вахтёр — две фоновые задачи; остановка — до `nats.stop()`."""
|
||
|
||
async def unavailable() -> None:
|
||
return None
|
||
|
||
fake_nats = _RecordingNats()
|
||
monkeypatch.setattr(app_module, "_connect_s3", unavailable)
|
||
monkeypatch.setattr(app_module, "_create_nats", lambda: fake_nats)
|
||
|
||
async with app_module.lifespan(app_module.app):
|
||
exchange = app_module.app.state.exchange
|
||
assert exchange is not None
|
||
assert len(exchange.tasks) == 2 # noqa: PLR2004 - публикатор и вахтёр
|
||
await asyncio.sleep(0)
|
||
assert sorted(exchange_loops) == ["publisher", "watchdog"]
|
||
assert not any(task.done() for task in exchange.tasks)
|
||
|
||
assert fake_nats.exchange_done_at_stop is True
|
||
assert fake_nats.stopped
|
||
|
||
|
||
@pytest.mark.usefixtures("_no_smb")
|
||
async def test_lifespan_without_nats_starts_no_exchange_tasks(
|
||
monkeypatch: pytest.MonkeyPatch, exchange_loops: list[str]
|
||
):
|
||
"""NATS_ENABLED=false (`_create_nats` → None): фоновые задачи обмена не запускаются."""
|
||
|
||
async def unavailable() -> None:
|
||
return None
|
||
|
||
monkeypatch.setattr(app_module, "_connect_s3", unavailable)
|
||
monkeypatch.setattr(app_module.settings, "NATS_ENABLED", False)
|
||
tasks_before = asyncio.all_tasks()
|
||
|
||
async with app_module.lifespan(app_module.app):
|
||
assert app_module.app.state.exchange is None
|
||
await asyncio.sleep(0)
|
||
assert exchange_loops == []
|
||
assert asyncio.all_tasks() - tasks_before == set()
|
||
|
||
|
||
# --- SMB необязателен, остановка по независимым шагам (финальное ревью I-3, T4.5, M-3) -------------------------------
|
||
|
||
|
||
async def _unavailable_s3() -> None:
|
||
return None
|
||
|
||
|
||
def test_env_files_leave_smb_off():
|
||
"""В `.env.test` (и, значит, в тестах) SMB выключен: пустой SMB_HOST."""
|
||
assert app_module.settings.OPENBAO.SMB.SMB_HOST == ""
|
||
|
||
|
||
async def test_lifespan_skips_smb_when_host_empty(monkeypatch: pytest.MonkeyPatch, fake_engine: _FakeEngine):
|
||
def forbidden(*args: Any, **kwargs: Any) -> None:
|
||
raise AssertionError("SMB не настроен — вызова быть не должно")
|
||
|
||
records: list[Any] = []
|
||
handler_id = logger.add(lambda message: records.append(message.record), level="DEBUG")
|
||
monkeypatch.setattr(app_module.settings.OPENBAO.SMB, "SMB_HOST", "")
|
||
monkeypatch.setattr(app_module.smbclient, "register_session", forbidden)
|
||
monkeypatch.setattr(app_module.smbclient, "reset_connection_cache", forbidden)
|
||
monkeypatch.setattr(app_module, "_connect_s3", _unavailable_s3)
|
||
monkeypatch.setattr(app_module, "_create_nats", lambda: None)
|
||
try:
|
||
async with app_module.lifespan(app_module.app):
|
||
pass
|
||
finally:
|
||
logger.remove(handler_id)
|
||
|
||
assert [r for r in records if "SMB не настроен" in r["message"] and r["level"].name == "DEBUG"]
|
||
assert not [r for r in records if r["level"].name == "ERROR"]
|
||
assert fake_engine.disposed == 1
|
||
|
||
|
||
async def test_lifespan_registers_smb_in_worker_thread_when_configured(monkeypatch: pytest.MonkeyPatch):
|
||
"""Задан SMB_HOST: `register_session` и `reset_connection_cache` — вне потока цикла событий (`to_thread`)."""
|
||
loop_thread = threading.get_ident()
|
||
calls: list[tuple[str, Any, bool]] = []
|
||
|
||
def register_session(*args: Any, **kwargs: Any) -> None:
|
||
calls.append(("register", args, threading.get_ident() != loop_thread))
|
||
|
||
def reset_connection_cache() -> None:
|
||
calls.append(("reset", (), threading.get_ident() != loop_thread))
|
||
|
||
smb = app_module.settings.OPENBAO.SMB
|
||
monkeypatch.setattr(smb, "SMB_HOST", "smb.local")
|
||
monkeypatch.setattr(smb, "SMB_LOGIN", "user")
|
||
monkeypatch.setattr(smb, "SMB_PASSWORD", "password")
|
||
monkeypatch.setattr(app_module.smbclient, "register_session", register_session)
|
||
monkeypatch.setattr(app_module.smbclient, "reset_connection_cache", reset_connection_cache)
|
||
monkeypatch.setattr(app_module, "_connect_s3", _unavailable_s3)
|
||
monkeypatch.setattr(app_module, "_create_nats", lambda: None)
|
||
|
||
async with app_module.lifespan(app_module.app):
|
||
assert calls == [("register", ("smb.local", "user", "password"), True)]
|
||
|
||
assert calls[-1] == ("reset", (), True)
|
||
|
||
|
||
class _FailingNats(_FakeNats):
|
||
def __init__(self, events: list[str]) -> None:
|
||
super().__init__()
|
||
self.events = events
|
||
|
||
async def stop(self) -> None:
|
||
self.events.append("nats")
|
||
raise RuntimeError("подставная ошибка остановки NATS")
|
||
|
||
|
||
class _FailingS3Client:
|
||
def __init__(self, events: list[str]) -> None:
|
||
self.events = events
|
||
|
||
async def close_session(self) -> None:
|
||
self.events.append("s3")
|
||
raise RuntimeError("подставная ошибка закрытия S3")
|
||
|
||
|
||
async def test_shutdown_steps_are_independent_and_engine_disposed_last(
|
||
monkeypatch: pytest.MonkeyPatch, exchange_loops: list[str]
|
||
):
|
||
"""Ошибки `nats.stop()` и закрытия S3 не отменяют сброс SMB и `engine.dispose()`; порядок шагов сохраняется."""
|
||
events: list[str] = []
|
||
monkeypatch.setattr(app_module, "engine", _FakeEngine(events))
|
||
monkeypatch.setattr(app_module.settings.OPENBAO.SMB, "SMB_HOST", "smb.local")
|
||
monkeypatch.setattr(app_module.smbclient, "register_session", lambda *args, **kwargs: None)
|
||
monkeypatch.setattr(app_module.smbclient, "reset_connection_cache", lambda: events.append("smb"))
|
||
storage = SimpleNamespace(client=_FailingS3Client(events))
|
||
|
||
async def connect_s3() -> SimpleNamespace:
|
||
return storage
|
||
|
||
monkeypatch.setattr(app_module, "_connect_s3", connect_s3)
|
||
monkeypatch.setattr(app_module, "_create_nats", lambda: _FailingNats(events))
|
||
records: list[Any] = []
|
||
handler_id = logger.add(lambda message: records.append(message.record), level="ERROR")
|
||
try:
|
||
async with app_module.lifespan(app_module.app):
|
||
exchange = app_module.app.state.exchange
|
||
assert exchange is not None
|
||
finally:
|
||
logger.remove(handler_id)
|
||
|
||
assert all(task.done() for task in exchange.tasks)
|
||
assert events == ["nats", "s3", "smb", "db"]
|
||
failed_steps = sorted(r["extra"].get("step") for r in records)
|
||
assert failed_steps == ["nats", "s3"]
|
||
|
||
|
||
# --- /docs и CORS по DEBUG (OQ-66, OQ-65, issue #9 п.2) -----------------------------------------------------------
|
||
|
||
DOC_PATHS = ["/docs", "/redoc", "/openapi.json"]
|
||
ORIGIN = "https://arm-ka.example.ru"
|
||
|
||
|
||
def _app_with(tmp_path: Any, **update: Any) -> Any:
|
||
"""Приложение `create_app` поверх копии настроек процесса; SPA — пустой временный каталог."""
|
||
return app_module.create_app(app_module.settings.model_copy(update=update), web_dir=str(tmp_path))
|
||
|
||
|
||
async def _request(application: Any, method: str, path: str, **kwargs: Any) -> httpx.Response:
|
||
transport = httpx.ASGITransport(app=application)
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
||
return await c.request(method, path, **kwargs)
|
||
|
||
|
||
@pytest.mark.parametrize("path", DOC_PATHS)
|
||
async def test_docs_disabled_without_debug(tmp_path: Any, path: str) -> None:
|
||
(tmp_path / "index.html").write_text("<html>SPA</html>", encoding="utf-8")
|
||
application = _app_with(tmp_path, DEBUG=False)
|
||
|
||
response = await _request(application, "GET", path)
|
||
|
||
assert response.status_code == HTTPStatus.NOT_FOUND # не схема и не index.html SPA
|
||
assert "SPA" not in response.text
|
||
|
||
|
||
@pytest.mark.parametrize("path", DOC_PATHS)
|
||
async def test_docs_enabled_with_debug(tmp_path: Any, path: str) -> None:
|
||
response = await _request(_app_with(tmp_path, DEBUG=True), "GET", path)
|
||
|
||
assert response.status_code == HTTPStatus.OK
|
||
|
||
|
||
def _preflight_headers() -> dict[str, str]:
|
||
return {"Origin": ORIGIN, "Access-Control-Request-Method": "GET"}
|
||
|
||
|
||
async def test_cors_debug_allows_any_origin(tmp_path: Any) -> None:
|
||
response = await _request(_app_with(tmp_path, DEBUG=True), "OPTIONS", "/healthcheck", headers=_preflight_headers())
|
||
|
||
assert response.status_code == HTTPStatus.OK
|
||
assert response.headers["access-control-allow-origin"] == "*"
|
||
assert "access-control-allow-credentials" not in response.headers # `*` без credentials (ревью R-11)
|
||
|
||
|
||
async def test_cors_prod_without_origins_denies_cross_origin(tmp_path: Any) -> None:
|
||
application = _app_with(tmp_path, DEBUG=False, CORS_ORIGINS=[])
|
||
|
||
preflight = await _request(application, "OPTIONS", "/healthcheck", headers=_preflight_headers())
|
||
simple = await _request(application, "GET", "/healthcheck", headers={"Origin": ORIGIN})
|
||
|
||
assert "access-control-allow-origin" not in preflight.headers
|
||
assert simple.status_code == HTTPStatus.OK
|
||
assert "access-control-allow-origin" not in simple.headers
|
||
|
||
|
||
async def test_cors_prod_allows_listed_origin_with_credentials(tmp_path: Any) -> None:
|
||
application = _app_with(tmp_path, DEBUG=False, CORS_ORIGINS=[ORIGIN])
|
||
|
||
allowed = await _request(application, "OPTIONS", "/healthcheck", headers=_preflight_headers())
|
||
other = await _request(
|
||
application,
|
||
"OPTIONS",
|
||
"/healthcheck",
|
||
headers={"Origin": "https://evil.example.com", "Access-Control-Request-Method": "GET"},
|
||
)
|
||
|
||
assert allowed.status_code == HTTPStatus.OK
|
||
assert allowed.headers["access-control-allow-origin"] == ORIGIN
|
||
assert allowed.headers["access-control-allow-credentials"] == "true"
|
||
assert "access-control-allow-origin" not in other.headers
|
||
|
||
|
||
async def test_cors_prod_wildcard_without_credentials(tmp_path: Any) -> None:
|
||
application = _app_with(tmp_path, DEBUG=False, CORS_ORIGINS=["*"])
|
||
|
||
response = await _request(application, "OPTIONS", "/healthcheck", headers=_preflight_headers())
|
||
|
||
assert response.headers["access-control-allow-origin"] == "*"
|
||
assert "access-control-allow-credentials" not in response.headers
|
||
|
||
|
||
def test_cors_options_rules() -> None:
|
||
base = app_module.settings
|
||
assert app_module.cors_options(base.model_copy(update={"DEBUG": True, "CORS_ORIGINS": []})) == {
|
||
"allow_origins": ["*"],
|
||
"allow_credentials": False,
|
||
}
|
||
assert app_module.cors_options(base.model_copy(update={"DEBUG": False, "CORS_ORIGINS": []})) is None
|
||
assert app_module.cors_options(base.model_copy(update={"DEBUG": False, "CORS_ORIGINS": [ORIGIN]})) == {
|
||
"allow_origins": [ORIGIN],
|
||
"allow_credentials": True,
|
||
}
|
||
assert app_module.cors_options(base.model_copy(update={"DEBUG": False, "CORS_ORIGINS": ["*", ORIGIN]})) == {
|
||
"allow_origins": ["*", ORIGIN],
|
||
"allow_credentials": False,
|
||
}
|
||
|
||
|
||
# --- GET / без редиректа на //, 405 и 404 под /api/v1 (OQ-61, OQ-62, issue #9 п.3) ---------------------------------
|
||
|
||
SPA_HTML = "<html>АРМ КА SPA</html>"
|
||
|
||
|
||
def _spa_app(tmp_path: Any, **update: Any) -> Any:
|
||
(tmp_path / "index.html").write_text(SPA_HTML, encoding="utf-8")
|
||
return _app_with(tmp_path, **update)
|
||
|
||
|
||
@pytest.mark.parametrize("root_path", ["/", ""])
|
||
async def test_root_serves_spa_index_without_redirect(tmp_path: Any, root_path: str) -> None:
|
||
response = await _request(_spa_app(tmp_path, ROOT_PATH=root_path), "GET", "/")
|
||
|
||
assert response.status_code == HTTPStatus.OK
|
||
assert response.text == SPA_HTML
|
||
assert "location" not in response.headers
|
||
|
||
|
||
async def test_root_path_prefix_serves_spa(tmp_path: Any) -> None:
|
||
"""ROOT_PATH за прокси (`/armka`, с хвостовым `/` или без): корень приложения — index.html."""
|
||
for root_path in ("/armka", "/armka/"):
|
||
response = await _request(_spa_app(tmp_path, ROOT_PATH=root_path), "GET", "/armka/")
|
||
assert response.status_code == HTTPStatus.OK, root_path
|
||
assert response.text == SPA_HTML
|
||
|
||
|
||
async def test_spa_client_route_serves_index(tmp_path: Any) -> None:
|
||
response = await _request(_spa_app(tmp_path), "GET", "/admin")
|
||
|
||
assert response.status_code == HTTPStatus.OK
|
||
assert response.text == SPA_HTML
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("method", "path", "allowed"),
|
||
[
|
||
("POST", "/api/v1/me", {"GET"}),
|
||
("DELETE", "/api/v1/me", {"GET"}),
|
||
("PATCH", "/api/v1/directories/company-groups", {"GET", "POST"}),
|
||
("GET", "/api/v1/directories/import", {"POST"}),
|
||
("PUT", "/api/v1/directories/company-groups/00000000-0000-0000-0000-000000000001", {"GET", "PATCH"}),
|
||
("GET", "/api/v1/dev/token", {"POST"}), # dev-эмитент включён в .env.test
|
||
],
|
||
)
|
||
async def test_wrong_method_on_existing_api_route_is_405(
|
||
tmp_path: Any, method: str, path: str, allowed: set[str]
|
||
) -> None:
|
||
response = await _request(_spa_app(tmp_path), method, path)
|
||
|
||
assert response.status_code == HTTPStatus.METHOD_NOT_ALLOWED
|
||
assert {m.strip() for m in response.headers["allow"].split(",")} == allowed
|
||
assert response.json() == {"detail": "метод не поддерживается этим маршрутом"}
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("method", "path"),
|
||
[
|
||
("GET", "/api/v1/no-such-route"),
|
||
("POST", "/api/v1/no-such-route"),
|
||
("GET", "/api/v1"),
|
||
("GET", "/api/v1/"),
|
||
("GET", "/api/v1/me/extra"),
|
||
],
|
||
)
|
||
async def test_unknown_api_path_is_404(tmp_path: Any, method: str, path: str) -> None:
|
||
response = await _request(_spa_app(tmp_path), method, path)
|
||
|
||
assert response.status_code == HTTPStatus.NOT_FOUND
|
||
assert SPA_HTML not in response.text
|
||
|
||
|
||
async def test_dev_route_absent_without_debug_is_404_not_405(tmp_path: Any) -> None:
|
||
"""Поведение Среза 0: при выключенных флагах dev-маршрута нет в таблице — 404 на любой метод."""
|
||
application = _spa_app(tmp_path, DEBUG=False)
|
||
for method in ("GET", "POST"):
|
||
response = await _request(application, method, "/api/v1/dev/token")
|
||
assert response.status_code == HTTPStatus.NOT_FOUND, method
|
||
|
||
|
||
# --- доверенные прокси FORWARDED_ALLOW_IPS (ревью R-2, OQ-65) -------------------------------------------------------
|
||
|
||
|
||
def _with_ip_route(application: Any) -> Any:
|
||
"""Служебный маршрут `/whoip` перед монтированием SPA: адрес клиента глазами приложения."""
|
||
|
||
async def whoip(request: Request) -> dict[str, str | None]:
|
||
return {"ip": request.client.host if request.client else None}
|
||
|
||
application.router.routes.insert(0, APIRoute("/whoip", whoip, methods=["GET"]))
|
||
return application
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("trusted", "expected"),
|
||
[("127.0.0.1", "203.0.113.7"), ("10.0.0.1,10.0.0.2", "127.0.0.1"), ("*", "203.0.113.7")],
|
||
ids=["default-loopback-trusted", "client-not-trusted", "all-trusted"],
|
||
)
|
||
async def test_forwarded_for_only_from_trusted_proxy(tmp_path: Any, trusted: str, expected: str) -> None:
|
||
"""X-Forwarded-For принимается только от адресов FORWARDED_ALLOW_IPS (клиент ASGITransport — 127.0.0.1)."""
|
||
application = _with_ip_route(_app_with(tmp_path, FORWARDED_ALLOW_IPS=trusted))
|
||
|
||
response = await _request(application, "GET", "/whoip", headers={"X-Forwarded-For": "203.0.113.7"})
|
||
|
||
assert response.json() == {"ip": expected}
|
||
|
||
|
||
def test_forwarded_allow_ips_default_is_loopback() -> None:
|
||
assert Settings.model_fields["FORWARDED_ALLOW_IPS"].default == "127.0.0.1"
|