"""End-to-end tests for ``POST /cmdb/iam-users/{id}/cleanup``.

The cleanup endpoint has two modes:

* **Disown** — Helix loads the cloud driver and runs the IAM revoke
  chain (login profile, access keys, MFA, attached policies, group
  memberships, then ``DeleteUser``). On driver ``ok=True`` Helix
  clears ``owner_person_id`` and stamps ``revoked_at`` /
  ``revoke_reason``. On ``ok=False`` we surface a 502 with the
  driver's verbatim error_message and leave the row untouched. On
  ``NotImplementedError`` we surface 501.

* **Transfer** — Helix-side rebind only; the IAM user keeps working.

These tests cover:

  - happy path: disown success → row revoked, audit row written, FK cleared
  - failure: driver returns ok=False → 502, row untouched, failed audit
  - 501: driver raises NotImplementedError → 501, row untouched
  - idempotency: calling cleanup twice on a revoked row is a no-op (no
    second driver call)
  - transfer mode: still skips the driver, rebinds owner, no revoked_at
  - RBAC: employee → 403
"""
from __future__ import annotations

import uuid
from typing import Any
from unittest.mock import AsyncMock, patch

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

from app.audit.models import AuditLog
from app.domains.cmdb.models import CmdbCloudScope, CmdbIamUser
from app.identity.person_models import IdentityPerson
from app.integrations.driver_base import ProvisioningResult
from app.integrations.models import Integration


async def _make_integration(
    db: AsyncSession, provider: str = "aws_cloud",
) -> Integration:
    row = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider=provider,
        name=f"{provider}-test",
        slug=f"{provider}-test-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["iam"],
    )
    db.add(row)
    await db.commit()
    return row


async def _make_iam_user(
    db: AsyncSession,
    integration: Integration,
    *,
    user_name: str = "alice",
    owner_person_id: str | None = None,
) -> CmdbIamUser:
    row = CmdbIamUser(
        id=uuid.uuid4(),
        integration_id=str(integration.id),
        user_name=user_name,
        arn=f"arn:aws:iam::123:user/{user_name}",
        user_id_external=f"AIDA{uuid.uuid4().hex.upper()[:16]}",
        attached_policies=[],
        groups=[],
        mfa_enabled=False,
        tags={},
        owner_person_id=owner_person_id,
        raw_data={},
    )
    db.add(row)
    await db.commit()
    return row


def _patch_driver(driver: Any):
    """Patch ``_load_cloud_driver`` for the duration of the test so
    the cleanup route loads a controllable stub instead of the real
    AWS / Aliyun / etc. driver."""
    async def _loader(_session, _integration):
        return driver
    # The route imports ``_load_cloud_driver`` from the module
    # ``app.domains.cmdb.lifecycle_service`` at request time, so we
    # patch the source. (cmdb.router rebinds to the imported symbol
    # via ``from … import _load_cloud_driver`` inside the handler.)
    return patch(
        "app.domains.cmdb.lifecycle_service._load_cloud_driver",
        new=_loader,
    )


def _patch_audit_session_factory(db_session: AsyncSession):
    """Redirect the fresh-session audit helper at the test's session
    so failed-audit rows are visible in the same connection the test
    queries against. Without this the failure-path audit row goes to a
    real DB connection that the test_session can't see."""
    from contextlib import asynccontextmanager

    @asynccontextmanager
    async def _factory():
        yield db_session

    return patch(
        "app.core.database.async_session_factory",
        new=_factory,
    )


class _DriverOK:
    """Stand-in driver whose ``revoke_account`` succeeds with the
    canonical AWS-shaped response (``raw_response.steps`` is the list
    of revoke chain steps the driver actually executed)."""

    def __init__(self) -> None:
        self.calls: list[Any] = []
        self.close = AsyncMock()

    async def revoke_account(self, ctx):
        self.calls.append(ctx)
        return ProvisioningResult(
            ok=True,
            external_ref=ctx.account_id,
            raw_response={
                "steps": [
                    "delete_login_profile",
                    "delete_access_key:AKIA...",
                    "detach_user_policy:AdminAccess",
                    "delete_user",
                ],
            },
        )


class _DriverFails:
    """Driver returns ``ok=False`` — provider-side rejection."""

    def __init__(self, msg: str) -> None:
        self._msg = msg
        self.calls: list[Any] = []
        self.close = AsyncMock()

    async def revoke_account(self, ctx):
        self.calls.append(ctx)
        return ProvisioningResult(
            ok=False, error_message=self._msg,
        )


class _DriverUnsupported:
    """Driver does not implement IAM-level revocation."""

    def __init__(self) -> None:
        self.close = AsyncMock()

    async def revoke_account(self, _ctx):
        raise NotImplementedError(
            "this driver does not support per-user IAM revocation",
        )


@pytest.mark.asyncio
async def test_disown_calls_driver_and_marks_revoked(
    client: AsyncClient,
    db_session: AsyncSession,
    admin_user,
    admin_headers,
):
    integ = await _make_integration(db_session)
    iam = await _make_iam_user(db_session, integ, user_name="alice")
    driver = _DriverOK()

    with _patch_driver(driver):
        resp = await client.post(
            f"/api/v1/cmdb/iam-users/{iam.id}/cleanup",
            headers=admin_headers,
            json={"reason": "alice has left the company"},
        )

    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["success"] is True
    payload = body["data"]
    assert payload["revoked_at"] is not None
    assert payload["revoke_reason"] == "admin_cleanup"
    assert payload["owner_person_id"] is None
    # Enriched response carries provider + integration_name (otherwise
    # the frontend's in-place row replace loses the cloud tag).
    assert payload["provider"] == "aws_cloud"
    assert payload["integration_name"] == integ.name

    # Driver was invoked exactly once, with the IAM user's user_name
    # passed through verbatim — drivers expect either a UserName or
    # email; we hand them what the sync wrote.
    assert len(driver.calls) == 1
    assert driver.calls[0].account_id == "alice"
    assert driver.calls[0].provider == "aws_cloud"
    driver.close.assert_awaited()

    # Refresh from DB and assert the row matches what the response
    # payload claimed (no trust-the-handler ambiguity).
    refreshed = await db_session.get(CmdbIamUser, iam.id)
    await db_session.refresh(refreshed)
    assert refreshed.revoked_at is not None
    assert refreshed.revoke_reason == "admin_cleanup"
    assert refreshed.owner_person_id is None
    assert refreshed.raw_data["_helix"]["cleanup_revoke_steps"] == [
        "delete_login_profile",
        "delete_access_key:AKIA...",
        "detach_user_policy:AdminAccess",
        "delete_user",
    ]
    assert refreshed.raw_data["_helix"]["cleanup_reason"] == (
        "alice has left the company"
    )

    # Audit trail: one ``cmdb.iam_user.cleanup.disown/success`` row.
    audits = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "cmdb.iam_user.cleanup.disown",
            )
        )
    ).scalars().all()
    assert len(audits) == 1
    assert audits[0].status == "success"
    assert audits[0].detail["revoke_steps"] == [
        "delete_login_profile",
        "delete_access_key:AKIA...",
        "detach_user_policy:AdminAccess",
        "delete_user",
    ]


@pytest.mark.asyncio
async def test_disown_provider_error_returns_502_and_leaves_row_untouched(
    client: AsyncClient,
    db_session: AsyncSession,
    admin_user,
    admin_headers,
):
    integ = await _make_integration(db_session)
    iam = await _make_iam_user(db_session, integ, user_name="bob")
    driver = _DriverFails("AWS ClientError AccessDenied: not authorized")

    with _patch_driver(driver), _patch_audit_session_factory(db_session):
        resp = await client.post(
            f"/api/v1/cmdb/iam-users/{iam.id}/cleanup",
            headers=admin_headers,
            json={"reason": "bob is leaving"},
        )

    assert resp.status_code == 502, resp.text
    body = resp.json()
    assert body["success"] is False
    assert "AccessDenied" in body["errors"][0]["message"]
    assert body["errors"][0]["code"] == "IAM_CLEANUP_PROVIDER_ERROR"

    refreshed = await db_session.get(CmdbIamUser, iam.id)
    await db_session.refresh(refreshed)
    assert refreshed.revoked_at is None
    assert refreshed.revoke_reason is None

    # The failed attempt must still be audit-logged via the fresh
    # session helper (otherwise the handler rollback drops the trail).
    audits = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "cmdb.iam_user.cleanup.disown",
                AuditLog.status == "failed",
            )
        )
    ).scalars().all()
    assert len(audits) == 1
    assert "AccessDenied" in audits[0].detail["error_message"]


@pytest.mark.asyncio
async def test_disown_unsupported_returns_501(
    client: AsyncClient,
    db_session: AsyncSession,
    admin_user,
    admin_headers,
):
    integ = await _make_integration(db_session, provider="some_new_cloud")
    iam = await _make_iam_user(db_session, integ, user_name="charlie")
    driver = _DriverUnsupported()

    with _patch_driver(driver), _patch_audit_session_factory(db_session):
        resp = await client.post(
            f"/api/v1/cmdb/iam-users/{iam.id}/cleanup",
            headers=admin_headers,
            json={"reason": "trying to clean up"},
        )

    assert resp.status_code == 501, resp.text
    body = resp.json()
    assert body["errors"][0]["code"] == "IAM_CLEANUP_NOT_SUPPORTED"

    refreshed = await db_session.get(CmdbIamUser, iam.id)
    await db_session.refresh(refreshed)
    assert refreshed.revoked_at is None


@pytest.mark.asyncio
async def test_disown_already_revoked_short_circuits(
    client: AsyncClient,
    db_session: AsyncSession,
    admin_user,
    admin_headers,
):
    """Calling cleanup on a row that is already ``revoked_at != NULL``
    must not invoke the driver again — a duplicate ``DeleteUser`` is a
    no-op cloud-side but burns IAM API budget and clutters audit. The
    endpoint returns 200 with the existing row state."""
    integ = await _make_integration(db_session)
    iam = await _make_iam_user(db_session, integ, user_name="dora")
    driver_one = _DriverOK()

    with _patch_driver(driver_one):
        first = await client.post(
            f"/api/v1/cmdb/iam-users/{iam.id}/cleanup",
            headers=admin_headers,
            json={"reason": "first cleanup"},
        )
    assert first.status_code == 200
    assert len(driver_one.calls) == 1

    driver_two = _DriverOK()
    with _patch_driver(driver_two):
        second = await client.post(
            f"/api/v1/cmdb/iam-users/{iam.id}/cleanup",
            headers=admin_headers,
            json={"reason": "second cleanup attempt"},
        )

    assert second.status_code == 200, second.text
    # Driver must NOT have been called the second time.
    assert driver_two.calls == []


@pytest.mark.asyncio
async def test_transfer_skips_driver_and_rebinds_owner(
    client: AsyncClient,
    db_session: AsyncSession,
    admin_user,
    admin_headers,
):
    integ = await _make_integration(db_session)
    successor = IdentityPerson(
        id=uuid.uuid4(),
        work_email="successor@test.com",
        display_name="Successor Person",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(successor)
    await db_session.commit()

    iam = await _make_iam_user(db_session, integ, user_name="eli")
    driver = _DriverOK()

    with _patch_driver(driver):
        resp = await client.post(
            f"/api/v1/cmdb/iam-users/{iam.id}/cleanup",
            headers=admin_headers,
            json={
                "reason": "transfer to successor",
                "transfer_to_person_id": str(successor.id),
            },
        )

    assert resp.status_code == 200, resp.text
    payload = resp.json()["data"]
    assert payload["owner_person_id"] == str(successor.id)
    assert payload["revoked_at"] is None  # transfer doesn't revoke

    # Driver MUST NOT have been called — transfer is Helix-side only.
    assert driver.calls == []


@pytest.mark.asyncio
async def test_employee_cannot_cleanup(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    employee_headers,
):
    integ = await _make_integration(db_session)
    iam = await _make_iam_user(db_session, integ, user_name="frank")

    # No driver patch — RBAC must short-circuit before we hit the
    # driver code path.
    resp = await client.post(
        f"/api/v1/cmdb/iam-users/{iam.id}/cleanup",
        headers=employee_headers,
        json={"reason": "trying to escalate"},
    )

    assert resp.status_code == 403
    refreshed = await db_session.get(CmdbIamUser, iam.id)
    await db_session.refresh(refreshed)
    assert refreshed.revoked_at is None


@pytest.mark.asyncio
async def test_disown_cascades_revoke_to_discovered_and_helix_grants(
    client: AsyncClient,
    db_session: AsyncSession,
    admin_user,
    admin_headers,
):
    """Regression for the 2026-05-05 ghost-grants bug.

    User report: AWS SG (Singapore) "成员 / 生效 grant 37" persisted
    after every IAM user was disowned + manual ``立即同步`` was
    triggered. Two underlying causes:

      1. ``source='helix'`` grants are NEVER touched by the
         drift-sweep at the end of ``sync_grants_for_integration`` —
         Helix is system-of-record for those, so the sweep skips
         them by design.
      2. The drift sweep's 50%-safety-floor refuses to fire when a
         big batch of cleanups would close >50% of discovered
         grants. Mass IAM cleanup hits this floor and the discovered
         grants ALSO get stranded.

    Fix: ``cleanup_iam_user`` cascades synchronously, in the same
    transaction, to:

      (a) all active discovered grants whose ``external_ref`` starts
          with ``iam-user:{user_name}::`` (the canonical FK back to
          the IAM user written by ``_extract_iam_user_grants``)
      (b) all active helix-issued grants for the same person on the
          scopes the discovered rows live on (the IAM user was the
          auth handle that enacted those helix grants too)

    All cascades stamp ``revoke_reason='iam_user_disowned'`` so
    forensics can distinguish from natural ``user_released`` /
    ``inactive`` / ``drift_removed`` revocations.
    """
    from app.domains.cmdb.models import (
        CmdbCloudAccessGrant,
        CmdbCloudScope,
    )

    integ = await _make_integration(db_session)
    # Active scope tied to the integration. Cascade scopes by
    # ``scope_id IN (scopes hosting this IAM user's discovered
    # grants)``, so a real scope row must exist.
    scope = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integ.id),
        scope_type="standalone_iam",
        external_id="123456789012",
        display_name="AWS SG (test)",
        status="active",
        first_seen_at=None,
    )
    db_session.add(scope)

    # Person who owned the IAM user; both discovered + helix grants
    # are attributed to this person.
    person = IdentityPerson(
        id=uuid.uuid4(),
        work_email="li.da@test.com",
        display_name="李达 / Li Da",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(person)
    await db_session.commit()

    iam = await _make_iam_user(
        db_session, integ,
        user_name="li.da",
        owner_person_id=str(person.id),
    )

    # Discovered grants (3 roles for this IAM user) + one helix-issued
    # grant for the same (scope, person) pair. The 4th grant has a
    # totally unrelated external_ref pattern and tests that helix-
    # source rows are caught by the (scope, person) branch even when
    # they don't carry the ``iam-user:`` external_ref prefix.
    discovered_role_keys = [
        "AdministratorAccess",
        "IAMUserChangePassword",
        "ReadOnlyAccess",
    ]
    for rk in discovered_role_keys:
        db_session.add(CmdbCloudAccessGrant(
            id=uuid.uuid4(),
            scope_id=str(scope.id),
            person_id=str(person.id),
            role_key=rk,
            role_display_name=rk,
            status="active",
            source="discovered",
            external_ref=f"iam-user:{iam.user_name}::{rk}",
            granted_at=None,
        ))
    helix_grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(person.id),
        role_key="custom-helix-issued",
        role_display_name="Custom (Helix-issued)",
        status="active",
        source="helix",
        # helix grants have a different external_ref schema (request id /
        # provisioning ref) — emulate that here so the test exercises
        # the (scope, person) branch, not the prefix branch.
        external_ref=f"helix-grant:{uuid.uuid4()}",
        granted_at=None,
    )
    db_session.add(helix_grant)

    # Unrelated grant on the same scope but a different person —
    # MUST NOT be touched by the cascade.
    other_person = IdentityPerson(
        id=uuid.uuid4(),
        work_email="other@test.com",
        display_name="Other Person",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(other_person)
    await db_session.commit()
    other_grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(other_person.id),
        role_key="ReadOnlyAccess",
        role_display_name="ReadOnlyAccess",
        status="active",
        source="discovered",
        external_ref="iam-user:other::ReadOnlyAccess",
        granted_at=None,
    )
    db_session.add(other_grant)
    await db_session.commit()

    driver = _DriverOK()
    with _patch_driver(driver):
        resp = await client.post(
            f"/api/v1/cmdb/iam-users/{iam.id}/cleanup",
            headers=admin_headers,
            json={"reason": "li.da has left"},
        )
    assert resp.status_code == 200, resp.text

    # 3 discovered + 1 helix = 4 cascaded; ``other_grant`` untouched.
    refreshed_iam = await db_session.get(CmdbIamUser, iam.id)
    await db_session.refresh(refreshed_iam)
    cascaded_ids = (
        refreshed_iam.raw_data["_helix"]["cascade_revoked_grant_ids"]
    )
    assert len(cascaded_ids) == 4
    assert (
        refreshed_iam.raw_data["_helix"]["cascade_revoked_grant_count"]
        == 4
    )

    # Verify each cascaded row landed in DB with the new status +
    # reason.
    rows = list((await db_session.execute(
        select(CmdbCloudAccessGrant).where(
            CmdbCloudAccessGrant.id.in_(
                [uuid.UUID(i) for i in cascaded_ids],
            )
        )
    )).scalars().all())
    assert len(rows) == 4
    for r in rows:
        assert r.status == "revoked"
        assert r.revoke_reason == "iam_user_disowned"
        assert r.revoked_at is not None
        # Inactivity stage cleared so the inactivity sweep can't pick
        # the row back up via leftover state.
        assert r.last_notified_stage is None

    # Other person's grant on the same scope is UNTOUCHED.
    other_after = await db_session.get(
        CmdbCloudAccessGrant, other_grant.id,
    )
    await db_session.refresh(other_after)
    assert other_after.status == "active"
    assert other_after.revoke_reason is None

    # Audit row carries the cascade count + ID list.
    audits = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "cmdb.iam_user.cleanup.disown",
                AuditLog.status == "success",
            )
        )
    ).scalars().all()
    assert len(audits) == 1
    assert audits[0].detail["cascade_revoked_grant_count"] == 4
    assert set(audits[0].detail["cascade_revoked_grant_ids"]) == set(
        cascaded_ids,
    )


@pytest.mark.asyncio
async def test_disown_cascade_does_not_leak_across_integrations(
    client: AsyncClient,
    db_session: AsyncSession,
    admin_user,
    admin_headers,
):
    """A same-named IAM user in a DIFFERENT integration MUST NOT have
    its grants cascaded. AWS prod + AWS staging can both legitimately
    have an IAM user named ``li.da`` — disowning one must not touch
    the other's grants.

    Regression fence for the integration-scoping bug caught in the
    pre-merge review. Without the ``scope.integration_id ==
    iam_row.integration_id`` filter on the discovered query, the
    ``external_ref LIKE 'iam-user:li.da::%'`` prefix matches every
    tenant's grants for any ``li.da``.
    """
    from app.domains.cmdb.models import (
        CmdbCloudAccessGrant,
        CmdbCloudScope,
    )

    # Two separate integrations, both with an IAM user named
    # ``li.da``.
    integ_a = await _make_integration(db_session, provider="aws_cloud")
    integ_b = await _make_integration(db_session, provider="aws_cloud")
    scope_a = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integ_a.id),
        scope_type="standalone_iam",
        external_id="aws-prod",
        display_name="AWS prod",
        status="active",
        first_seen_at=None,
    )
    scope_b = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integ_b.id),
        scope_type="standalone_iam",
        external_id="aws-staging",
        display_name="AWS staging",
        status="active",
        first_seen_at=None,
    )
    db_session.add_all([scope_a, scope_b])

    person = IdentityPerson(
        id=uuid.uuid4(),
        work_email="li.da@test.com",
        display_name="Li Da",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(person)
    await db_session.commit()

    # Same user_name in both tenants. We disown only iam_a; iam_b's
    # grant must survive.
    iam_a = await _make_iam_user(
        db_session, integ_a,
        user_name="li.da",
        owner_person_id=str(person.id),
    )
    await _make_iam_user(
        db_session, integ_b,
        user_name="li.da",
        owner_person_id=str(person.id),
    )

    # One discovered grant per scope. SAME external_ref prefix.
    grant_a = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope_a.id),
        person_id=str(person.id),
        role_key="AdministratorAccess",
        role_display_name="AdministratorAccess",
        status="active",
        source="discovered",
        external_ref="iam-user:li.da::AdministratorAccess",
        granted_at=None,
    )
    grant_b = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope_b.id),
        person_id=str(person.id),
        role_key="AdministratorAccess",
        role_display_name="AdministratorAccess",
        status="active",
        source="discovered",
        external_ref="iam-user:li.da::AdministratorAccess",
        granted_at=None,
    )
    db_session.add_all([grant_a, grant_b])
    await db_session.commit()

    driver = _DriverOK()
    with _patch_driver(driver):
        resp = await client.post(
            f"/api/v1/cmdb/iam-users/{iam_a.id}/cleanup",
            headers=admin_headers,
            json={"reason": "disown li.da on prod only"},
        )
    assert resp.status_code == 200, resp.text

    # iam_a's grant cascaded.
    after_a = await db_session.get(CmdbCloudAccessGrant, grant_a.id)
    await db_session.refresh(after_a)
    assert after_a.status == "revoked"
    assert after_a.revoke_reason == "iam_user_disowned"

    # iam_b's grant — DIFFERENT integration — MUST remain active.
    after_b = await db_session.get(CmdbCloudAccessGrant, grant_b.id)
    await db_session.refresh(after_b)
    assert after_b.status == "active", (
        "Cross-integration cascade leak: disowning iam_a (integ_a) "
        "incorrectly cascaded grant_b (integ_b) which has the same "
        "user_name 'li.da' but lives in a different tenant."
    )
    assert after_b.revoke_reason is None


@pytest.mark.asyncio
async def test_transfer_does_not_cascade_revoke_grants(
    client: AsyncClient,
    db_session: AsyncSession,
    admin_user,
    admin_headers,
):
    """Transfer mode rebinds Helix-side ownership only — the cloud
    IAM user keeps working, so its grants must NOT be cascaded
    revoked. Regression-fence for the disown-only cascade."""
    from app.domains.cmdb.models import (
        CmdbCloudAccessGrant,
        CmdbCloudScope,
    )

    integ = await _make_integration(db_session)
    scope = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integ.id),
        scope_type="standalone_iam",
        external_id="123456789012",
        display_name="AWS SG (test)",
        status="active",
        first_seen_at=None,
    )
    db_session.add(scope)
    person = IdentityPerson(
        id=uuid.uuid4(),
        work_email="staying@test.com",
        display_name="Staying Person",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    successor = IdentityPerson(
        id=uuid.uuid4(),
        work_email="successor2@test.com",
        display_name="Successor",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db_session.add_all([person, successor])
    await db_session.commit()
    iam = await _make_iam_user(
        db_session, integ,
        user_name="staying",
        owner_person_id=str(person.id),
    )
    grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(person.id),
        role_key="AdministratorAccess",
        role_display_name="AdministratorAccess",
        status="active",
        source="discovered",
        external_ref=f"iam-user:{iam.user_name}::AdministratorAccess",
        granted_at=None,
    )
    db_session.add(grant)
    await db_session.commit()

    driver = _DriverOK()
    with _patch_driver(driver):
        resp = await client.post(
            f"/api/v1/cmdb/iam-users/{iam.id}/cleanup",
            headers=admin_headers,
            json={
                "reason": "transfer to successor",
                "transfer_to_person_id": str(successor.id),
            },
        )
    assert resp.status_code == 200, resp.text

    # Driver NOT called — transfer is Helix-side only.
    assert driver.calls == []
    # Grant unchanged.
    after = await db_session.get(CmdbCloudAccessGrant, grant.id)
    await db_session.refresh(after)
    assert after.status == "active"
    assert after.revoke_reason is None


# ── Regression (code-review pass 1): irreversible disown hardening ──
#
# Disown runs the provider's IAM ``DeleteUser`` chain and is irreversible,
# so it is reserved for global_admin (matches ``can_manage_cloud_scope_
# lifecycle`` + the endpoint's own error message) and honours the global
# prod-write lock — exactly like the sibling server ``destroy`` lifecycle
# op. The reversible *transfer* path stays open to cloud_admin.


@pytest.mark.asyncio
async def test_cloud_admin_cannot_disown(
    client: AsyncClient,
    db_session: AsyncSession,
    cloud_admin_user,
    cloud_admin_headers,
):
    """cloud_admin may operate cloud resources, but the irreversible IAM
    disown (real ``DeleteUser``) is reserved for global_admin. The gate
    must short-circuit before the driver is ever loaded."""
    integ = await _make_integration(db_session)
    iam = await _make_iam_user(db_session, integ, user_name="grace")

    driver = _DriverOK()
    with _patch_driver(driver):
        resp = await client.post(
            f"/api/v1/cmdb/iam-users/{iam.id}/cleanup",
            headers=cloud_admin_headers,
            json={"reason": "trying to disown as cloud_admin"},
        )

    assert resp.status_code == 403, resp.text
    # Driver MUST NOT have been called — the gate stops the destructive
    # path before any provider call.
    assert driver.calls == []
    refreshed = await db_session.get(CmdbIamUser, iam.id)
    await db_session.refresh(refreshed)
    assert refreshed.revoked_at is None


@pytest.mark.asyncio
async def test_cloud_admin_can_still_transfer(
    client: AsyncClient,
    db_session: AsyncSession,
    cloud_admin_user,
    cloud_admin_headers,
):
    """The reversible Helix-side transfer path stays open to cloud_admin —
    only the irreversible disown is gated up to global_admin."""
    integ = await _make_integration(db_session)
    successor = IdentityPerson(
        id=uuid.uuid4(),
        work_email="successor2@test.com",
        display_name="Successor Two",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(successor)
    await db_session.commit()
    iam = await _make_iam_user(db_session, integ, user_name="heidi")

    driver = _DriverOK()
    with _patch_driver(driver):
        resp = await client.post(
            f"/api/v1/cmdb/iam-users/{iam.id}/cleanup",
            headers=cloud_admin_headers,
            json={
                "reason": "transfer to successor",
                "transfer_to_person_id": str(successor.id),
            },
        )

    assert resp.status_code == 200, resp.text
    payload = resp.json()["data"]
    assert payload["owner_person_id"] == str(successor.id)
    assert payload["revoked_at"] is None
    assert driver.calls == []  # transfer is Helix-side only


@pytest.mark.asyncio
async def test_disown_blocked_when_prod_scope_and_writes_locked(
    client: AsyncClient,
    db_session: AsyncSession,
    admin_user,
    admin_headers,
):
    """When the IAM user's integration hosts an active prod scope and the
    global prod-write lock is engaged (default-locked: setting absent),
    even global_admin's disown is refused with PROD_CLOUD_LOCKED — the
    destructive provider call never fires."""
    integ = await _make_integration(db_session)
    # Active PROD scope on the same integration → the IAM user inherits
    # prod classification for the kill-switch.
    scope = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integ.id),
        scope_type="standalone_iam",
        external_id="999999999999",
        display_name="AWS Prod (test)",
        status="active",
        environment="prod",
        first_seen_at=None,
    )
    db_session.add(scope)
    await db_session.commit()
    iam = await _make_iam_user(db_session, integ, user_name="ivan")

    driver = _DriverOK()
    with _patch_driver(driver):
        resp = await client.post(
            f"/api/v1/cmdb/iam-users/{iam.id}/cleanup",
            headers=admin_headers,
            json={"reason": "leaver cleanup during prod lockdown"},
        )

    assert resp.status_code == 403, resp.text
    assert resp.json()["errors"][0]["code"] == "PROD_CLOUD_LOCKED"
    # Kill-switch fired before the driver — nothing was deleted cloud-side.
    assert driver.calls == []
    refreshed = await db_session.get(CmdbIamUser, iam.id)
    await db_session.refresh(refreshed)
    assert refreshed.revoked_at is None