gendesign/tradein-mvp/backend/tests/test_role_single_source.py

"""#3316 — роль резолвится из ОДНОГО источника: реестр (БД) первый, roles.yaml — fallback.

Проверяется значение роли, а не факт вызова механизма:
  * имя из roles.yaml, заведённое в реестре сотрудником → роль `employee`
    (эскалации в admin нет: ни IDOR по чужим оценкам, ни безлимитной квоты);
  * сотрудник, которого в roles.yaml НЕТ → роль резолвится, ownership-check
    пропускает его к СВОЕЙ оценке и держит на чужой (раньше был KeyError → 403);
  * legacy-юзер (есть в YAML, в реестре строки нет) → роль ровно как раньше —
    сверяется ВЕСЬ маппинг roles.yaml, а не один аккаунт;
  * реестр недоступен → fallback на YAML (падение БД не выключает legacy-вход).

FastAPI здесь не поднимается: резолвер — чистая функция от (реестр, YAML),
реестр подменяется фейковой сессией.
"""

from __future__ import annotations

import os
from collections.abc import Callable, Iterator
from contextlib import contextmanager
from types import SimpleNamespace
from typing import Any

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import pytest

from app.core import auth
from app.services import identity_store

_MISSING = object()


class _FakeSession:
    """Отдаёт одну строку `SELECT role ...` — или ничего, если роли нет."""

    def __init__(self, role: str | None) -> None:
        self.role = role
        self.sql: str | None = None
        self.params: dict[str, Any] | None = None

    def execute(self, sql: Any, params: dict[str, Any] | None = None) -> Any:
        self.sql = str(sql)
        self.params = params
        row = None if self.role is None else SimpleNamespace(role=self.role)
        return SimpleNamespace(fetchone=lambda: row)


@pytest.fixture
def registry(monkeypatch: pytest.MonkeyPatch) -> Callable[..., _FakeSession | None]:
    """`registry(role)` — что реестр отвечает на запрос роли.

    role=None → строки нет (legacy-юзер); role=_MISSING → реестр падает.
    """

    def install(role: str | None | object) -> _FakeSession | None:
        if role is _MISSING:

            @contextmanager
            def broken_session() -> Iterator[Any]:
                raise RuntimeError("registry down")
                yield  # pragma: no cover — нужен, чтобы функция была генератором

            monkeypatch.setattr(identity_store, "identity_session", broken_session)
            return None

        session = _FakeSession(role)  # type: ignore[arg-type]

        @contextmanager
        def fake_session() -> Iterator[_FakeSession]:
            yield session

        monkeypatch.setattr(identity_store, "identity_session", fake_session)
        return session

    return install


def _yaml_users() -> dict[str, str]:
    return dict(auth._load_roles_config()["users"])


def _yaml_admin() -> str:
    for username, role in _yaml_users().items():
        if role == "admin":
            return username
    pytest.skip("в auth/roles.yaml нет ни одного admin — тест неприменим")


def test_registry_employee_beats_yaml_admin(registry: Callable[..., Any]) -> None:
    """Эскалация закрыта: имя YAML-админа + строка `employee` в реестре = employee."""
    victim_name = _yaml_admin()
    session = registry("employee")

    assert auth.get_role(victim_name) == "employee"
    # username едет bind-параметром, а не склейкой в SQL.
    assert session.params == {"username": victim_name}
    assert victim_name not in (session.sql or "")


def test_employee_absent_from_yaml_resolves_and_owns_estimate(
    registry: Callable[..., Any],
) -> None:
    """Сотрудник вне roles.yaml: роль есть, своя оценка читается, чужая — нет."""
    name = "employee_not_in_yaml_3316"
    assert name not in _yaml_users()
    registry("employee")

    from fastapi import HTTPException

    from app.api.v1.trade_in import _assert_estimate_access

    # Продуктовое поведение проверяется ПЕРВЫМ: до #3316 здесь прилетал 403
    # («user not in roles config») на СОБСТВЕННУЮ оценку сотрудника.
    _assert_estimate_access(name, name)

    assert auth.get_role(name) == "employee"

    with pytest.raises(HTTPException) as exc:
        _assert_estimate_access("someone_else", name)
    assert exc.value.status_code == 404


def test_legacy_yaml_users_keep_their_roles(registry: Callable[..., Any]) -> None:
    """В реестре строки нет → роли ВСЕХ YAML-юзеров ровно те же, что и были."""
    registry(None)
    users = _yaml_users()
    assert users, "roles.yaml без юзеров — сверять нечего"
    assert {username: auth.get_role(username) for username in users} == users


def _yaml_pilot() -> str:
    for username, role in _yaml_users().items():
        if role == "pilot":
            return username
    pytest.skip("в auth/roles.yaml нет ни одного pilot — тест неприменим")


def test_prod_config_pilot_in_yaml_employee_in_registry(registry: Callable[..., Any]) -> None:
    """Прод-конфигурация 12 из 13 аккаунтов: строка в реестре ЕСТЬ и роль там иная.

    Реестр главнее (`employee`), а объём прав от этого не меняется: scope
    DB-роли `employee` обязан совпадать с вчерашним YAML-scope роли `pilot`.
    Списки сверяются целиком — дрейф ЛЮБОГО из двух ловится здесь, а не
    тихой потерей/выдачей раздела в проде.
    """
    from app.services.auth_session import DB_ROLE_PATHS

    registry("employee")
    assert auth.get_role(_yaml_pilot()) == "employee"

    pilot = auth._load_roles_config()["roles"]["pilot"]
    allow, deny = DB_ROLE_PATHS["employee"]
    assert sorted(allow) == sorted(pilot["paths"])
    assert sorted(deny) == sorted(pilot["deny"] or [])


def test_prod_config_pilot_in_yaml_manager_in_registry(registry: Callable[..., Any]) -> None:
    """Конфигурация kopylov: YAML pilot + реестр manager → manager.

    Лишних путей на tradein-периметре это не даёт: manager отличается от
    employee ровно одним префиксом `/api/v1/team/**` (дашборд «Команды»,
    ВНЕ `/trade-in/**`), а deny-списки совпадают.
    """
    from app.services.auth_session import DB_ROLE_PATHS

    registry("manager")
    assert auth.get_role(_yaml_pilot()) == "manager"

    emp_allow, emp_deny = DB_ROLE_PATHS["employee"]
    mgr_allow, mgr_deny = DB_ROLE_PATHS["manager"]
    extra = set(mgr_allow) - set(emp_allow)
    assert extra == {"/api/v1/team/**"}
    assert not any(p.startswith("/trade-in") for p in extra)
    assert sorted(mgr_deny) == sorted(emp_deny)


def test_registry_failure_falls_back_to_yaml(registry: Callable[..., Any]) -> None:
    """Реестр недоступен → legacy-вход продолжает работать по YAML."""
    registry(_MISSING)
    assert auth.get_role(_yaml_admin()) == "admin"


def test_yaml_role_is_yaml_only(registry: Callable[..., Any]) -> None:
    """Предикат гварда create_employee смотрит ИМЕННО в YAML, мимо реестра."""
    registry("employee")
    assert auth.yaml_role(_yaml_admin()) == "admin"
    assert auth.yaml_role("employee_not_in_yaml_3316") is None