"""Tests for the ``X-View-Role`` request header — the request-scoped
self-downgrade mechanism behind the web UI's role-switcher chip.

Contract:

  * Valid header value (a role the caller holds, OR ``employee``, OR the
    ``leader`` UI synonym) NARROWS the request's effective role set to
    that value. All downstream RBAC helpers see the narrowed list.
  * Invalid header value (a role the user does not hold, an unknown
    string, a malformed payload) is silently ignored. The header is a
    UI hint — it can never escalate, only narrow.
  * The ``/auth/me`` payload always returns the REAL role set so the
    frontend role-switcher menu remains navigable.
  * PAT-authenticated requests ignore the header entirely (PATs have no
    UI role switcher; the only reason to set the header would be to
    confuse audit attribution).
"""

import pytest


@pytest.mark.asyncio
async def test_admin_can_simulate_employee_view(client, admin_user):
    """A global_admin sending ``X-View-Role: employee`` is narrowed to
    employee — the same admin endpoint that previously returned 200
    now returns 403, matching the real-employee experience.
    """
    _, token = admin_user
    no_view = await client.get(
        "/api/v1/users",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert no_view.status_code == 200

    as_employee = await client.get(
        "/api/v1/users",
        headers={
            "Authorization": f"Bearer {token}",
            "X-View-Role": "employee",
        },
    )
    assert as_employee.status_code == 403


@pytest.mark.asyncio
async def test_admin_simulating_employee_cannot_manage_cloud(
    client, admin_user,
):
    """Cloud lifecycle endpoints gate on ``can_manage_cloud``. Narrowed
    to employee, the admin loses write authority — the destroy endpoint
    must 403 just as it would for a real employee.

    We probe the gate via the existing list endpoint here (it is the
    cheapest read with the same RBAC seam) — a separate cmdb-server
    fixture would just exercise the same ``can_manage_cloud`` branch.
    """
    _, token = admin_user

    # /api/v1/cmdb/integrations is admin-only; it's the same gate
    # ``can_manage_cloud`` controls.
    no_view = await client.get(
        "/api/v1/cmdb/integrations",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert no_view.status_code in (200, 404)  # 404 if route absent in test

    as_employee = await client.get(
        "/api/v1/cmdb/integrations",
        headers={
            "Authorization": f"Bearer {token}",
            "X-View-Role": "employee",
        },
    )
    # Either the route 403s because we're narrowed (production path)
    # or it 404s because the test environment doesn't mount cmdb
    # routes. Both confirm "narrowed admin does not retain admin
    # authority" — the failure mode we'd worry about is 200.
    assert as_employee.status_code != 200


@pytest.mark.asyncio
async def test_employee_cannot_escalate_to_admin(client, employee_user):
    """The defense-in-depth: an employee setting ``X-View-Role:
    global_admin`` MUST be ignored. The header is a UI hint, not an
    auth claim — narrowing-only by construction.
    """
    _, token = employee_user
    resp = await client.get(
        "/api/v1/users",
        headers={
            "Authorization": f"Bearer {token}",
            "X-View-Role": "global_admin",
        },
    )
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_unknown_view_role_is_ignored(client, admin_user):
    """Garbage / unknown role values are silently dropped — the admin
    keeps full authority, no 4xx.
    """
    _, token = admin_user
    for bogus in ("rumplestiltskin", "GLOBAL_ADMIN", "admin", "", "  "):
        resp = await client.get(
            "/api/v1/users",
            headers={
                "Authorization": f"Bearer {token}",
                "X-View-Role": bogus,
            },
        )
        assert resp.status_code == 200, f"bogus value {bogus!r} narrowed"


@pytest.mark.asyncio
async def test_oversize_header_is_ignored(client, admin_user):
    """Pathologically long header bodies are dropped before validation
    (defense against CRLF injection / log poisoning vectors).
    """
    _, token = admin_user
    resp = await client.get(
        "/api/v1/users",
        headers={
            "Authorization": f"Bearer {token}",
            "X-View-Role": "x" * 256,
        },
    )
    assert resp.status_code == 200


@pytest.mark.asyncio
async def test_leader_view_narrows_to_employee(client, admin_user):
    """``leader`` is a UI synonym — it narrows to ``employee`` so org-
    tree subordinate expansion (already used by every router that
    handles non-admin callers) takes effect organically.
    """
    _, token = admin_user
    resp = await client.get(
        "/api/v1/users",
        headers={
            "Authorization": f"Bearer {token}",
            "X-View-Role": "leader",
        },
    )
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_me_endpoint_returns_real_roles_even_when_narrowed(
    client, admin_user,
):
    """``/auth/me`` is the source of truth for the role-switcher menu —
    it MUST return the un-narrowed role set, otherwise a global_admin
    who switched to ``X-View-Role: employee`` would lose the ability
    to switch back (the menu would only show ``employee`` as available).
    """
    _, token = admin_user
    resp = await client.get(
        "/api/v1/auth/me",
        headers={
            "Authorization": f"Bearer {token}",
            "X-View-Role": "employee",
        },
    )
    assert resp.status_code == 200
    body = resp.json()["data"]
    assert "global_admin" in body["roles"]


@pytest.mark.asyncio
async def test_zero_grant_person_cannot_self_elevate_to_employee(
    client, db_session,
):
    """Edge case caught in security scan: a person whose ``employee``
    role grant was revoked (e.g. contractor with bespoke grants only)
    must NOT be able to acquire ``employee`` capability by sending
    ``X-View-Role: employee``.

    Narrowing must only ever narrow the user's REAL role set, never
    widen it. We instantiate a person with zero role grants, mint a JWT
    for them, and assert that ``X-View-Role: employee`` does not let
    them through ``can_view_cloud_resources`` (which admits
    ``employee``). The header is silently ignored — the request
    reaches ``can_view_cloud_resources([])`` which returns False.
    """
    import uuid as _uuid

    from app.core.security import create_access_token
    from app.identity.person_models import IdentityPerson

    user_id = _uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email="zero-grant@test.com",
        display_name="Zero Grant Contractor",
        employment_status="active",
        source_system="manual",
        hris_person_id=f"manual-{user_id}",
    )
    db_session.add(user)
    # Deliberately NO AuthUserRole insertion.
    await db_session.commit()

    token = create_access_token(str(user_id), [])

    resp = await client.get(
        "/api/v1/cmdb/servers",
        headers={
            "Authorization": f"Bearer {token}",
            "X-View-Role": "employee",
        },
    )
    # The cmdb router gates on ``can_view_cloud_resources`` which
    # admits ``employee``. Without our narrowing carve-out tightening,
    # this would 200. With the fix, the header is silently ignored
    # and the request fails authz.
    assert resp.status_code == 403, (
        f"Zero-grant person acquired employee capability via header — "
        f"narrowing widened the role set. Status: {resp.status_code}"
    )


@pytest.mark.asyncio
async def test_pat_ignores_view_role_header(client, db_session):
    """PAT-bearing clients (MCP, automation) have NO UI role-switcher chip,
    so an ``X-View-Role`` header on a PAT request is suspicious — the
    backend must NOT narrow. This protects audit attribution (a stolen
    PAT can't disguise itself as a less-privileged role) and prevents
    a confused MCP client from accidentally losing capability mid-task.

    We assert the contract by minting a PAT under an admin identity,
    then calling an admin-only endpoint with ``X-View-Role: employee`` —
    the call must succeed. If narrowing leaked through it would 403.
    """
    import uuid as _uuid

    from sqlalchemy import text

    from app.auth.models import AuthUserRole
    from app.core.security import create_access_token
    from app.identity.person_models import IdentityPerson

    user_id = _uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email="pat-admin@test.com",
        display_name="PAT Admin",
        employment_status="active",
        source_system="manual",
        hris_person_id=f"manual-{user_id}",
    )
    db_session.add(user)
    await db_session.flush()
    role_id = await db_session.scalar(
        text("SELECT id FROM auth_roles WHERE name = 'global_admin'")
    )
    if isinstance(role_id, str):
        role_id = _uuid.UUID(role_id)
    db_session.add(AuthUserRole(user_id=user_id, role_id=role_id))
    await db_session.commit()

    # Mint a PAT for this admin via the standard endpoint.
    jwt = create_access_token(str(user_id), ["global_admin"])
    create = await client.post(
        "/api/v1/auth/me/tokens",
        json={"name": "for-test", "expires_in_days": 30},
        headers={"Authorization": f"Bearer {jwt}"},
    )
    assert create.status_code == 200, create.text
    pat = create.json()["data"]["plaintext"]

    # Hit an admin endpoint with the PAT + a narrowing header. The
    # header must be ignored so the admin endpoint still returns 200.
    resp = await client.get(
        "/api/v1/users",
        headers={
            "Authorization": f"Bearer {pat}",
            "X-View-Role": "employee",
        },
    )
    assert resp.status_code == 200, (
        f"PAT narrowing leaked through — should be ignored. "
        f"Status: {resp.status_code} Body: {resp.text}"
    )


@pytest.mark.asyncio
async def test_cloud_admin_view_keeps_cloud_read_access(
    client, cloud_admin_user,
):
    """Regression for the alex.quan@pingcap.com bug (2026-06-02):

    A cloud_admin switching the header role chip to ``cloud_admin``
    used to lose read access to the very cloud surface they manage —
    ``apply_view_role_narrow`` strips every other role from the
    effective set, and ``can_view_cloud_resources`` historically did
    NOT admit ``cloud_admin``. Result: ``/cmdb/scopes``,
    ``/cmdb/servers``, ``/cmdb/domains``, ``/cmdb/load-balancers``
    all 403'd in cloud-admin view, the global search returned 0
    results for any cloud query, and ``/admin/cloud/scopes`` (the
    "子账号 / 权限" menu) surfaced a "loadFailed" toast.

    The fix admits ``cloud_admin`` to ``can_view_cloud_resources``;
    this test pins the behaviour so a future role-helper edit
    can't silently re-break it.
    """
    _, token = cloud_admin_user

    # Sanity: without view narrowing the request succeeds via the
    # ``employee`` DB role that ships alongside cloud_admin.
    no_view = await client.get(
        "/api/v1/cmdb/scopes",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert no_view.status_code == 200

    # The actual regression — narrowed to ``[cloud_admin]`` only,
    # ``employee`` is dropped, and the request must still succeed.
    narrowed = await client.get(
        "/api/v1/cmdb/scopes",
        headers={
            "Authorization": f"Bearer {token}",
            "X-View-Role": "cloud_admin",
        },
    )
    assert narrowed.status_code == 200, (
        f"cloud_admin view narrowed to [cloud_admin] lost cloud read "
        f"access (status={narrowed.status_code}). "
        f"can_view_cloud_resources must admit CLOUD_ADMIN_ROLE."
    )

    # Same shape for /cmdb/servers and /cmdb/domains — every cloud
    # read gate routes through the same helper.
    for path in ("/api/v1/cmdb/servers", "/api/v1/cmdb/domains"):
        resp = await client.get(
            path,
            headers={
                "Authorization": f"Bearer {token}",
                "X-View-Role": "cloud_admin",
            },
        )
        assert resp.status_code == 200, (
            f"{path} 403 in cloud_admin view — status={resp.status_code}"
        )


@pytest.mark.asyncio
async def test_finance_view_keeps_cloud_scope_read_access(
    client, finance_user,
):
    """Regression (2026-06-03): finance could not look up which cloud
    sub-account / subscription a given owner holds.

    Finance already reads the cloud surface via ``can_read_cloud`` (cost
    attribution) and day-to-day passes ``can_view_cloud_resources``
    through the ``employee`` bit. But clicking the「财务」view chip
    narrows ``role_names`` to ``[finance]`` — dropping ``employee`` — and
    ``can_view_cloud_resources`` historically did NOT admit ``finance``,
    so ``GET /cmdb/scopes`` (list + search) 403'd in finance view. The
    fix admits ``finance`` to ``can_view_cloud_resources``; this pins it.
    """
    _, token = finance_user

    # Without narrowing: succeeds via the employee role shipped alongside.
    no_view = await client.get(
        "/api/v1/cmdb/scopes",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert no_view.status_code == 200

    # Narrowed to [finance] — employee dropped — must still succeed,
    # including the owner-lookup search param.
    narrowed = await client.get(
        "/api/v1/cmdb/scopes",
        params={"search": "prod"},
        headers={
            "Authorization": f"Bearer {token}",
            "X-View-Role": "finance",
        },
    )
    assert narrowed.status_code == 200, (
        f"finance view narrowed to [finance] lost cloud scope read "
        f"access (status={narrowed.status_code}). "
        f"can_view_cloud_resources must admit FINANCE_ROLE."
    )


@pytest.mark.asyncio
async def test_admin_list_shows_own_actual_roles_not_narrowed_view(
    client, admin_user, db_session,
):
    """Regression (2026-06-19): an admin who held an extra role (e.g.
    ``finance``) saw it on the header role-switcher chip but NOT in the
    权限管控 / 授权工作台 admin list — and so could never remove it.

    Root cause: ``get_current_user`` and ``GET /users`` share one session.
    The requester's own ORM row is the SAME identity-mapped instance that
    carries the ``X-View-Role`` ``_role_names_override``, so serializing it
    with ``role_names`` (narrowed) collapsed the row to just the active
    view chip — hiding every other grant and its remove button. The header
    switcher reads ``actual_role_names`` and still offered the role, so the
    two surfaces disagreed. Fix: list/detail serialize ``actual_role_names``.
    """
    from sqlalchemy import text

    from app.auth.models import AuthUserRole

    admin, token = admin_user

    # Grant the admin an extra ``finance`` role — the scenario the user hit.
    finance_role_id = await db_session.scalar(
        text("SELECT id FROM auth_roles WHERE name = 'finance'")
    )
    import uuid as _uuid
    if isinstance(finance_role_id, str):
        finance_role_id = _uuid.UUID(finance_role_id)
    db_session.add(
        AuthUserRole(user_id=admin.id, role_id=finance_role_id)
    )
    await db_session.commit()

    # Admin browses the workbench in their default ``global_admin`` view.
    resp = await client.get(
        "/api/v1/users",
        params={"search": "admin@test.com"},
        headers={
            "Authorization": f"Bearer {token}",
            "X-View-Role": "global_admin",
        },
    )
    assert resp.status_code == 200, resp.text
    rows = resp.json()["data"]
    own = next(r for r in rows if r["email"] == "admin@test.com")
    # The own row must list the REAL grants, not the narrowed [global_admin].
    assert "finance" in own["roles"], (
        f"admin's own row hid the finance grant under X-View-Role narrowing "
        f"— roles={own['roles']}"
    )
    assert "global_admin" in own["roles"]


@pytest.mark.asyncio
async def test_admin_can_simulate_admin_view_noop(client, admin_user):
    """Setting ``X-View-Role`` to the user's actual role is a no-op —
    the admin retains every existing authority. Regression guard
    against the narrowing code accidentally over-restricting.
    """
    _, token = admin_user
    resp = await client.get(
        "/api/v1/users",
        headers={
            "Authorization": f"Bearer {token}",
            "X-View-Role": "global_admin",
        },
    )
    assert resp.status_code == 200