arm-core/api/tests/test_app.py

"""Быстрые тесты каркаса приложения: lifespan, /healthcheck, префикс API, клиент S3 (без БД и NATS).

Аутентификация и supervisor NATS — tests/test_nats_core.py (подключение переехало в src/core/nats.py, рулинг R-3).
"""

import asyncio
import importlib
import subprocess
import sys
import threading
from http import HTTPStatus
from pathlib import Path
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 pydantic import SecretStr

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)


def _expected_subjects() -> dict[str, str]:
    nats = app_module.settings.OPENBAO.NATS
    return {
        "tabular_request": nats.NATS_TABULAR_REQUEST_SUBJECT,
        "image_request": nats.NATS_IMAGE_REQUEST_SUBJECT,
        "result": nats.NATS_RESULT_SUBJECT,
    }


@pytest.mark.parametrize("enabled", [False, True], ids=["when_nats_off", "when_nats_down"])
async def test_healthcheck_subjects(client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch, enabled: bool):
    """Review Focus 5: subjects отдаются при любом состоянии NATS, JWT в ответе нет."""
    monkeypatch.setattr(app_module.settings, "NATS_ENABLED", enabled)
    body = (await client.get("/healthcheck")).json()
    assert body["subjects"] == _expected_subjects()
    assert "NATS_JWT" not in str(body)
    assert "jwt" not in str(body).lower()


@pytest.mark.parametrize("enabled", [False, True], ids=["when_nats_off", "when_nats_down"])
async def test_healthcheck_returns_non_default_subjects_without_jwt(
    client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch, enabled: bool
):
    """T3.1: отдаются именно настроенные (не умолчательные) subject'ы; непустой JWT в тело не попадает."""
    jwt = "JWTVALUEMARK"
    nats = app_module.settings.OPENBAO.NATS
    monkeypatch.setattr(app_module.settings, "NATS_ENABLED", enabled)
    monkeypatch.setattr(nats, "NATS_TABULAR_REQUEST_SUBJECT", "env.files.tabular_uploaded")
    monkeypatch.setattr(nats, "NATS_IMAGE_REQUEST_SUBJECT", "env.files.img_uploaded")
    monkeypatch.setattr(nats, "NATS_RESULT_SUBJECT", "env.files.tabular_parsed")
    monkeypatch.setattr(nats, "NATS_JWT", SecretStr(jwt))
    assert nats.NATS_JWT.get_secret_value() == jwt

    response = await client.get("/healthcheck")

    assert response.json()["subjects"] == {
        "tabular_request": "env.files.tabular_uploaded",
        "image_request": "env.files.img_uploaded",
        "result": "env.files.tabular_parsed",
    }
    assert jwt not in response.text


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()

    async def mapper(session_factory: Any, storage: Any, settings: Any, **kwargs: Any) -> None:
        started.append("mapper")
        await asyncio.Event().wait()

    monkeypatch.setattr(recognition_exchange, "run_publisher_loop", publisher)
    monkeypatch.setattr(recognition_exchange, "run_watchdog_loop", watchdog)
    monkeypatch.setattr(recognition_exchange, "run_mapping_loop", mapper)
    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


@pytest.mark.usefixtures("_no_smb")
async def test_lifespan_logs_nats_subjects_at_start(monkeypatch: pytest.MonkeyPatch):
    """Tasks 1.2: при старте в лог пишутся используемые subject'ы NATS."""
    records: list[Any] = []
    handler_id = logger.add(lambda message: records.append(message.record), level="INFO")
    monkeypatch.setattr(app_module, "_connect_s3", _unavailable_s3)
    monkeypatch.setattr(app_module, "_create_nats", lambda: None)
    monkeypatch.setattr(app_module.settings.OPENBAO.NATS, "NATS_RESULT_SUBJECT", "env.files.tabular_parsed")
    try:
        async with app_module.lifespan(app_module.app):
            pass
    finally:
        logger.remove(handler_id)

    messages = [r["message"] for r in records if "Subject'ы NATS" in r["message"]]
    assert len(messages) == 1
    assert "env.files.tabular_parsed" in messages[0]
    assert "files.tabular_uploaded" in messages[0]


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) == ("files.tabular_parsed", "arm-core")
    assert not nats.is_connected  # создание не подключает: подключение — в start()


async def test_result_subscription_uses_inbox_handler_with_dependencies():
    """Подписка `files.tabular_parsed` получает `handle_message` с фабрикой сессий приложения (без `mapper`: маппинг
    результата — отдельный цикл, Срез 2c D-1)."""
    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"}
    assert handler.keywords["session_factory"] is app_module.async_session_maker


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_starts_mapping_loop_with_storage(monkeypatch: pytest.MonkeyPatch, exchange_loops: list[str]):
    """S3 подключено: третья фоновая задача — цикл маппинга результатов (Срез 2c, design D-1) с хранилищем
    приложения; без S3 — только публикатор и вахтёр и предупреждение в лог."""
    storage = SimpleNamespace(client=_FakeS3Client(), bucket="arm-ka")
    received: list[Any] = []

    async def connect_s3() -> SimpleNamespace:
        return storage

    async def mapper(session_factory: Any, mapper_storage: Any, settings: Any, **kwargs: Any) -> None:
        received.append(mapper_storage)
        exchange_loops.append("mapper")
        await asyncio.Event().wait()

    monkeypatch.setattr(recognition_exchange, "run_mapping_loop", mapper)
    monkeypatch.setattr(app_module, "_connect_s3", connect_s3)
    monkeypatch.setattr(app_module, "_create_nats", _RecordingNats)

    async with app_module.lifespan(app_module.app):
        exchange = app_module.app.state.exchange
        assert exchange is not None
        assert sorted(task.get_name() for task in exchange.tasks) == [
            "recognition-deadline-watchdog",
            "recognition-outbox-publisher",
            "recognition-result-mapper",
        ]
        await asyncio.sleep(0)
        assert sorted(exchange_loops) == ["mapper", "publisher", "watchdog"]
        assert received == [storage]
    assert all(task.done() for task in exchange.tasks)

    async def unavailable() -> None:
        return None

    exchange_loops.clear()
    monkeypatch.setattr(app_module, "_connect_s3", unavailable)
    records: list[Any] = []
    handler_id = logger.add(lambda message: records.append(message.record), level="WARNING")
    try:
        async with app_module.lifespan(app_module.app):
            await asyncio.sleep(0)
            assert sorted(exchange_loops) == ["publisher", "watchdog"]
    finally:
        logger.remove(handler_id)
    assert any("маппинг результатов не запущен" in record["message"] for record in records)


@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"


# --- Логгер: значения локальных переменных не попадают в лог (#63) ----------------------------------------------------

_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[1],
        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