"""Round 16 audit: ``CloudLifecycleService.revoke_grant(...,
suppress_driver_error=False)`` must NOT silently flip the Helix-side
grant to REVOKED when the cloud provider rejects the revoke.

Bug class — the previous router endpoint passed
``suppress_driver_error=True`` to the lifecycle service, which meant:

  * driver returns ok=False (5xx, IAM lock, quota) → grant.status =
    REVOKED, audit row says "failure", but the UI showed "已撤销".
  * Helix's UI hides the grant, employee thinks the seat is gone, but
    cloud-side still has the role attached.

Real compliance + offboarding-trust hazard: the audit log disagrees
with the UI, and the next sync would re-discover the grant and
either raise a "drift" alert or quietly recreate the row.

The fix flips the router to ``suppress_driver_error=False`` and the
service raises ``GrantDispatchError`` on hard cloud failure (502 to
the user), while still treating ``already_absent=True`` as
idempotent success so the "sweeper-then-admin-click" race doesn't
502.

Tested at the service layer (not the HTTP layer) because the test
fixture's shared session makes assertion-on-rollback brittle through
the FastAPI test client.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime
from unittest.mock import patch

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.cmdb.lifecycle_service import (
    CloudLifecycleService,
    GrantDispatchError,
)
from app.domains.cmdb.models import (
    GRANT_STATUS_ACTIVE,
    GRANT_STATUS_REVOKED,
    SCOPE_STATUS_ACTIVE,
    CmdbCloudAccessGrant,
    CmdbCloudScope,
)
from app.identity.person_models import IdentityPerson
from app.integrations.driver_base import CloudGrantResult
from app.integrations.models import Integration


async def _setup_grant(
    db: AsyncSession,
) -> tuple[Integration, CmdbCloudScope, CmdbCloudAccessGrant, IdentityPerson]:
    holder = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"holder-{uuid.uuid4().hex[:6]}@test.com",
        display_name="Grant Holder",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db.add(holder)
    await db.flush()

    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="azure",
        name=f"azure-{uuid.uuid4().hex[:6]}",
        slug=f"azure-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers"],
    )
    db.add(integ)
    await db.flush()

    scope = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integ.id),
        scope_type="azure_subscription",
        external_id=uuid.uuid4().hex[:16],
        display_name=f"sub-{uuid.uuid4().hex[:6]}",
        status=SCOPE_STATUS_ACTIVE,
        first_seen_at=datetime.now(UTC),
        last_synced_at=datetime.now(UTC),
        environment="dev",
    )
    db.add(scope)
    await db.flush()

    grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(holder.id),
        external_principal_email=holder.work_email,
        role_key="reader",
        role_display_name="Reader",
        status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC),
        external_ref="ra-fake-1",
    )
    db.add(grant)
    await db.commit()
    return integ, scope, grant, holder


@pytest.mark.asyncio
async def test_revoke_grant_cloud_failure_raises_and_keeps_active(
    db_session: AsyncSession,
):
    """Driver returns ok=False with no ``already_absent`` flag →
    service raises GrantDispatchError; the in-memory grant.status
    flip is rolled back so the Helix-side row stays ACTIVE.
    """
    _integ, _scope, grant, holder = await _setup_grant(db_session)
    # Snapshot ids before the rollback below — the rollback expires
    # the grant ORM row, so reaching ``grant.id`` afterwards triggers
    # an autoload that fails with a ``MissingGreenlet`` error.
    grant_pk = grant.id
    holder_id = holder.id

    class _FailingDriver:
        async def revoke_access(self, ctx):
            return CloudGrantResult(
                ok=False,
                error_message="Azure RBAC: 503 Service Unavailable",
            )
        async def close(self):
            return None

    async def _stub_loader(_db, _scope):
        return None, _FailingDriver()

    service = CloudLifecycleService(db_session)
    with patch(
        "app.domains.cmdb.lifecycle_service._load_driver_for_scope",
        _stub_loader,
    ):
        with pytest.raises(GrantDispatchError):
            await service.revoke_grant(
                str(grant_pk),
                reason="user_released",
                actor_user_id=holder_id,
                trace_id="test-trace",
                suppress_driver_error=False,
            )

    # Rollback the in-memory mutations queued before the raise so the
    # next SELECT against the shared session reads the on-disk truth.
    await db_session.rollback()

    refreshed = await db_session.get(
        CmdbCloudAccessGrant, grant_pk, populate_existing=True,
    )
    assert refreshed is not None
    assert refreshed.status == GRANT_STATUS_ACTIVE, (
        f"grant.status={refreshed.status} — must NOT flip to REVOKED "
        "when the cloud rejected the revoke and the caller passed "
        "suppress_driver_error=False"
    )


@pytest.mark.asyncio
async def test_revoke_grant_already_absent_succeeds_idempotently(
    db_session: AsyncSession,
):
    """Driver returns ``ok=False, already_absent=True`` (cloud-side
    404 — assignment already gone) → service treats as success and
    flips status to REVOKED. Pins the idempotent path so the
    suppress-flag fix doesn't 502 on harmless "user clicked revoke
    twice" / "sweeper then admin" races.
    """
    _integ, _scope, grant, holder = await _setup_grant(db_session)

    class _AlreadyAbsentDriver:
        async def revoke_access(self, ctx):
            return CloudGrantResult(
                ok=False,
                already_absent=True,
                error_message="404 Not Found (already absent)",
            )
        async def close(self):
            return None

    async def _stub_loader(_db, _scope):
        return None, _AlreadyAbsentDriver()

    service = CloudLifecycleService(db_session)
    with patch(
        "app.domains.cmdb.lifecycle_service._load_driver_for_scope",
        _stub_loader,
    ):
        out = await service.revoke_grant(
            str(grant.id),
            reason="user_released",
            actor_user_id=holder.id,
            trace_id="test-trace",
            suppress_driver_error=False,
        )

    assert out.status == GRANT_STATUS_REVOKED


@pytest.mark.asyncio
async def test_revoke_grant_suppress_true_keeps_legacy_offboarding_path(
    db_session: AsyncSession,
):
    """Offboarding / sweep callers still pass
    ``suppress_driver_error=True`` and rely on the legacy "flip
    REVOKED regardless of cloud-side outcome" semantics. This test
    pins that contract so the round-16 fix doesn't accidentally
    break the sweep flow.
    """
    _integ, _scope, grant, _holder = await _setup_grant(db_session)

    class _FailingDriver:
        async def revoke_access(self, ctx):
            return CloudGrantResult(ok=False, error_message="cloud 500")
        async def close(self):
            return None

    async def _stub_loader(_db, _scope):
        return None, _FailingDriver()

    service = CloudLifecycleService(db_session)
    with patch(
        "app.domains.cmdb.lifecycle_service._load_driver_for_scope",
        _stub_loader,
    ):
        out = await service.revoke_grant(
            str(grant.id),
            reason="leaver_offboarding",
            actor_user_id=None,
            trace_id="test-trace",
            suppress_driver_error=True,
        )

    # Sweep path: status flips to revoked even on cloud failure (so
    # the next sweep can retry; the audit row records the failure).
    assert out.status == GRANT_STATUS_REVOKED


@pytest.mark.asyncio
async def test_scope_owner_can_revoke_other_holder_grant(
    db_session: AsyncSession,
):
    """Round-16 widened the service-layer RBAC to also accept scope
    owners (the router gate already accepts them; the service-layer
    check would otherwise 403 the legitimate scope-owner-revokes-
    someone-else's-grant flow when the router calls
    ``suppress_driver_error=False``). Pin the contract so a future
    refactor doesn't silently re-narrow it.
    """
    _integ, scope, grant, holder = await _setup_grant(db_session)

    # Materialise a scope owner who is NOT the grant holder and NOT
    # an admin — exactly the case the legacy service-level check
    # would have 403'd before the round-16 widening.
    scope_owner = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"owner-{uuid.uuid4().hex[:6]}@test.com",
        display_name="Scope Owner",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(scope_owner)
    await db_session.flush()
    scope.owner_person_id = str(scope_owner.id)
    await db_session.commit()

    class _OkDriver:
        async def revoke_access(self, ctx):
            return CloudGrantResult(ok=True, external_ref="ok")
        async def close(self):
            return None

    async def _stub_loader(_db, _scope):
        return None, _OkDriver()

    service = CloudLifecycleService(db_session)
    with patch(
        "app.domains.cmdb.lifecycle_service._load_driver_for_scope",
        _stub_loader,
    ):
        out = await service.revoke_grant(
            str(grant.id),
            reason="admin_removed",
            actor_user_id=scope_owner.id,
            trace_id="test-trace",
            suppress_driver_error=False,
            actor_role_names=["employee"],  # NOT cloud_admin
        )

    assert out.status == GRANT_STATUS_REVOKED


@pytest.mark.asyncio
async def test_random_employee_cannot_revoke_other_grant_via_service(
    db_session: AsyncSession,
):
    """Service-layer defense-in-depth still rejects a non-holder,
    non-scope-owner, non-cloud-admin caller with 403. Pins the
    "fall-through" branch of the round-16 widening so a future
    refactor doesn't accidentally drop the gate entirely.
    """
    from app.core.exceptions import AppException

    _integ, _scope, grant, _holder = await _setup_grant(db_session)
    intruder = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"intruder-{uuid.uuid4().hex[:6]}@test.com",
        display_name="Intruder",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(intruder)
    await db_session.commit()

    service = CloudLifecycleService(db_session)
    with pytest.raises(AppException) as exc:
        await service.revoke_grant(
            str(grant.id),
            reason="admin_removed",
            actor_user_id=intruder.id,
            trace_id="test-trace",
            suppress_driver_error=False,
            actor_role_names=["employee"],
        )
    assert exc.value.status_code == 403