"""Tests for the ``asset_global_admin`` role (全局资产管理员, 2026-06-11).

Semantics under test (see ``rbac.ASSET_GLOBAL_ADMIN_ROLE`` docstring):

* manually granted via the role API (unlike ``asset_admin``, which
  rejects manual grants because it's location-derived);
* holders co-derive ``asset_admin`` (grant + reconciler + revoke);
* every location-region gate bypasses for holders
  (``has_global_asset_scope``) — including orphan assets with no
  resolvable location, and asset intake at locations the holder is not
  listed against;
* ``/auth/me`` materializes ALL location ids into
  ``managed_location_ids`` so the frontend region widgets need no
  special case;
* view-role narrowing to ``asset_admin`` keeps the global scope.
"""
from __future__ import annotations

import uuid
from types import SimpleNamespace

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.models import AuthRole, AuthUserRole
from app.domains.asset.location_admin_sync import (
    person_can_manage_asset,
    reconcile_asset_admin_role,
    sync_asset_admin_role_for_person,
)
from app.domains.asset.models import AssetItem, AssetLocation
from app.identity.person_models import IdentityPerson


async def _make_person(db: AsyncSession, email: str) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=email.split("@")[0],
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db.add(p)
    await db.commit()
    return p


async def _role_id(db: AsyncSession, name: str) -> uuid.UUID:
    row = await db.execute(select(AuthRole).where(AuthRole.name == name))
    role = row.scalar_one_or_none()
    if role is None:
        role = AuthRole(id=uuid.uuid4(), name=name, is_system=True)
        db.add(role)
        await db.commit()
    return role.id


async def _holds(db: AsyncSession, person_id: uuid.UUID, role_name: str) -> bool:
    row = await db.execute(
        select(AuthUserRole.user_id)
        .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
        .where(
            AuthUserRole.user_id == person_id,
            AuthRole.name == role_name,
        )
    )
    return row.scalar_one_or_none() is not None


# ── Co-derivation: asset_global_admin ⟹ asset_admin ────────────────


@pytest.mark.asyncio
async def test_sync_grants_asset_admin_for_global_holder(
    db_session: AsyncSession,
):
    """A person with asset_global_admin and ZERO location rows still
    derives asset_admin; dropping the global grant revokes it."""
    person = await _make_person(db_session, "global-aa@test.com")
    global_role_id = await _role_id(db_session, "asset_global_admin")
    await _role_id(db_session, "asset_admin")

    db_session.add(AuthUserRole(user_id=person.id, role_id=global_role_id))
    await db_session.flush()

    await sync_asset_admin_role_for_person(db_session, person.id)
    await db_session.commit()
    assert await _holds(db_session, person.id, "asset_admin")

    # Drop the global grant → derived asset_admin goes away (no location).
    await db_session.execute(
        AuthUserRole.__table__.delete().where(
            AuthUserRole.user_id == person.id,
            AuthUserRole.role_id == global_role_id,
        )
    )
    await sync_asset_admin_role_for_person(db_session, person.id)
    await db_session.commit()
    assert not await _holds(db_session, person.id, "asset_admin")


@pytest.mark.asyncio
async def test_reconciler_keeps_asset_admin_for_global_holder(
    db_session: AsyncSession,
):
    """The daily reconciler must treat asset_global_admin as a grant
    source — otherwise it would revoke asset_admin from every global
    holder who isn't listed against a location (regression guard for
    the should_set union in ``reconcile_asset_admin_role``)."""
    person = await _make_person(db_session, "global-aa-recon@test.com")
    global_role_id = await _role_id(db_session, "asset_global_admin")
    await _role_id(db_session, "asset_admin")

    db_session.add(AuthUserRole(user_id=person.id, role_id=global_role_id))
    await db_session.commit()

    # Reconcile from a drifted state (no asset_admin row yet).
    counters = await reconcile_asset_admin_role(db_session)
    await db_session.commit()
    assert counters["granted"] >= 1
    assert await _holds(db_session, person.id, "asset_admin")

    # A second run is a no-op for this person (idempotent steady state).
    counters = await reconcile_asset_admin_role(db_session)
    await db_session.commit()
    assert await _holds(db_session, person.id, "asset_admin")


# ── Region gate bypass ──────────────────────────────────────────────


@pytest.mark.asyncio
async def test_person_can_manage_asset_global_bypasses_region(
    db_session: AsyncSession,
):
    person = await _make_person(db_session, "global-aa-gate@test.com")

    location = AssetLocation(
        id=uuid.uuid4(), code=f"GA{uuid.uuid4().hex[:8]}", name="GateCity",
    )
    db_session.add(location)
    await db_session.commit()

    # Asset at a location the person does NOT administer.
    held = AssetItem(
        id=uuid.uuid4(),
        asset_tag=f"GA-{uuid.uuid4().hex[:6]}",
        category="laptop",
        status="idle",
        location="GateCity",
    )
    # Orphan asset — no location / assignee resolvable. Only the global
    # tier may act on it.
    orphan = AssetItem(
        id=uuid.uuid4(),
        asset_tag=f"GA-{uuid.uuid4().hex[:6]}",
        category="laptop",
        status="idle",
    )
    db_session.add_all([held, orphan])
    await db_session.commit()

    roles = ["employee", "asset_admin", "asset_global_admin"]
    assert await person_can_manage_asset(
        db_session, person.id, held, role_names=roles,
    )
    assert await person_can_manage_asset(
        db_session, person.id, orphan, role_names=roles,
    )

    # A plain regional asset_admin with no locations fails both.
    regional = ["employee", "asset_admin"]
    assert not await person_can_manage_asset(
        db_session, person.id, held, role_names=regional,
    )
    assert not await person_can_manage_asset(
        db_session, person.id, orphan, role_names=regional,
    )


# ── View-role narrowing keeps the global scope ─────────────────────


def test_view_role_narrow_to_asset_admin_keeps_global_scope():
    from app.core.rbac import apply_view_role_narrow

    def _user_with(*names: str):
        return SimpleNamespace(
            user_roles=[
                SimpleNamespace(role=SimpleNamespace(name=n)) for n in names
            ],
        )

    # Global holder narrowing to the asset-admin chip keeps both.
    user = _user_with("employee", "asset_admin", "asset_global_admin")
    apply_view_role_narrow(user, "asset_admin")
    assert set(user.__dict__["_role_names_override"]) == {
        "asset_admin", "asset_global_admin",
    }

    # ... and narrowing to asset_global_admin collapses to the same pair.
    user = _user_with("employee", "asset_admin", "asset_global_admin")
    apply_view_role_narrow(user, "asset_global_admin")
    assert set(user.__dict__["_role_names_override"]) == {
        "asset_admin", "asset_global_admin",
    }

    # A plain regional asset_admin narrows to exactly [asset_admin] —
    # no widening.
    user = _user_with("employee", "asset_admin")
    apply_view_role_narrow(user, "asset_admin")
    assert user.__dict__["_role_names_override"] == ["asset_admin"]

    # Narrowing the global holder to employee still fully narrows.
    user = _user_with("employee", "asset_admin", "asset_global_admin")
    apply_view_role_narrow(user, "employee")
    assert user.__dict__["_role_names_override"] == ["employee"]


# ── API level ───────────────────────────────────────────────────────


async def _make_global_asset_admin(db_session: AsyncSession):
    """Person holding employee + asset_global_admin (+ co-derived
    asset_admin), with NO office rows. Returns (person, token)."""
    from app.core.security import create_access_token

    person = await _make_person(db_session, f"ga-{uuid.uuid4().hex[:6]}@test.com")
    for name in ("employee", "asset_global_admin"):
        db_session.add(AuthUserRole(
            user_id=person.id, role_id=await _role_id(db_session, name),
        ))
    await db_session.flush()
    await sync_asset_admin_role_for_person(db_session, person.id)
    await db_session.commit()

    token = create_access_token(
        str(person.id), ["employee", "asset_admin", "asset_global_admin"],
    )
    return person, token


@pytest.mark.asyncio
async def test_grant_api_co_derives_and_revokes_asset_admin(
    client, db_session: AsyncSession, admin_user,
):
    """POST /users/{id}/roles accepts asset_global_admin (manual grant
    path) and immediately co-derives asset_admin; removal by name
    revokes both. Manual asset_admin grants stay rejected."""
    _, admin_token = admin_user
    headers = {"Authorization": f"Bearer {admin_token}"}
    target = await _make_person(db_session, "ga-grant-api@test.com")
    await _role_id(db_session, "asset_global_admin")
    await _role_id(db_session, "asset_admin")

    resp = await client.post(
        f"/api/v1/users/{target.id}/roles",
        headers=headers,
        json={"role_name": "asset_global_admin"},
    )
    assert resp.status_code == 200, resp.text
    assert await _holds(db_session, target.id, "asset_global_admin")
    assert await _holds(db_session, target.id, "asset_admin")

    # asset_admin itself is still not manually grantable.
    other = await _make_person(db_session, "ga-grant-api-2@test.com")
    resp = await client.post(
        f"/api/v1/users/{other.id}/roles",
        headers=headers,
        json={"role_name": "asset_admin"},
    )
    assert resp.status_code == 400
    assert resp.json()["errors"][0]["code"] == "ROLE_AUTO_DERIVED"

    # Removing the global role takes the co-derived asset_admin with it.
    resp = await client.delete(
        f"/api/v1/users/{target.id}/roles/by-name/asset_global_admin",
        headers=headers,
    )
    assert resp.status_code == 200, resp.text
    assert not await _holds(db_session, target.id, "asset_global_admin")
    assert not await _holds(db_session, target.id, "asset_admin")


@pytest.mark.asyncio
async def test_global_asset_admin_can_intake_into_any_location(
    client, db_session: AsyncSession,
):
    """Asset intake (``enforce_intake_region``) bypasses for the global
    tier: a holder with zero location rows creates an asset at a
    location they're not listed against."""
    location = AssetLocation(
        id=uuid.uuid4(),
        code=f"EL{uuid.uuid4().hex[:8]}",
        name="ElsewhereCity",
    )
    db_session.add(location)
    await db_session.commit()

    _, token = await _make_global_asset_admin(db_session)

    resp = await client.post(
        "/api/v1/assets",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "asset_tag": f"GA-INTAKE-{uuid.uuid4().hex[:6]}",
            "category": "laptop",
            "brand": "Test",
            "model": "Test",
            "serial_number": f"SN-{uuid.uuid4().hex[:8]}",
            "location": "ElsewhereCity",
        },
    )
    assert resp.status_code == 200, resp.text
    assert resp.json()["success"] is True


@pytest.mark.asyncio
async def test_auth_me_returns_all_locations_for_global_asset_admin(
    client, db_session: AsyncSession,
):
    """/auth/me materializes every location id into
    ``managed_location_ids`` for the global tier (the frontend's
    no-restriction sentinel ``[]`` is keyed on global_admin and would
    read as "manages nothing")."""
    locations = [
        AssetLocation(
            id=uuid.uuid4(), code=f"ME{uuid.uuid4().hex[:8]}", name=f"MeCity{i}",
        )
        for i in range(2)
    ]
    db_session.add_all(locations)
    await db_session.commit()

    _, token = await _make_global_asset_admin(db_session)

    resp = await client.get(
        "/api/v1/auth/me",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200
    data = resp.json()["data"]
    assert "asset_global_admin" in data["roles"]
    got = set(data["managed_location_ids"])
    assert {str(loc.id) for loc in locations} <= got