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

"""RBAC unit + integration tests for tradein backend.

MIRROR of backend/tests/test_rbac.py — kept in sync manually.

Coverage:
  - get_role / get_user_scope happy + error paths
  - is_path_allowed glob semantics (admin everywhere, pilot blocked from /admin/**)
  - 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 → 403)

Тесты используют изолированный FastAPI app (см. _build_test_app), чтобы не
тянуть тяжёлые модули из app.main (lifespan task + DB session + scheduler).
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:
    auth_mod.reset_cache_for_tests()


# ---------------------------------------------------------------------------
# Test app — копия rbac_guard из app/main.py.
# ---------------------------------------------------------------------------


_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"])

    @app.get("/api/v1/admin/dummy")
    async def admin_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_excludes_admin() -> None:
    assert auth_mod.is_path_allowed("pilot", "/")
    assert auth_mod.is_path_allowed("pilot", "/analytics/dashboard")
    assert auth_mod.is_path_allowed("pilot", "/site-finder/123")
    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", "/concept/abc")
    assert auth_mod.is_path_allowed("pilot", "/api/v1/parcels/123")
    assert auth_mod.is_path_allowed("pilot", "/trade-in/api/v1/search")

    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_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"
    assert "/admin/**" in scope["deny_paths"]
    assert "/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:
    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:
    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_no_header_on_admin_path_returns_401(client: TestClient) -> None:
    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:
    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:
    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:
    """Неизвестный юзер → 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."""
    resp = client.get("/api/v1/me")
    assert resp.status_code == 401
    assert "no authenticated user" in resp.json()["detail"].lower()