gendesign/backend/tests/test_rbac.py

"""RBAC unit + integration tests.

Coverage:
  - get_role / get_user_scope happy + error paths
  - is_path_allowed glob semantics (admin everywhere, pilot blocked from /admin/**,
    analyst sees /** except admin/data-management — #962)
  - GET /me — header missing / unknown user / pilot / admin
  - rbac_guard middleware:
    * любой non-public path требует X-Authenticated-User (no-header → 401)
    * неизвестный юзер → 403 на ВСЁ (включая non-admin paths) —
      «человек без ролей вообще ничего не видит» (decided 2026-05-25)
    * /api/v1/admin/* — только role=admin (pilot/analyst → 403; #962)
    * public /health пропускаем без проверки

Тесты используют изолированный FastAPI app (см. _build_test_app), чтобы не
тянуть тяжёлые модули из app.main (weasyprint требует gobject который не
ставится на macOS dev-машинах из коробки). RBAC middleware и /me router
импортируются напрямую — это даёт нам ровно тот же поведение что в проде.
"""

from __future__ import annotations

import re
from collections.abc import Awaitable, Callable

import pytest
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, Response
from fastapi.testclient import TestClient

from app.api.v1 import me as me_router
from app.core import auth as auth_mod


@pytest.fixture(autouse=True)
def _reset_auth_cache() -> None:
    """Каждый тест получает свежий YAML-кэш."""
    auth_mod.reset_cache_for_tests()


# ---------------------------------------------------------------------------
# Test app — копия rbac_guard из app/main.py, чтобы не подтягивать тяжёлые
# импорты (weasyprint, celery worker, ...). Если поведение middleware меняется
# в проде — синхронизируй здесь.
# ---------------------------------------------------------------------------


_ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})


def _build_test_app() -> FastAPI:
    app = FastAPI()

    @app.middleware("http")
    async def rbac_guard(
        request: Request,
        call_next: Callable[[Request], Awaitable[Response]],
    ) -> Response:
        path = request.url.path
        if path in _PUBLIC_PATHS:
            return await call_next(request)

        username = request.headers.get("X-Authenticated-User")
        if not username:
            return JSONResponse(
                status_code=401,
                content={"detail": "no authenticated user (Caddy basic_auth required)"},
            )
        try:
            role = auth_mod.get_role(username)
        except KeyError:
            return JSONResponse(
                status_code=403,
                content={"detail": "user not in roles config"},
            )
        if _ADMIN_API_RE.match(path) and role != "admin":
            return JSONResponse(
                status_code=403,
                content={"detail": "admin only"},
            )
        return await call_next(request)

    app.include_router(me_router.router, prefix="/api/v1", tags=["me"])

    # Реальный admin endpoint, который должен быть заблокирован middleware'ом.
    @app.get("/api/v1/admin/dummy")
    async def admin_dummy() -> dict:
        return {"ok": True}

    # Non-admin /api/v1/* endpoint — analyst/pilot/admin должны доходить до него.
    @app.get("/api/v1/parcels/dummy")
    async def parcels_dummy() -> dict:
        return {"ok": True}

    @app.get("/health")
    async def health() -> dict:
        return {"status": "ok"}

    return app


@pytest.fixture
def client() -> TestClient:
    return TestClient(_build_test_app())


# ---------------------------------------------------------------------------
# get_role
# ---------------------------------------------------------------------------


def test_get_role_known_users() -> None:
    assert auth_mod.get_role("admin") == "admin"
    assert auth_mod.get_role("kopylov") == "pilot"
    for n in range(1, 11):
        assert auth_mod.get_role(f"user{n}") == "pilot"


def test_get_role_unknown_user_raises() -> None:
    with pytest.raises(KeyError):
        auth_mod.get_role("nobody")


# ---------------------------------------------------------------------------
# is_path_allowed
# ---------------------------------------------------------------------------


def test_is_path_allowed_admin_everywhere() -> None:
    assert auth_mod.is_path_allowed("admin", "/")
    assert auth_mod.is_path_allowed("admin", "/admin")
    assert auth_mod.is_path_allowed("admin", "/admin/scrape/runs")
    assert auth_mod.is_path_allowed("admin", "/api/v1/admin/scrape/status")
    assert auth_mod.is_path_allowed("admin", "/trade-in/api/v1/admin/scrape")
    assert auth_mod.is_path_allowed("admin", "/concept/123")
    assert auth_mod.is_path_allowed("admin", "/analytics/dashboard")


def test_is_path_allowed_pilot_only_tradein() -> None:
    """Pilot имеет доступ ТОЛЬКО к /trade-in/** (decision 2026-05-26).
    Landing, analytics, site-finder, concept, остальной /api/v1/* — закрыты."""
    # Pilot ALLOWED — только trade-in
    assert auth_mod.is_path_allowed("pilot", "/trade-in")
    assert auth_mod.is_path_allowed("pilot", "/trade-in/")
    assert auth_mod.is_path_allowed("pilot", "/trade-in/123")
    assert auth_mod.is_path_allowed("pilot", "/trade-in/api/v1/search")
    assert auth_mod.is_path_allowed("pilot", "/trade-in/api/v1/me")

    # Pilot DENIED — landing + остальные разделы (admin-only)
    assert not auth_mod.is_path_allowed("pilot", "/")
    assert not auth_mod.is_path_allowed("pilot", "/analytics/dashboard")
    assert not auth_mod.is_path_allowed("pilot", "/site-finder/123")
    assert not auth_mod.is_path_allowed("pilot", "/concept/abc")
    assert not auth_mod.is_path_allowed("pilot", "/api/v1/parcels/123")
    assert not auth_mod.is_path_allowed("pilot", "/api/v1/analytics/dashboard")

    # Pilot DENIED — admin paths (explicit deny + not in allowed)
    assert not auth_mod.is_path_allowed("pilot", "/admin")
    assert not auth_mod.is_path_allowed("pilot", "/admin/jobs")
    assert not auth_mod.is_path_allowed("pilot", "/admin/scrape/runs/42")
    assert not auth_mod.is_path_allowed("pilot", "/api/v1/admin/scrape/status")
    assert not auth_mod.is_path_allowed("pilot", "/api/v1/admin/jobs")
    assert not auth_mod.is_path_allowed("pilot", "/trade-in/api/v1/admin/scrape")


def test_is_path_allowed_analyst_everything_except_admin() -> None:
    """Analyst (#962) видит ВСЁ кроме admin/data-management.
    paths=/** ; deny=/admin/**, /api/v1/admin/**, /trade-in/api/v1/admin/**."""
    # Analyst ALLOWED — данные, аналитика, site-finder, concept, trade-in
    assert auth_mod.is_path_allowed("analyst", "/")
    assert auth_mod.is_path_allowed("analyst", "/analytics/dashboard")
    assert auth_mod.is_path_allowed("analyst", "/site-finder/123")
    assert auth_mod.is_path_allowed("analyst", "/concept/abc")
    assert auth_mod.is_path_allowed("analyst", "/trade-in/123")
    assert auth_mod.is_path_allowed("analyst", "/api/v1/parcels/66:41:0204016:10/analyze")
    assert auth_mod.is_path_allowed("analyst", "/api/v1/parcels/66:41:0204016:10/forecast")
    assert auth_mod.is_path_allowed("analyst", "/api/v1/analytics/dashboard")

    # Analyst DENIED — admin/data-management (scraper, sync, user/role mgmt)
    assert not auth_mod.is_path_allowed("analyst", "/admin")
    assert not auth_mod.is_path_allowed("analyst", "/admin/jobs")
    assert not auth_mod.is_path_allowed("analyst", "/admin/scrape/runs/42")
    assert not auth_mod.is_path_allowed("analyst", "/api/v1/admin/jobs/settings")
    assert not auth_mod.is_path_allowed("analyst", "/api/v1/admin/scrape/status")
    assert not auth_mod.is_path_allowed("analyst", "/trade-in/api/v1/admin/scrape")


def test_is_path_allowed_unknown_role_denied() -> None:
    assert not auth_mod.is_path_allowed("ghost", "/")
    assert not auth_mod.is_path_allowed("ghost", "/admin")


# ---------------------------------------------------------------------------
# get_user_scope
# ---------------------------------------------------------------------------


def test_get_user_scope_admin() -> None:
    scope = auth_mod.get_user_scope("admin")
    assert scope["username"] == "admin"
    assert scope["role"] == "admin"
    assert "/**" in scope["allowed_paths"]
    assert scope["deny_paths"] == []


def test_get_user_scope_pilot() -> None:
    scope = auth_mod.get_user_scope("kopylov")
    assert scope["username"] == "kopylov"
    assert scope["role"] == "pilot"
    # Pilot allowed только /trade-in/** (decision 2026-05-26).
    assert "/trade-in/**" in scope["allowed_paths"]
    assert "/trade-in/api/v1/**" in scope["allowed_paths"]
    assert "/" not in scope["allowed_paths"]
    assert "/analytics/**" not in scope["allowed_paths"]
    assert "/admin/**" in scope["deny_paths"]
    assert "/api/v1/admin/**" in scope["deny_paths"]


def test_get_user_scope_analyst() -> None:
    """analysttest (#962) → analyst scope: allowed /** + admin deny_paths."""
    scope = auth_mod.get_user_scope("analysttest")
    assert scope["username"] == "analysttest"
    assert scope["role"] == "analyst"
    assert "/**" in scope["allowed_paths"]
    # deny_paths драйвит фронтовый RouteGuard для /admin/** страниц.
    assert "/admin/**" in scope["deny_paths"]
    assert "/api/v1/admin/**" in scope["deny_paths"]
    assert "/trade-in/api/v1/admin/**" in scope["deny_paths"]


def test_get_user_scope_unknown_raises() -> None:
    with pytest.raises(KeyError):
        auth_mod.get_user_scope("nobody")


# ---------------------------------------------------------------------------
# GET /me
# ---------------------------------------------------------------------------


def test_me_endpoint_admin(client: TestClient) -> None:
    resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "admin"})
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["username"] == "admin"
    assert body["role"] == "admin"
    assert body["allowed_paths"] == ["/**"]


def test_me_endpoint_pilot(client: TestClient) -> None:
    resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "kopylov"})
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["username"] == "kopylov"
    assert body["role"] == "pilot"
    assert "/admin/**" in body["deny_paths"]


def test_me_endpoint_no_header_401(client: TestClient) -> None:
    resp = client.get("/api/v1/me")
    assert resp.status_code == 401
    assert "no authenticated user" in resp.json()["detail"].lower()


def test_me_endpoint_unknown_user_403(client: TestClient) -> None:
    resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "ghost"})
    assert resp.status_code == 403
    assert "not in roles config" in resp.json()["detail"].lower()


# ---------------------------------------------------------------------------
# rbac_guard middleware
# ---------------------------------------------------------------------------


def test_rbac_guard_pilot_blocked_on_admin_api(client: TestClient) -> None:
    """Pilot, ходящий в /api/v1/admin/* — должен получить 403, даже если
    target endpoint существует."""
    resp = client.get(
        "/api/v1/admin/dummy",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 403
    assert resp.json()["detail"] == "admin only"


def test_rbac_guard_admin_allowed(client: TestClient) -> None:
    """Admin доходит до handler'а — 200 от admin_dummy."""
    resp = client.get(
        "/api/v1/admin/dummy",
        headers={"X-Authenticated-User": "admin"},
    )
    assert resp.status_code == 200
    assert resp.json() == {"ok": True}


def test_rbac_guard_analyst_blocked_on_admin_api(client: TestClient) -> None:
    """#962 backend-hard: existing rbac_guard АВТО-блокирует analyst на
    /api/v1/admin/* (role != admin) — без доп. кода, просто потому что роль
    добавлена. Подтверждаем что 403 'admin only'."""
    resp = client.get(
        "/api/v1/admin/dummy",
        headers={"X-Authenticated-User": "analysttest"},
    )
    assert resp.status_code == 403
    assert resp.json()["detail"] == "admin only"


def test_rbac_guard_analyst_allowed_on_non_admin_api(client: TestClient) -> None:
    """Analyst свободно ходит в обычные /api/v1/* (не admin) — 200."""
    resp = client.get(
        "/api/v1/parcels/dummy",
        headers={"X-Authenticated-User": "analysttest"},
    )
    assert resp.status_code == 200
    assert resp.json() == {"ok": True}


def test_rbac_guard_no_header_on_admin_path_returns_401(client: TestClient) -> None:
    """Локальный curl без Caddy → /api/v1/admin/* отдаёт 401."""
    resp = client.get("/api/v1/admin/dummy")
    assert resp.status_code == 401


def test_rbac_guard_unknown_user_on_admin_path_returns_403(client: TestClient) -> None:
    resp = client.get(
        "/api/v1/admin/dummy",
        headers={"X-Authenticated-User": "ghost"},
    )
    assert resp.status_code == 403
    assert "not in roles config" in resp.json()["detail"].lower()


def test_rbac_guard_skips_health(client: TestClient) -> None:
    """/health публичный — middleware не должен потребовать заголовок."""
    resp = client.get("/health")
    assert resp.status_code == 200
    assert resp.json()["status"] == "ok"


def test_rbac_guard_pilot_can_hit_non_admin_api(client: TestClient) -> None:
    """Pilot должен спокойно ходить в обычные /api/v1/* (не admin)."""
    resp = client.get(
        "/api/v1/me",
        headers={"X-Authenticated-User": "user1"},
    )
    assert resp.status_code == 200
    assert resp.json()["role"] == "pilot"


def test_rbac_guard_unknown_user_blocked_on_non_admin_path(client: TestClient) -> None:
    """Неизвестный юзер (не в roles.yaml) → 403 даже на non-admin path.
    «Человек без ролей вообще ничего не видит» (decided 2026-05-25)."""
    resp = client.get(
        "/api/v1/me",
        headers={"X-Authenticated-User": "ghost"},
    )
    assert resp.status_code == 403
    assert "not in roles config" in resp.json()["detail"].lower()


def test_rbac_guard_no_header_on_non_admin_path_returns_401(client: TestClient) -> None:
    """Любой non-public path без X-Authenticated-User → 401 (не только admin)."""
    resp = client.get("/api/v1/me")
    assert resp.status_code == 401
    assert "no authenticated user" in resp.json()["detail"].lower()