"""Tests for the self-service API-Key transfer flow (PR2 of the API-Key
lifecycle closure, 2026-05-19).

Three properties under test:

1. ``submit_transfer_request`` delegates to
   :meth:`OffboardingService.create_transfer` correctly: an
   AssetTransferRequest is queued (status='pending'), the apikey row is
   NOT touched yet (cloud credential stays live), and an
   apikey.transfer.request audit lands.

2. When the recipient accepts (via the existing
   ``OffboardingService.respond_to_transfer`` path),
   :meth:`ApiKeyService.apply_transfer` fires:
     * ``assigned_to`` flips to the recipient
     * ``transferred_from`` captures the prior owner
     * Cost-allocation rules pointed at by
       ``detail.<bucket>.allocation_rule_id`` have their
       ``assign_person_id`` swapped to the new owner
     * ``detail.last_transfer`` is stamped with the transfer metadata
     * an ``apikey.transfer.complete`` audit row is written

3. Owner can cancel their own pending transfer via
   ``cancel_transfer_request`` — the AssetTransferRequest moves to
   ``cancelled`` and the assignment is unchanged.

The cloud-side IAM tag rewrite is exercised via a monkeypatched driver
(no boto3 / real AWS calls); we verify the tag dict that would have
been pushed.
"""
from __future__ import annotations

import uuid
from decimal import Decimal
from types import SimpleNamespace
from typing import Any
from unittest.mock import patch

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

from app.audit.models import AuditLog
from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider
from app.domains.apikey.schemas import ApiKeyTransferRequestSubmit
from app.domains.apikey.service import (
    STATUS_ACTIVE,
    STATUS_REVOKED,
    ApiKeyService,
)
from app.domains.cmdb.models import CmdbCostAllocationRule
from app.domains.offboarding.models import (
    XFER_CANCELLED,
    XFER_PENDING,
    AssetTransferRequest,
)
from app.domains.offboarding.service import OffboardingService
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration


def _user_auth(person: IdentityPerson) -> SimpleNamespace:
    return SimpleNamespace(
        id=person.id,
        role_names=["employee"],
        work_email=person.work_email,
        display_name=person.display_name,
    )


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


async def _make_provider(db: AsyncSession, *, name: str) -> ApiKeyProvider:
    p = ApiKeyProvider(
        id=uuid.uuid4(),
        name=name,
        description="",
        is_active=True,
        sort_order=0,
    )
    db.add(p)
    await db.commit()
    return p


async def _make_active_assignment(
    db: AsyncSession,
    *,
    provider: ApiKeyProvider,
    applicant: IdentityPerson,
    detail: dict[str, Any] | None = None,
    iam_integration: Integration | None = None,
    iam_user_arn: str | None = None,
) -> ApiKeyAssignment:
    a = ApiKeyAssignment(
        id=uuid.uuid4(),
        provider_id=provider.id,
        assigned_to=applicant.id,
        status=STATUS_ACTIVE,
        request_number=f"AK-{uuid.uuid4().hex[:8].upper()}",
        project_name="transfer-test",
        reason="transfer flow tests",
        monthly_budget_usd=Decimal("100.00"),
        key_value=None,
        detail=detail or {},
        iam_integration_id=str(iam_integration.id) if iam_integration else None,
        iam_user_arn=iam_user_arn,
    )
    db.add(a)
    await db.commit()
    return a


async def _make_aws_integration(db: AsyncSession) -> Integration:
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="aws",
        name="AWS Test",
        slug=f"aws-xfer-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=[],
        config_public={},
    )
    db.add(integ)
    await db.commit()
    return integ


async def _make_allocation_rule(
    db: AsyncSession,
    *,
    name: str,
    integration: Integration,
    assign_person: IdentityPerson,
    match_tag_value: str,
) -> CmdbCostAllocationRule:
    rule = CmdbCostAllocationRule(
        id=str(uuid.uuid4()),
        name=name,
        priority=10,
        match_integration_ids=[str(integration.id)],
        match_provider=integration.provider,
        match_tag_key="Owner",
        match_tag_value=match_tag_value,
        assign_person_id=str(assign_person.id),
        enabled=True,
        description="autogrant test rule",
    )
    db.add(rule)
    await db.commit()
    return rule


@pytest.mark.asyncio
async def test_submit_transfer_request_creates_pending_xfer(
    db_session: AsyncSession,
):
    """Submit transfer → AssetTransferRequest with status=pending,
    initiator=owner, recipient=chosen person, asset_type='api_key'.
    Assignment row UNCHANGED (cloud credential stays live)."""
    owner = await _make_person(db_session, email="ada-xfer@pingcap.com")
    recipient = await _make_person(db_session, email="bob-xfer@pingcap.com")
    provider = await _make_provider(db_session, name="Provider X")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=owner,
    )

    service = ApiKeyService(db_session)
    await service.submit_transfer_request(
        assignment.id,
        ApiKeyTransferRequestSubmit(
            recipient_person_id=recipient.id,
            note="taking on this project",
        ),
        current_user=_user_auth(owner),
        trace_id="t-xfer-1",
        ip_address="127.0.0.1",
        user_agent="pytest",
    )
    await db_session.commit()

    delivered: list[str] = []

    async def fake_deliver(_self, _outbox, envelope, **_kwargs):
        delivered.append(envelope.template_key)

    with patch(
        "app.notifications.service.NotificationService.deliver",
        new=fake_deliver,
    ):
        await service.dispatch_transfer_notifications_after_commit(
            trace_id="t-xfer-1",
        )
    assert delivered == ["transfer_received"]

    # Assignment unchanged at this point.
    await db_session.refresh(assignment)
    assert assignment.assigned_to == owner.id
    assert assignment.transferred_from is None
    assert assignment.status == STATUS_ACTIVE

    # AssetTransferRequest exists, pending.
    xfer = (
        await db_session.execute(
            select(AssetTransferRequest).where(
                AssetTransferRequest.asset_type == "api_key",
                AssetTransferRequest.asset_ref_id == assignment.id,
            ),
        )
    ).scalar_one()
    assert xfer.status == XFER_PENDING
    assert xfer.initiator_id == owner.id
    assert xfer.recipient_id == recipient.id
    assert xfer.reason == "taking on this project"

    # Audit row landed.
    audits = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "apikey.transfer.request",
                AuditLog.resource_id == str(assignment.id),
            ),
        )
    ).scalars().all()
    assert len(audits) == 1


@pytest.mark.asyncio
async def test_submit_transfer_owner_only(db_session: AsyncSession):
    """A non-owner attempting to transfer someone else's key → 403."""
    from app.core.exceptions import ForbiddenError

    owner = await _make_person(db_session, email="ada-owner@pingcap.com")
    intruder = await _make_person(
        db_session, email="intruder-xfer@pingcap.com",
    )
    recipient = await _make_person(db_session, email="recip-xfer@pingcap.com")
    provider = await _make_provider(db_session, name="Provider X2")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=owner,
    )

    with pytest.raises(ForbiddenError):
        await ApiKeyService(db_session).submit_transfer_request(
            assignment.id,
            ApiKeyTransferRequestSubmit(recipient_person_id=recipient.id),
            current_user=_user_auth(intruder),
            trace_id="t-xfer-intruder",
            ip_address=None,
            user_agent=None,
        )


@pytest.mark.asyncio
async def test_submit_transfer_rejects_revoked_assignment(
    db_session: AsyncSession,
):
    """Transfer is only valid for active / expired keys; a revoked
    key returns INVALID_STATUS_FOR_TRANSFER 400."""
    from app.core.exceptions import AppException

    owner = await _make_person(db_session, email="evan-xfer@pingcap.com")
    recipient = await _make_person(db_session, email="frank-xfer@pingcap.com")
    provider = await _make_provider(db_session, name="Provider Z")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=owner,
    )
    assignment.status = STATUS_REVOKED
    await db_session.commit()

    with pytest.raises(AppException) as exc:
        await ApiKeyService(db_session).submit_transfer_request(
            assignment.id,
            ApiKeyTransferRequestSubmit(recipient_person_id=recipient.id),
            current_user=_user_auth(owner),
            trace_id="t-xfer-revoked",
            ip_address=None,
            user_agent=None,
        )
    assert exc.value.code == "INVALID_STATUS_FOR_TRANSFER"


@pytest.mark.asyncio
async def test_accept_transfer_swaps_owner_cost_rule_and_iam_tag(
    db_session: AsyncSession,
    monkeypatch,
):
    """End-to-end: submit → recipient accepts → assigned_to flips,
    transferred_from captured, cost-allocation rule's assign_person_id
    swapped, AWS IAM driver was called with the new owner's email
    tags."""
    owner = await _make_person(
        db_session, email="ada-bedrock@pingcap.com",
        display_name="Ada",
    )
    recipient = await _make_person(
        db_session, email="bob-bedrock@pingcap.com",
        display_name="Bob",
    )
    provider = await _make_provider(db_session, name="AWS Bedrock")
    aws_integ = await _make_aws_integration(db_session)
    rule = await _make_allocation_rule(
        db_session,
        name=f"AWS Bedrock · {owner.work_email}",
        integration=aws_integ,
        assign_person=owner,
        match_tag_value=owner.work_email or "",
    )
    assignment = await _make_active_assignment(
        db_session,
        provider=provider,
        applicant=owner,
        detail={
            "bedrock": {
                "granted": True,
                "iam_user_name": "helix-bedrock-ada",
                "allocation_rule_id": rule.id,
            },
        },
        iam_integration=aws_integ,
        iam_user_arn=(
            "arn:aws:iam::000000000000:user/helix-bedrock-ada"
        ),
    )

    # Capture the driver call without hitting AWS.
    captured: dict[str, Any] = {}

    class _FakeDriver:
        async def tag_iam_user(self, user_name: str, tags: dict[str, str]):
            captured["user_name"] = user_name
            captured["tags"] = dict(tags)
            return SimpleNamespace(
                ok=True, external_ref=user_name, error_message=None,
            )

        async def close(self):
            return None

    async def fake_build_driver(self, integ):  # noqa: ANN001
        return _FakeDriver()

    monkeypatch.setattr(
        ApiKeyService,
        "_build_aws_driver_for_tag_update",
        fake_build_driver,
    )

    # Submit then accept.
    await ApiKeyService(db_session).submit_transfer_request(
        assignment.id,
        ApiKeyTransferRequestSubmit(
            recipient_person_id=recipient.id,
            note="end-to-end",
        ),
        current_user=_user_auth(owner),
        trace_id="t-xfer-e2e",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    xfer = (
        await db_session.execute(
            select(AssetTransferRequest).where(
                AssetTransferRequest.asset_ref_id == assignment.id,
            ),
        )
    ).scalar_one()

    await OffboardingService(db_session).respond_to_transfer(
        transfer_id=xfer.id,
        accept=True,
        acting_user_id=recipient.id,
        response_note=None,
        trace_id="t-xfer-e2e-accept",
    )
    await db_session.commit()

    await db_session.refresh(assignment)
    assert assignment.assigned_to == recipient.id
    assert assignment.transferred_from == owner.id
    assert assignment.status == STATUS_ACTIVE

    await db_session.refresh(rule)
    assert rule.assign_person_id == str(recipient.id)
    assert "[transferred " in (rule.description or "")

    # detail.last_transfer captures the metadata for the admin tooltip.
    last = (assignment.detail or {}).get("last_transfer") or {}
    assert last.get("from_person_id") == str(owner.id)
    assert last.get("to_person_id") == str(recipient.id)
    assert last.get("transfer_request_id") == str(xfer.id)
    rule_swaps = last.get("rule_swaps") or []
    assert any(
        rs.get("bucket") == "bedrock" and rs.get("status") == "swapped"
        for rs in rule_swaps
    )

    # AWS tag_iam_user was called with the recipient's email.
    assert captured.get("user_name") == "helix-bedrock-ada"
    assert captured.get("tags", {}).get("Owner") == recipient.work_email
    assert (
        captured.get("tags", {}).get("helix-owner-email")
        == recipient.work_email
    )

    # apikey.transfer.complete audit landed.
    complete_audits = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "apikey.transfer.complete",
                AuditLog.resource_id == str(assignment.id),
            ),
        )
    ).scalars().all()
    assert len(complete_audits) == 1


@pytest.mark.asyncio
async def test_accept_transfer_without_cloud_binding_is_a_noop(
    db_session: AsyncSession,
    monkeypatch,
):
    """Non-cloud assignment (no IAM integration / no allocation rule) —
    transfer still flips the owner and writes the audit row, but the
    cloud-tag summary in detail.last_transfer is empty."""
    owner = await _make_person(db_session, email="ada-saas@pingcap.com")
    recipient = await _make_person(db_session, email="bob-saas@pingcap.com")
    provider = await _make_provider(db_session, name="OpenAI Direct")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=owner,
    )

    await ApiKeyService(db_session).submit_transfer_request(
        assignment.id,
        ApiKeyTransferRequestSubmit(recipient_person_id=recipient.id),
        current_user=_user_auth(owner),
        trace_id="t-xfer-saas",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()
    xfer = (
        await db_session.execute(
            select(AssetTransferRequest).where(
                AssetTransferRequest.asset_ref_id == assignment.id,
            ),
        )
    ).scalar_one()

    await OffboardingService(db_session).respond_to_transfer(
        transfer_id=xfer.id,
        accept=True,
        acting_user_id=recipient.id,
        response_note=None,
        trace_id="t-xfer-saas-accept",
    )
    await db_session.commit()

    await db_session.refresh(assignment)
    assert assignment.assigned_to == recipient.id
    assert assignment.transferred_from == owner.id

    last = (assignment.detail or {}).get("last_transfer") or {}
    # No cloud bindings → empty tag summary, no rule swaps.
    assert last.get("cloud_tag_summary") == {}
    assert last.get("rule_swaps") == []


@pytest.mark.asyncio
async def test_accept_transfer_survives_aws_tag_failure(
    db_session: AsyncSession,
    monkeypatch,
):
    """AWS IAM tag rewrite is best-effort: a failure (e.g. IAM user
    deleted out of band before transfer) must NOT roll back the
    transfer. The local row + cost-allocation rule are already
    authoritative for Helix attribution; the tag failure lands in
    audit detail with status='failed' so an admin can spot it and
    retry manually via the AWS console.
    """
    owner = await _make_person(db_session, email="hank-fail@pingcap.com")
    recipient = await _make_person(db_session, email="iris-fail@pingcap.com")
    provider = await _make_provider(db_session, name="AWS Bedrock")
    aws_integ = await _make_aws_integration(db_session)
    rule = await _make_allocation_rule(
        db_session,
        name=f"AWS Bedrock · {owner.work_email} · fail-test",
        integration=aws_integ,
        assign_person=owner,
        match_tag_value=owner.work_email or "",
    )
    assignment = await _make_active_assignment(
        db_session,
        provider=provider,
        applicant=owner,
        detail={
            "bedrock": {
                "granted": True,
                "iam_user_name": "helix-bedrock-hank-deleted",
                "allocation_rule_id": rule.id,
            },
        },
        iam_integration=aws_integ,
        iam_user_arn=(
            "arn:aws:iam::000000000000:user/helix-bedrock-hank-deleted"
        ),
    )

    # Driver whose tag call returns ok=False (simulates AWS NoSuchEntity
    # after the IAM user was manually deleted by an admin).
    class _FailingDriver:
        async def tag_iam_user(self, user_name: str, tags: dict):  # noqa: ANN001
            return SimpleNamespace(
                ok=False,
                external_ref=user_name,
                error_message="AWS ClientError NoSuchEntity: User does not exist",
            )

        async def close(self):
            return None

    async def fake_build_driver(self, integ):  # noqa: ANN001
        return _FailingDriver()

    monkeypatch.setattr(
        ApiKeyService,
        "_build_aws_driver_for_tag_update",
        fake_build_driver,
    )

    # Submit + accept; transfer MUST complete despite the tag failure.
    await ApiKeyService(db_session).submit_transfer_request(
        assignment.id,
        ApiKeyTransferRequestSubmit(recipient_person_id=recipient.id),
        current_user=_user_auth(owner),
        trace_id="t-xfer-fail",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()
    xfer = (
        await db_session.execute(
            select(AssetTransferRequest).where(
                AssetTransferRequest.asset_ref_id == assignment.id,
            ),
        )
    ).scalar_one()

    await OffboardingService(db_session).respond_to_transfer(
        transfer_id=xfer.id,
        accept=True,
        acting_user_id=recipient.id,
        response_note=None,
        trace_id="t-xfer-fail-accept",
    )
    await db_session.commit()

    # Transfer completed despite the AWS failure.
    await db_session.refresh(assignment)
    assert assignment.assigned_to == recipient.id
    assert assignment.transferred_from == owner.id
    # Cost rule still swapped.
    await db_session.refresh(rule)
    assert rule.assign_person_id == str(recipient.id)

    # And the audit / detail captures the AWS tag failure for forensics.
    summary = (
        (assignment.detail or {}).get("last_transfer", {}).get(
            "cloud_tag_summary", {},
        )
    )
    assert summary.get("bedrock", {}).get("status") == "failed"
    assert "NoSuchEntity" in summary["bedrock"].get("reason", "")


@pytest.mark.asyncio
async def test_cancel_transfer_request_flips_to_cancelled(
    db_session: AsyncSession,
):
    """Owner withdraws their pending transfer → AssetTransferRequest
    moves to status='cancelled'; assignment unchanged; idempotent on
    second call."""
    owner = await _make_person(db_session, email="ada-cancel@pingcap.com")
    recipient = await _make_person(db_session, email="bob-cancel@pingcap.com")
    provider = await _make_provider(db_session, name="Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=owner,
    )

    await ApiKeyService(db_session).submit_transfer_request(
        assignment.id,
        ApiKeyTransferRequestSubmit(recipient_person_id=recipient.id),
        current_user=_user_auth(owner),
        trace_id="t-xfer-cancel",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    await ApiKeyService(db_session).cancel_transfer_request(
        assignment.id,
        current_user=_user_auth(owner),
        trace_id="t-xfer-cancel-1",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    xfer = (
        await db_session.execute(
            select(AssetTransferRequest).where(
                AssetTransferRequest.asset_ref_id == assignment.id,
            ),
        )
    ).scalar_one()
    assert xfer.status == XFER_CANCELLED

    # Idempotent: second cancel is a no-op (returns current state, no
    # second audit row, no exception).
    await ApiKeyService(db_session).cancel_transfer_request(
        assignment.id,
        current_user=_user_auth(owner),
        trace_id="t-xfer-cancel-2",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()
    # Single cancel audit only.
    audits = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "apikey.transfer.cancel",
                AuditLog.resource_id == str(assignment.id),
            ),
        )
    ).scalars().all()
    assert len(audits) == 1


@pytest.mark.asyncio
async def test_retry_iam_retag_succeeds_when_driver_recovers(
    db_session: AsyncSession,
    monkeypatch,
):
    """Admin retries the IAM tag rewrite after the underlying AWS
    issue is fixed; the call succeeds and detail.last_transfer
    captures the clean-status retag_attempt history."""
    admin = await _make_person(db_session, email="admin-retag@pingcap.com")
    owner = await _make_person(db_session, email="cur-owner@pingcap.com")
    provider = await _make_provider(db_session, name="AWS Bedrock")
    aws_integ = await _make_aws_integration(db_session)
    assignment = await _make_active_assignment(
        db_session,
        provider=provider,
        applicant=owner,
        detail={
            "bedrock": {"granted": True},
            "last_transfer": {
                "cloud_tag_summary": {
                    "bedrock": {
                        "status": "failed",
                        "reason": "AWS ClientError NoSuchEntity: …",
                    },
                },
            },
        },
        iam_integration=aws_integ,
        iam_user_arn=(
            "arn:aws:iam::000000000000:user/helix-bedrock-current"
        ),
    )

    class _OkDriver:
        async def tag_iam_user(self, user_name: str, tags: dict):  # noqa: ANN001
            return SimpleNamespace(
                ok=True, external_ref=user_name, error_message=None,
            )

        async def close(self):
            return None

    async def fake_build_driver(self, integ):  # noqa: ANN001
        return _OkDriver()

    monkeypatch.setattr(
        ApiKeyService,
        "_build_aws_driver_for_tag_update",
        fake_build_driver,
    )

    admin_auth = SimpleNamespace(
        id=admin.id,
        role_names=["it_admin"],
        work_email=admin.work_email,
    )
    await ApiKeyService(db_session).retry_iam_retag(
        assignment.id,
        current_user=admin_auth,
        trace_id="t-retag-ok",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    await db_session.refresh(assignment)
    summary = (
        (assignment.detail or {}).get("last_transfer", {}).get(
            "cloud_tag_summary", {},
        )
    )
    assert summary["bedrock"]["status"] == "tagged"
    attempts = (
        (assignment.detail or {}).get("last_transfer", {}).get(
            "retag_attempts", [],
        )
    )
    assert len(attempts) == 1
    assert attempts[0]["by_person_id"] == str(admin.id)


@pytest.mark.asyncio
async def test_retry_iam_retag_502_on_persistent_failure(
    db_session: AsyncSession,
    monkeypatch,
):
    """If the underlying driver still returns ok=False, the endpoint
    raises IAM_RETAG_FAILED 502 and the detail captures the failed
    attempt in retag_attempts for forensics."""
    from app.core.exceptions import AppException

    admin = await _make_person(db_session, email="admin-stuck@pingcap.com")
    owner = await _make_person(db_session, email="cur-owner2@pingcap.com")
    provider = await _make_provider(db_session, name="AWS Bedrock")
    aws_integ = await _make_aws_integration(db_session)
    assignment = await _make_active_assignment(
        db_session,
        provider=provider,
        applicant=owner,
        detail={
            "bedrock": {"granted": True},
            "last_transfer": {
                "cloud_tag_summary": {
                    "bedrock": {"status": "failed", "reason": "old-failure"},
                },
            },
        },
        iam_integration=aws_integ,
        iam_user_arn="arn:aws:iam::000000000000:user/helix-stuck",
    )

    class _StillFailingDriver:
        async def tag_iam_user(self, user_name: str, tags: dict):  # noqa: ANN001
            return SimpleNamespace(
                ok=False,
                external_ref=user_name,
                error_message="AccessDenied: still no perms",
            )

        async def close(self):
            return None

    async def fake_build_driver(self, integ):  # noqa: ANN001
        return _StillFailingDriver()

    monkeypatch.setattr(
        ApiKeyService,
        "_build_aws_driver_for_tag_update",
        fake_build_driver,
    )

    admin_auth = SimpleNamespace(
        id=admin.id,
        role_names=["it_admin"],
        work_email=admin.work_email,
    )
    with pytest.raises(AppException) as excinfo:
        await ApiKeyService(db_session).retry_iam_retag(
            assignment.id,
            current_user=admin_auth,
            trace_id="t-retag-fail",
            ip_address=None,
            user_agent=None,
        )
    assert excinfo.value.code == "IAM_RETAG_FAILED"
    assert excinfo.value.status_code == 502
    assert "AccessDenied" in excinfo.value.message

    await db_session.commit()
    await db_session.refresh(assignment)
    # New attempt is recorded.
    attempts = (
        (assignment.detail or {}).get("last_transfer", {}).get(
            "retag_attempts", [],
        )
    )
    assert len(attempts) == 1
    assert attempts[0]["summary"]["bedrock"]["status"] == "failed"


@pytest.mark.asyncio
async def test_retry_iam_retag_admin_only(db_session: AsyncSession):
    """A non-admin caller gets FORBIDDEN."""
    from app.core.exceptions import AppException

    non_admin = await _make_person(db_session, email="emp-nope@pingcap.com")
    owner = await _make_person(db_session, email="o-nope@pingcap.com")
    provider = await _make_provider(db_session, name="AWS Bedrock")
    aws_integ = await _make_aws_integration(db_session)
    assignment = await _make_active_assignment(
        db_session,
        provider=provider,
        applicant=owner,
        iam_integration=aws_integ,
        iam_user_arn="arn:aws:iam::0:user/x",
    )

    with pytest.raises(AppException) as excinfo:
        await ApiKeyService(db_session).retry_iam_retag(
            assignment.id,
            current_user=_user_auth(non_admin),
            trace_id="t-retag-403",
            ip_address=None,
            user_agent=None,
        )
    assert excinfo.value.code == "FORBIDDEN"


@pytest.mark.asyncio
async def test_retry_iam_retag_400_when_revoked(
    db_session: AsyncSession,
):
    """A revoked assignment's IAM user is already deleted by the
    revoke teardown chain; retrying the tag would 404 at AWS. Refuse
    up-front so the admin sees a truthful error."""
    from app.core.exceptions import AppException

    admin = await _make_person(db_session, email="admin-revoked-retag@pingcap.com")
    owner = await _make_person(db_session, email="o-revoked@pingcap.com")
    provider = await _make_provider(db_session, name="AWS Bedrock")
    aws_integ = await _make_aws_integration(db_session)
    assignment = await _make_active_assignment(
        db_session,
        provider=provider,
        applicant=owner,
        iam_integration=aws_integ,
        iam_user_arn="arn:aws:iam::0:user/x",
    )
    assignment.status = STATUS_REVOKED
    await db_session.commit()

    admin_auth = SimpleNamespace(
        id=admin.id, role_names=["it_admin"],
        work_email=admin.work_email,
    )
    with pytest.raises(AppException) as excinfo:
        await ApiKeyService(db_session).retry_iam_retag(
            assignment.id,
            current_user=admin_auth,
            trace_id="t-retag-revoked",
            ip_address=None,
            user_agent=None,
        )
    assert excinfo.value.code == "ASSIGNMENT_REVOKED"


@pytest.mark.asyncio
async def test_retry_iam_retag_400_when_no_iam_binding(
    db_session: AsyncSession,
):
    """An assignment that's never been bound to an IAM user has nothing
    to retag — endpoint refuses with NO_IAM_BINDING 400."""
    from app.core.exceptions import AppException

    admin = await _make_person(db_session, email="admin-nope@pingcap.com")
    owner = await _make_person(db_session, email="o2-nope@pingcap.com")
    provider = await _make_provider(db_session, name="OpenAI Direct")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=owner,
    )  # no IAM binding

    admin_auth = SimpleNamespace(
        id=admin.id,
        role_names=["it_admin"],
        work_email=admin.work_email,
    )
    with pytest.raises(AppException) as excinfo:
        await ApiKeyService(db_session).retry_iam_retag(
            assignment.id,
            current_user=admin_auth,
            trace_id="t-retag-nobinding",
            ip_address=None,
            user_agent=None,
        )
    assert excinfo.value.code == "NO_IAM_BINDING"


@pytest.mark.asyncio
async def test_cannot_self_transfer(db_session: AsyncSession):
    """``OffboardingService.create_transfer`` rejects recipient == initiator
    with 400 TRANSFER_SELF. Our wrapper bubbles that up."""
    from app.core.exceptions import AppException

    owner = await _make_person(db_session, email="ada-self@pingcap.com")
    provider = await _make_provider(db_session, name="Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=owner,
    )

    with pytest.raises(AppException) as exc:
        await ApiKeyService(db_session).submit_transfer_request(
            assignment.id,
            ApiKeyTransferRequestSubmit(recipient_person_id=owner.id),
            current_user=_user_auth(owner),
            trace_id="t-xfer-self",
            ip_address=None,
            user_agent=None,
        )
    assert exc.value.code == "TRANSFER_SELF"