arm-core/api/tests/test_auth_deps.py

"""Зависимости аутентификации и прав (spec authn-authz, задача 17): `current_user`, `require(code)`, 401/403.

БД-тесты (`integration`) работают на `db_session` с seed dev-данных; мини-приложение собрано в тесте без lifespan,
верификатор токенов подменён на `DevHs256Verifier`. Быстрые тесты (без БД) проверяют разбор заголовка и то, что
401-ветки не обращаются к сессии.
"""

import ast
import uuid
from collections.abc import AsyncIterator, Callable, Iterator
from datetime import timedelta
from pathlib import Path
from typing import Annotated, Any

import httpx
import pytest
import yaml
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import event, select, update
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession

from scripts.seed import run_seed
from src.api.dependencies.auth import (
    Principal,
    current_user,
    get_token_verifier,
    parse_bearer,
    require,
)
from src.core.auth import DevHs256Verifier
from src.core.exception_handlers import register_exception_handlers
from src.db.db import get_async_session
from src.models.access import AppUser, Role, UserRole
from src.models.directories import RegionalOffice
from src.models.enums import BusinessSegment
from src.repositories.users import AmbiguousUserError, load_principal_data
from src.schemas.directories import RegionalOfficeBrief
from src.schemas.me import MeResponse

SECRET = "test-secret-for-auth-deps-0123456789abcdef"
OTHER_SECRET = "foreign-secret-for-auth-deps-0123456789ab"
SRC = Path(__file__).resolve().parent.parent / "src"
SEED_CORE = Path(__file__).resolve().parent.parent / "seed" / "core"

NOT_REGISTERED = "пользователь не зарегистрирован в АРМ"
DISABLED = "учётная запись отключена"
AMBIGUOUS = "неоднозначная учётная запись"


class _ForbiddenSession:
    """Заглушка сессии для 401-веток: любое обращение — ошибка теста."""

    def __getattr__(self, name: str) -> Any:
        raise AssertionError(f"обращение к БД при 401: session.{name}")


def _build_app(session_factory: Callable[[], AsyncIterator[Any]], verifier: DevHs256Verifier | None) -> FastAPI:
    app = FastAPI()
    register_exception_handlers(app)

    @app.get("/whoami")
    async def whoami(principal: Annotated[Principal, Depends(current_user)]) -> dict[str, Any]:
        return {
            "email": principal.user.email,
            "roles": list(principal.roles),
            "permissions": sorted(principal.permissions),
            "office": principal.regional_office.name,
        }

    @app.get("/kfo-verify", dependencies=[Depends(require("KFO_VERIFY"))])
    async def kfo_verify() -> dict[str, bool]:
        return {"ok": True}

    @app.get("/doc-edit", dependencies=[Depends(require("DOCUMENT_EDIT"))])
    async def doc_edit() -> dict[str, bool]:
        return {"ok": True}

    @app.get("/both", dependencies=[Depends(require("KFO_VERIFY")), Depends(require("DOCUMENT_EDIT"))])
    async def both(principal: Annotated[Principal, Depends(current_user)]) -> dict[str, bool]:
        return {"ok": True}

    app.dependency_overrides[get_async_session] = session_factory
    app.dependency_overrides[get_token_verifier] = lambda: verifier
    return app


def _client(app: FastAPI) -> httpx.AsyncClient:
    return httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test")


def _bearer(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


# --- быстрые тесты без БД ---------------------------------------------------------------------------------------


def test_parse_bearer_ok() -> None:
    assert parse_bearer("Bearer abc.def.ghi") == "abc.def.ghi"
    assert parse_bearer("bearer abc") == "abc"


@pytest.mark.parametrize("header", [None, "", "Bearer", "Bearer ", "Bearer   ", "Basic abc", "abc.def", "Bearer a b"])
def test_parse_bearer_rejects(header: str | None) -> None:
    with pytest.raises(HTTPException) as info:
        parse_bearer(header)
    assert info.value.status_code == 401  # noqa: PLR2004
    assert info.value.headers == {"WWW-Authenticate": "Bearer"}
    assert "Authorization: Bearer" in info.value.detail


def test_require_rejects_empty_code() -> None:
    with pytest.raises(ValueError, match="код права"):
        require("")
    with pytest.raises(ValueError, match="код права"):
        require("   ")


@pytest.fixture
def untouchable_app() -> tuple[FastAPI, list[str]]:
    """Приложение с заглушкой сессии; список `touched` пополняется, если зависимость сессии вообще вызвана."""
    touched: list[str] = []

    async def session() -> AsyncIterator[_ForbiddenSession]:
        touched.append("session")
        yield _ForbiddenSession()

    verifier = DevHs256Verifier(SECRET)
    return _build_app(session, verifier), touched


async def _expect_401(app: FastAPI, headers: dict[str, str]) -> httpx.Response:
    async with _client(app) as client:
        response = await client.get("/whoami", headers=headers)
    assert response.status_code == 401  # noqa: PLR2004
    assert response.headers["www-authenticate"] == "Bearer"
    assert isinstance(response.json()["detail"], str)
    return response


async def test_401_without_header_does_not_touch_db(untouchable_app: tuple[FastAPI, list[str]]) -> None:
    app, touched = untouchable_app
    response = await _expect_401(app, {})
    assert "Authorization: Bearer" in response.json()["detail"]
    assert touched == []


async def test_401_not_bearer_does_not_touch_db(untouchable_app: tuple[FastAPI, list[str]]) -> None:
    app, touched = untouchable_app
    await _expect_401(app, {"Authorization": "Basic dXNlcjpwYXNz"})
    assert touched == []


async def test_401_garbage_token_does_not_touch_db(untouchable_app: tuple[FastAPI, list[str]]) -> None:
    app, touched = untouchable_app
    response = await _expect_401(app, _bearer("not-a-jwt"))
    assert response.json() == {"detail": "некорректный токен"}
    assert touched == []


async def test_401_expired_token_does_not_touch_db(untouchable_app: tuple[FastAPI, list[str]]) -> None:
    app, touched = untouchable_app
    token = DevHs256Verifier(SECRET).issue_token("ka_go@armka.local", ttl=timedelta(minutes=-5))
    response = await _expect_401(app, _bearer(token))
    assert response.json() == {"detail": "срок действия токена истёк"}
    assert touched == []


async def test_401_foreign_signature_does_not_touch_db(untouchable_app: tuple[FastAPI, list[str]]) -> None:
    app, touched = untouchable_app
    token = DevHs256Verifier(OTHER_SECRET).issue_token("ka_go@armka.local")
    response = await _expect_401(app, _bearer(token))
    assert response.json() == {"detail": "недействительная подпись токена"}
    assert touched == []


async def test_401_when_verifier_not_configured() -> None:
    touched: list[str] = []

    async def session() -> AsyncIterator[_ForbiddenSession]:
        touched.append("session")
        yield _ForbiddenSession()

    app = _build_app(session, None)
    token = DevHs256Verifier(SECRET).issue_token("ka_go@armka.local")
    response = await _expect_401(app, _bearer(token))
    assert response.json() == {"detail": "аутентификация не настроена"}
    assert touched == []


def test_get_token_verifier_builds_lazily_and_caches(monkeypatch: pytest.MonkeyPatch) -> None:
    """Проверяющий строится один раз на приложение (в том числе `None`) и кладётся в `app.state.token_verifier`."""
    from types import SimpleNamespace  # noqa: PLC0415

    calls: list[object] = []
    sentinel = DevHs256Verifier(SECRET)

    def fake_build(settings: object) -> DevHs256Verifier:
        calls.append(settings)
        return sentinel

    monkeypatch.setattr("src.api.dependencies.auth.build_verifier", fake_build)
    request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace()))
    assert get_token_verifier(request) is sentinel  # type: ignore[arg-type]
    assert get_token_verifier(request) is sentinel  # type: ignore[arg-type]
    assert len(calls) == 1
    assert request.app.state.token_verifier is sentinel

    monkeypatch.setattr("src.api.dependencies.auth.build_verifier", lambda settings: None)
    request_none = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace()))
    assert get_token_verifier(request_none) is None  # type: ignore[arg-type]
    assert request_none.app.state.token_verifier is None


# --- статическая проверка: код роли в логике не используется ------------------------------------------------------


def _non_docstring_string_constants(tree: ast.AST) -> set[str]:
    """Строковые литералы модуля, кроме docstring'ов (модуля, классов, функций)."""
    docstrings: set[int] = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.Module | ast.ClassDef | ast.FunctionDef | ast.AsyncFunctionDef):
            first = node.body[0] if node.body else None
            if isinstance(first, ast.Expr) and isinstance(first.value, ast.Constant):
                docstrings.add(id(first.value))
    return {
        node.value
        for node in ast.walk(tree)
        if isinstance(node, ast.Constant) and isinstance(node.value, str) and id(node) not in docstrings
    }


def test_role_codes_are_not_used_in_auth_logic() -> None:
    """В `src/api/**` и `src/services/**` нет строковых констант — кодов ролей (права — данные; T17.4)."""
    roles = yaml.safe_load((SEED_CORE / "roles.yaml").read_text(encoding="utf-8"))
    role_codes = {item["code"] for item in roles}
    assert {"KA_RF", "KA_GO", "EXPERT"} <= role_codes
    files = [
        *sorted((SRC / "api").rglob("*.py")),
        *sorted((SRC / "services").rglob("*.py")),
        SRC / "repositories" / "users.py",
    ]
    assert SRC / "api" / "dependencies" / "auth.py" in files
    assert SRC / "api" / "endpoints" / "me.py" in files
    for path in files:
        literals = _non_docstring_string_constants(ast.parse(path.read_text(encoding="utf-8")))
        assert not literals & role_codes, f"{path.relative_to(SRC)}: код роли в логике: {sorted(literals & role_codes)}"


def test_role_code_scan_detects_literal() -> None:
    """Сканер видит код роли в обычной константе и пропускает docstring."""
    tree = ast.parse('"""KA_RF в docstring модуля."""\n\ndef f():\n    """KA_GO в docstring."""\n    return "EXPERT"\n')
    assert _non_docstring_string_constants(tree) == {"EXPERT"}


# --- интеграционные тесты на seed -------------------------------------------------------------------------------


@pytest.fixture
async def seeded(db_session: AsyncSession) -> AsyncSession:
    await run_seed(db_session, dev=True)
    await db_session.flush()
    return db_session


@pytest.fixture
async def app(seeded: AsyncSession) -> FastAPI:
    async def session() -> AsyncIterator[AsyncSession]:
        yield seeded

    return _build_app(session, DevHs256Verifier(SECRET))


@pytest.fixture
def token_for() -> Callable[[str], dict[str, str]]:
    verifier = DevHs256Verifier(SECRET)

    def make(email: str) -> dict[str, str]:
        return _bearer(verifier.issue_token(email))

    return make


async def _add_user(session: AsyncSession, email: str, *, active: bool = True, roles: tuple[str, ...] = ()) -> AppUser:
    office_id = (await session.execute(select(RegionalOffice.id).limit(1))).scalar_one()
    user = AppUser(email=email, full_name="Тест Тестов", regional_office_id=office_id, is_active=active)
    session.add(user)
    await session.flush()
    for code in roles:
        role_id = (await session.execute(select(Role.id).where(Role.code == code))).scalar_one()
        session.add(UserRole(user_id=user.id, role_id=role_id))
    await session.flush()
    return user


def _seed_permissions(*role_codes: str) -> set[str]:
    matrix = yaml.safe_load((SEED_CORE / "role_permissions.yaml").read_text(encoding="utf-8"))
    return {code for role in role_codes for code in matrix[role]}


@pytest.mark.integration
async def test_permission_present(app: FastAPI, token_for: Callable[[str], dict[str, str]]) -> None:
    """Сценарий «Право есть»: KA_GO, KFO_VERIFY."""
    async with _client(app) as client:
        response = await client.get("/kfo-verify", headers=token_for("ka_go@armka.local"))
    assert response.status_code == 200  # noqa: PLR2004
    assert response.json() == {"ok": True}


@pytest.mark.integration
async def test_permission_absent(app: FastAPI, token_for: Callable[[str], dict[str, str]]) -> None:
    """Сценарий «Права нет»: EXPERT, DOCUMENT_EDIT → 403 с названием права."""
    async with _client(app) as client:
        response = await client.get("/doc-edit", headers=token_for("expert@armka.local"))
    assert response.status_code == 403  # noqa: PLR2004
    assert response.json() == {"detail": "недостаточно прав: требуется DOCUMENT_EDIT"}


@pytest.mark.integration
async def test_whoami_returns_user_office_roles_and_permissions(
    app: FastAPI, token_for: Callable[[str], dict[str, str]]
) -> None:
    async with _client(app) as client:
        response = await client.get("/whoami", headers=token_for("ka_go@armka.local"))
    assert response.status_code == 200  # noqa: PLR2004
    body = response.json()
    assert body["email"] == "ka_go@armka.local"
    assert body["roles"] == ["KA_GO"]
    assert body["permissions"] == sorted(_seed_permissions("KA_GO"))
    assert body["office"]


@pytest.mark.integration
async def test_two_roles_union_without_duplicates(
    seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]]
) -> None:
    """Сценарий «Две роли»: KA_RF + CURATOR_GO → объединение прав без дубликатов."""
    await _add_user(seeded, "two.roles@armka.local", roles=("KA_RF", "CURATOR_GO"))
    ka_rf, curator = _seed_permissions("KA_RF"), _seed_permissions("CURATOR_GO")
    assert ka_rf & curator, "тест осмыслен, только если у ролей есть общие права"
    async with _client(app) as client:
        response = await client.get("/whoami", headers=token_for("two.roles@armka.local"))
    assert response.status_code == 200  # noqa: PLR2004
    body = response.json()
    assert body["roles"] == ["CURATOR_GO", "KA_RF"]
    assert body["permissions"] == sorted(ka_rf | curator)
    assert len(body["permissions"]) == len(set(body["permissions"]))


@pytest.mark.integration
async def test_unknown_user_is_403(app: FastAPI, token_for: Callable[[str], dict[str, str]]) -> None:
    async with _client(app) as client:
        response = await client.get("/whoami", headers=token_for("nobody@armka.local"))
    assert response.status_code == 403  # noqa: PLR2004
    assert response.json() == {"detail": NOT_REGISTERED}


@pytest.mark.integration
async def test_inactive_user_is_403(
    seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]]
) -> None:
    await _add_user(seeded, "off@armka.local", active=False, roles=("KA_GO",))
    async with _client(app) as client:
        response = await client.get("/kfo-verify", headers=token_for("off@armka.local"))
    assert response.status_code == 403  # noqa: PLR2004
    assert response.json() == {"detail": DISABLED}


@pytest.mark.integration
async def test_email_case_variants_are_ambiguous_403(
    seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]]
) -> None:
    """R-17: два пользователя с email, различающимися регистром → 403, а не «первый попавшийся»."""
    await _add_user(seeded, "Ivan@x.local", roles=("KA_GO",))
    await _add_user(seeded, "ivan@x.local", roles=("EXPERT",))
    async with _client(app) as client:
        response = await client.get("/whoami", headers=token_for("IVAN@x.local"))
    assert response.status_code == 403  # noqa: PLR2004
    assert response.json() == {"detail": AMBIGUOUS}


@pytest.mark.integration
async def test_token_email_case_insensitive(app: FastAPI, token_for: Callable[[str], dict[str, str]]) -> None:
    async with _client(app) as client:
        response = await client.get("/kfo-verify", headers=token_for("KA_GO@Armka.Local"))
    assert response.status_code == 200  # noqa: PLR2004


@pytest.mark.integration
async def test_user_email_stored_in_mixed_case_is_found(
    seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]]
) -> None:
    await _add_user(seeded, "Mixed.Case@Armka.Local", roles=("KA_GO",))
    async with _client(app) as client:
        response = await client.get("/kfo-verify", headers=token_for("mixed.case@armka.local"))
    assert response.status_code == 200  # noqa: PLR2004


@pytest.mark.integration
async def test_inactive_role_grants_nothing(
    seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]]
) -> None:
    await _add_user(seeded, "mix@armka.local", roles=("KA_GO", "EXPERT"))
    await seeded.execute(update(Role).where(Role.code == "KA_GO").values(is_active=False))
    async with _client(app) as client:
        verify = await client.get("/kfo-verify", headers=token_for("mix@armka.local"))
        who = await client.get("/whoami", headers=token_for("mix@armka.local"))
    assert verify.status_code == 403  # noqa: PLR2004
    assert who.json()["roles"] == ["EXPERT"]
    assert who.json()["permissions"] == sorted(_seed_permissions("EXPERT"))


@pytest.mark.integration
async def test_user_without_roles_has_no_permissions(
    seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]]
) -> None:
    await _add_user(seeded, "norole@armka.local")
    async with _client(app) as client:
        who = await client.get("/whoami", headers=token_for("norole@armka.local"))
        verify = await client.get("/kfo-verify", headers=token_for("norole@armka.local"))
    assert who.status_code == 200  # noqa: PLR2004
    assert who.json()["roles"] == []
    assert who.json()["permissions"] == []
    assert verify.status_code == 403  # noqa: PLR2004


@pytest.fixture
def sql_log(db_engine: AsyncEngine) -> Iterator[list[str]]:
    """Все SQL-выражения, выполненные через движок тестовой БД с момента запроса фикстуры."""
    statements: list[str] = []

    def record(_conn: Any, _cursor: Any, statement: str, *_rest: Any) -> None:
        statements.append(statement)

    event.listen(db_engine.sync_engine, "before_cursor_execute", record)
    try:
        yield statements
    finally:
        event.remove(db_engine.sync_engine, "before_cursor_execute", record)


def _principal_queries(statements: list[str]) -> list[str]:
    return [s for s in statements if "app_user" in s]


@pytest.mark.integration
async def test_db_queried_once_per_request(
    app: FastAPI, token_for: Callable[[str], dict[str, str]], sql_log: list[str]
) -> None:
    """Два `require` и `current_user` на одном маршруте — один запрос к пользователю и правам."""
    async with _client(app) as client:
        single = await client.get("/whoami", headers=token_for("ka_go@armka.local"))
        assert single.status_code == 200  # noqa: PLR2004
        queries_single = len(_principal_queries(sql_log))
        sql_log.clear()
        combined = await client.get("/both", headers=token_for("ka_go@armka.local"))
    assert combined.status_code == 200  # noqa: PLR2004
    assert queries_single == 1
    assert len(_principal_queries(sql_log)) == 1
    assert len(sql_log) == 1, sql_log


@pytest.mark.integration
async def test_both_requires_checked_on_one_route(app: FastAPI, token_for: Callable[[str], dict[str, str]]) -> None:
    async with _client(app) as client:
        response = await client.get("/both", headers=token_for("expert@armka.local"))
    assert response.status_code == 403  # noqa: PLR2004
    assert response.json() == {"detail": "недостаточно прав: требуется KFO_VERIFY"}


@pytest.mark.integration
async def test_repository_load_principal_data(seeded: AsyncSession) -> None:
    assert await load_principal_data(seeded, "nobody@armka.local") is None
    data = await load_principal_data(seeded, "ka_go@armka.local")
    assert data is not None
    assert data.user.email == "ka_go@armka.local"
    assert data.roles == ("KA_GO",)
    assert data.permissions == frozenset(_seed_permissions("KA_GO"))
    assert data.regional_office.id == data.user.regional_office_id


@pytest.mark.integration
async def test_repository_raises_on_ambiguity(seeded: AsyncSession) -> None:
    await _add_user(seeded, "Dup@armka.local", roles=("EXPERT",))
    await _add_user(seeded, "dup@armka.local", roles=("KA_GO",))
    with pytest.raises(AmbiguousUserError):
        await load_principal_data(seeded, "dup@armka.local")


def test_me_response_schema_serializes_enum_and_null_segment() -> None:
    office = RegionalOfficeBrief(id=uuid.uuid4(), name="ГО", is_head_office=True)
    fields: dict[str, Any] = {
        "id": uuid.uuid4(),
        "email": "a@b.c",
        "full_name": "А",
        "regional_office": office,
        "roles": ["KA_GO"],
        "permissions": ["KFO_VIEW"],
    }
    empty = MeResponse(business_segment=None, **fields)
    assert empty.model_dump(mode="json")["business_segment"] is None
    segment = next(iter(BusinessSegment))
    filled = MeResponse(business_segment=segment, **fields)
    assert filled.model_dump()["business_segment"] == str(segment)
    assert type(filled.model_dump()["business_segment"]) is str