"""Per-driver ``close_scope_on_provider`` contract tests.

Covers:
  * AWS aws_account → ``organizations.close_account``; idempotent on
    SUSPENDED / PENDING_CLOSURE / AccountNotFoundException.
  * AWS standalone_iam → evacuates Helix-tracked IAM users (real
    DeleteUser chain, not a stub).
  * Aliyun aliyun_account → ResourceDirectory ``GetAccount`` →
    (``UpdateAccount`` when the member is a CloudAccount) →
    ``CheckAccountDelete`` → ``GetAccountDeletionCheckResult`` →
    ``DeleteAccount``. Resumable across the 3–5 minute pre-check.
  * Azure azure_subscription → REST POST
    ``Microsoft.Subscription/cancel``; idempotent on 404.
  * GCP gcp_project → ``projects.delete``; idempotent on 404.
  * Tencent / UCloud standalone_iam → evacuate IAM users via
    ``revoke_account``.
  * Unsupported scope_type per driver → ``NotImplementedError``.

Each test mocks the relevant SDK / HTTP boundary so we exercise the
driver code path without hitting any real cloud.
"""
from __future__ import annotations

from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from app.integrations.driver_base import CloudScopeRef
from app.integrations.drivers import aliyun_cloud
from app.integrations.drivers.alicloud_intl_cloud import (
    AlicloudIntlCloudDriver,
)
from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver
from app.integrations.drivers.aws_cloud import AwsCloudDriver
from app.integrations.drivers.azure_cloud import AzureCloudDriver
from app.integrations.drivers.gcp_cloud import GcpCloudDriver
from app.integrations.drivers.tencent_cloud import TencentCloudDriver
from app.integrations.drivers.ucloud_cloud import UCloudDriver

# ── AWS aws_account → organizations.close_account ─────────────────


def _aws_driver() -> AwsCloudDriver:
    return AwsCloudDriver(
        integration_id="int-aws",
        config={
            "access_key_id": "AKIA-TEST",
            "secret_access_key": "secret",
            "default_region": "us-east-1",
        },
    )


@pytest.mark.asyncio
async def test_aws_close_account_happy_path() -> None:
    org = MagicMock()
    org.describe_account.return_value = {
        "Account": {"Id": "111111111111", "Status": "ACTIVE"},
    }
    org.close_account.return_value = {}
    driver = _aws_driver()
    with patch.object(AwsCloudDriver, "_boto_client", return_value=org):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="aws_account", external_id="111111111111"),
        )
    assert result.ok is True
    assert result.already_closed is False
    assert result.provider_state == "pending_closure"
    org.close_account.assert_called_once_with(AccountId="111111111111")


@pytest.mark.asyncio
async def test_aws_close_account_idempotent_when_already_suspended() -> None:
    org = MagicMock()
    org.describe_account.return_value = {
        "Account": {"Id": "222222222222", "Status": "SUSPENDED"},
    }
    driver = _aws_driver()
    with patch.object(AwsCloudDriver, "_boto_client", return_value=org):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="aws_account", external_id="222222222222"),
        )
    assert result.ok is True
    assert result.already_closed is True
    assert result.provider_state == "suspended"
    org.close_account.assert_not_called()


@pytest.mark.asyncio
async def test_aws_close_account_constraint_violation_returns_actionable_error() -> None:
    from botocore.exceptions import ClientError
    org = MagicMock()
    org.describe_account.return_value = {
        "Account": {"Id": "333333333333", "Status": "ACTIVE"},
    }
    org.close_account.side_effect = ClientError(
        error_response={
            "Error": {
                "Code": "ConstraintViolationException",
                "Message": "Account does not have a valid payment method.",
            },
        },
        operation_name="CloseAccount",
    )
    driver = _aws_driver()
    with patch.object(AwsCloudDriver, "_boto_client", return_value=org):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="aws_account", external_id="333333333333"),
        )
    assert result.ok is False
    assert result.error_code == "constraint_violation"
    assert "payment method" in (result.error_message or "")


@pytest.mark.asyncio
async def test_aws_close_account_access_denied_explains_management_account() -> None:
    from botocore.exceptions import ClientError
    org = MagicMock()
    org.describe_account.side_effect = ClientError(
        error_response={
            "Error": {"Code": "AccessDeniedException", "Message": "denied"},
        },
        operation_name="DescribeAccount",
    )
    driver = _aws_driver()
    with patch.object(AwsCloudDriver, "_boto_client", return_value=org):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="aws_account", external_id="444444444444"),
        )
    assert result.ok is False
    assert result.error_code == "access_denied"
    assert "management account" in (result.error_message or "")


@pytest.mark.asyncio
async def test_aws_evacuate_iam_users_calls_revoke_for_each() -> None:
    """standalone_iam scope: every Helix-tracked IAM user goes through
    the real multi-step delete chain."""
    driver = _aws_driver()
    with patch.object(
        AwsCloudDriver, "_revoke_sync",
        return_value=["delete_login_profile", "delete_user"],
    ) as mocked:
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="standalone_iam", external_id="123456789012",
            ),
            iam_account_ids=["alice", "bob"],
        )
    assert result.ok is True
    assert result.provider_state == "users_evacuated"
    assert result.revoked_users == 2
    assert mocked.call_count == 2


@pytest.mark.asyncio
async def test_aws_evacuate_no_helix_users_is_clean_zero() -> None:
    driver = _aws_driver()
    result = await driver.close_scope_on_provider(
        CloudScopeRef(scope_type="standalone_iam", external_id="999999999999"),
        iam_account_ids=[],
    )
    assert result.ok is True
    assert result.revoked_users == 0


@pytest.mark.asyncio
async def test_aws_unknown_scope_type_raises_not_implemented() -> None:
    driver = _aws_driver()
    with pytest.raises(NotImplementedError):
        await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="azure_subscription", external_id="x"),
        )


# ── Aliyun aliyun_account → real Resource Directory closure ───────
#
# The contract these lock down (2026-09-01): a CloudAccount member must
# be SWITCHED BACK to a ResourceAccount and really DELETED, never merely
# unlinked from the directory. ``RemoveCloudAccount`` leaves the account
# alive and billable; the pre-2026-09-01 driver called it and reported
# ``ok=True``, so Helix marked the scope closed while the account kept
# running. Several tests below assert ``RemoveCloudAccount`` is never
# called — that absence IS the fix.


def _aliyun_driver() -> AliyunCloudDriver:
    return AliyunCloudDriver(
        integration_id="int-aliyun",
        config={
            "access_key_id": "LTAI-TEST",
            "access_key_secret": "secret",
        },
    )


def _rd_stub(
    *,
    account_type: str,
    status: str = "SwitchSuccess",
    check_status: str = "PreCheckComplete",
    allow_delete: object = True,
    not_allow: list[dict] | None = None,
    abandonable: list[dict] | None = None,
    deletion_type: str = "0",
    check_task_exists: bool = False,
    update_error: str | None = None,
    delete_error: str | None = None,
    switch_lands: bool = True,
    check_start_error: str | None = None,
    deletion_task_status: str | None = None,
):
    """Fake ``_acs3_call`` covering the whole RD close conversation.

    Returns ``(fake, calls)``; ``calls`` is the ordered action list so a
    test can assert both what ran and what did NOT.
    """
    calls: list[str] = []
    state = {"type": account_type, "check_started": check_task_exists}
    sent: dict[str, dict] = {}

    async def fake(**kwargs):
        action = kwargs["action"]
        calls.append(action)
        if action == "GetAccountDeletionStatus":
            if deletion_task_status is None:
                raise RuntimeError(
                    "Aliyun GetAccountDeletionStatus 409: "
                    '{"Code":"EntityNotExists.CheckAccount"}'
                )
            return {"RdAccountDeletionStatus": {
                "Status": deletion_task_status, "DeletionType": 1,
                "DeletionTime": "", "FailReasonList": [],
            }}
        if action == "GetAccount":
            return {"Account": {
                "AccountId": kwargs["params"]["AccountId"],
                "Status": status,
                "Type": state["type"],
                "IdentityInformation": "PingCAP (US), Inc.",
            }}
        if action == "UpdateAccount":
            if update_error:
                raise RuntimeError(update_error)
            if switch_lands:
                state["type"] = kwargs["params"]["NewAccountType"]
            return {}
        if action == "GetAccountDeletionCheckResult":
            if not state["check_started"]:
                raise RuntimeError(
                    "Aliyun GetAccountDeletionCheckResult 409: "
                    '{"Code":"EntityNotExists.CheckAccount",'
                    '"Message":"There is no check task for this account"}',
                )
            info: dict = {"Status": check_status, "AllowDelete": allow_delete}
            if not_allow is not None:
                info["NotAllowReason"] = not_allow
            if abandonable is not None:
                info["AbandonableChecks"] = abandonable
            return {"AccountDeletionCheckResultInfo": info}
        if action == "CheckAccountDelete":
            if check_start_error:
                raise RuntimeError(check_start_error)
            state["check_started"] = True
            return {}
        if action == "DeleteAccount":
            sent["DeleteAccount"] = dict(kwargs.get("params") or {})
            if delete_error:
                raise RuntimeError(delete_error)
            return {"DeletionType": deletion_type}
        raise AssertionError(f"unexpected action {action}")

    fake.sent = sent  # type: ignore[attr-defined]
    return fake, calls


def _delete_params(_calls) -> dict:
    """Params the stub captured for the DeleteAccount call."""
    return _LAST_SENT.get("DeleteAccount", {})


_LAST_SENT: dict = {}


async def _close_aliyun(fake, *, external_id: str = "5995487234252122"):
    """Run the close with the pre-check poll interval collapsed to 0."""
    driver = _aliyun_driver()
    _LAST_SENT.clear()
    with patch.object(AliyunCloudDriver, "_acs3_call", side_effect=fake), \
         patch.object(aliyun_cloud, "_RD_PRECHECK_POLL_INTERVAL_SEC", 0):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="aliyun_account", external_id=external_id,
            ),
        )
    # Captured AFTER the call — the stub fills its dict during it.
    _LAST_SENT.update(getattr(fake, "sent", {}))
    return result


@pytest.mark.asyncio
async def test_aliyun_cloud_account_is_converted_then_really_deleted() -> None:
    """The headline fix: CloudAccount → UpdateAccount → DeleteAccount.

    Regression guard for the prod finding on scope 5995487234252122
    (apac-ai-ventured-hackathon): every one of the tenant's CloudAccount
    members used to take the unlink-only path.
    """
    fake, calls = _rd_stub(account_type="CloudAccount")
    result = await _close_aliyun(fake)

    assert result.ok is True
    assert result.provider_state == "deleted"
    assert result.detail["converted_from_cloud_account"] is True
    assert "UpdateAccount" in calls
    assert "DeleteAccount" in calls
    assert "RemoveCloudAccount" not in calls


@pytest.mark.asyncio
async def test_aliyun_delete_ordering_is_convert_check_then_delete() -> None:
    """DeleteAccount must come after the type switch AND the pre-check."""
    fake, calls = _rd_stub(account_type="CloudAccount")
    await _close_aliyun(fake)

    assert calls.index("UpdateAccount") < calls.index("CheckAccountDelete")
    assert calls.index("CheckAccountDelete") < calls.index("DeleteAccount")


@pytest.mark.asyncio
async def test_aliyun_resource_account_skips_the_type_switch() -> None:
    """A member that is already a ResourceAccount is not touched by
    UpdateAccount — which is also what makes a resumed close idempotent."""
    fake, calls = _rd_stub(account_type="ResourceAccount", status="CreateSuccess")
    result = await _close_aliyun(fake)

    assert result.ok is True
    assert result.detail["converted_from_cloud_account"] is False
    assert "UpdateAccount" not in calls
    assert "DeleteAccount" in calls


@pytest.mark.asyncio
async def test_aliyun_silence_period_is_reported_apart_from_deleted() -> None:
    """``DeletionType='1'`` means Aliyun parks the member in its silence
    period rather than deleting outright — a different fact for audit."""
    fake, _ = _rd_stub(account_type="ResourceAccount", deletion_type="1")
    result = await _close_aliyun(fake)

    assert result.ok is True
    assert result.provider_state == "deletion_silence_period"
    assert result.detail["deletion_type"] == "1"


@pytest.mark.asyncio
async def test_aliyun_switch_pending_is_retriable_not_a_failure() -> None:
    """A 200 on UpdateAccount does not mean ``Type`` has flipped yet.

    Aliyun runs member type changes through a verifying state. Pre-checking
    before it settles gets ``MemberTypeError.DeleteAccount``, which would
    read as a dead end when the truth is "wait a moment" — so the driver
    confirms the new type and reports a retriable state instead.
    """
    fake, calls = _rd_stub(account_type="CloudAccount", switch_lands=False)
    result = await _close_aliyun(fake)

    assert result.ok is False
    assert result.error_code == "cloud_account_switch_pending"
    assert result.detail["settled_type"] == "CloudAccount"
    assert "DeleteAccount" not in calls
    assert "RemoveCloudAccount" not in calls


@pytest.mark.asyncio
async def test_aliyun_member_type_error_right_after_switch_is_retriable() -> None:
    """Same race one step later: the type flipped but the deletion
    subsystem has not picked it up. Retriable, not ``precheck_failed``."""
    fake, calls = _rd_stub(
        account_type="CloudAccount",
        check_start_error=(
            "Aliyun CheckAccountDelete 400: "
            '{"Code":"MemberTypeError.DeleteAccount","Message":"resource '
            'account type only"}'
        ),
    )
    result = await _close_aliyun(fake)

    assert result.ok is False
    assert result.error_code == "cloud_account_switch_pending"
    assert "DeleteAccount" not in calls


@pytest.mark.asyncio
async def test_aliyun_member_type_error_without_a_switch_is_not_retriable() -> None:
    """The retriable reading is earned by having just converted. Without
    that, a member-type rejection is a real dead end and must not be
    dressed up as "click again shortly"."""
    fake, _ = _rd_stub(
        account_type="ResourceAccount",
        check_start_error=(
            "Aliyun CheckAccountDelete 400: "
            '{"Code":"MemberTypeError.DeleteAccount","Message":"invited '
            'member"}'
        ),
    )
    result = await _close_aliyun(fake)

    assert result.ok is False
    assert result.error_code == "deletion_precheck_failed"


@pytest.mark.asyncio
async def test_aliyun_precheck_still_running_does_not_delete() -> None:
    """Aliyun's pre-check takes 3–5 minutes. Catching it mid-flight must
    NOT close anything and must NOT report success."""
    fake, calls = _rd_stub(
        account_type="CloudAccount", check_status="PreChecking",
    )
    result = await _close_aliyun(fake)

    assert result.ok is False
    assert result.error_code == "deletion_precheck_running"
    assert result.detail["converted"] is True
    assert "DeleteAccount" not in calls
    assert "RemoveCloudAccount" not in calls


@pytest.mark.asyncio
async def test_aliyun_precheck_reuses_an_existing_check_task() -> None:
    """A resumed close must attach to the check already in flight.

    Restarting it would re-arm the 3–5 minute wait on every click and the
    close could never finish.
    """
    fake, calls = _rd_stub(account_type="ResourceAccount", check_task_exists=True)
    result = await _close_aliyun(fake)

    assert result.ok is True
    assert "CheckAccountDelete" not in calls


@pytest.mark.asyncio
async def test_aliyun_precheck_refusal_blocks_the_delete() -> None:
    fake, calls = _rd_stub(
        account_type="ResourceAccount",
        allow_delete=False,
        not_allow=[{
            "CheckId": "1",
            "CheckName": "PrePaidResource",
            "Description": "包年包月实例未释放",
        }],
    )
    result = await _close_aliyun(fake)

    assert result.ok is False
    assert result.error_code == "deletion_not_allowed"
    assert "包年包月实例未释放" in (result.error_message or "")
    assert "DeleteAccount" not in calls


@pytest.mark.asyncio
async def test_aliyun_allow_delete_accepts_the_string_form() -> None:
    """Aliyun returns JSON booleans as ``true`` or as ``"true"``."""
    fake, calls = _rd_stub(account_type="ResourceAccount", allow_delete="true")
    result = await _close_aliyun(fake)

    assert result.ok is True
    assert "DeleteAccount" in calls


@pytest.mark.asyncio
async def test_aliyun_foreign_legal_entity_fails_loudly_without_unlinking() -> None:
    """An invited third-party CloudAccount genuinely cannot be closed.

    The old fallback quietly unlinked it and called that a closure. Now
    it must fail, and must NOT reach for RemoveCloudAccount.
    """
    fake, calls = _rd_stub(
        account_type="CloudAccount",
        update_error=(
            "Aliyun UpdateAccount 400: "
            '{"Code":"LegalEntityDifferent","Message":"legal entity differs"}'
        ),
    )
    result = await _close_aliyun(fake)

    assert result.ok is False
    assert result.error_code == "foreign_legal_entity"
    assert "RemoveCloudAccount" not in calls
    assert "DeleteAccount" not in calls


@pytest.mark.asyncio
async def test_aliyun_member_type_error_is_classified() -> None:
    fake, _ = _rd_stub(
        account_type="ResourceAccount",
        delete_error=(
            "Aliyun DeleteAccount 400: "
            '{"Code":"MemberTypeError.DeleteAccount","Message":"resource '
            'account type only"}'
        ),
    )
    result = await _close_aliyun(fake)

    assert result.ok is False
    assert result.error_code == "unclosable_member_type"


@pytest.mark.asyncio
async def test_aliyun_payer_relationship_is_classified() -> None:
    fake, _ = _rd_stub(
        account_type="ResourceAccount",
        delete_error=(
            "Aliyun DeleteAccount 400: "
            '{"Code":"NotSupport.HasPayerAccount","Message":"payer bound"}'
        ),
    )
    result = await _close_aliyun(fake)

    assert result.ok is False
    assert result.error_code == "payer_relationship_active"


@pytest.mark.asyncio
@pytest.mark.parametrize("status", ["SwitchSuccess", "PromoteVerifying",
                                    "RemoveFailed", "DeleteFailed"])
async def test_aliyun_live_member_statuses_are_not_already_closed(
    status: str,
) -> None:
    """None of these means the member is gone.

    ``PromoteVerifying`` is a type conversion in flight and
    ``RemoveFailed`` / ``DeleteFailed`` mean the removal did NOT happen —
    yet the pre-2026-09-01 test (``status in {...} or "Remov" in status
    or "Delete" in status``) matched all three, reporting a live billing
    account as already closed and cascade-disabling its grants having
    called nothing on the provider.
    """
    fake, calls = _rd_stub(account_type="ResourceAccount", status=status)
    result = await _close_aliyun(fake)

    assert result.already_closed is False
    assert "DeleteAccount" in calls


@pytest.mark.asyncio
@pytest.mark.parametrize("status", ["Removed", "RemoveSuccess",
                                    "Deleting", "DeleteSuccess"])
async def test_aliyun_gone_statuses_are_idempotent(status: str) -> None:
    fake, calls = _rd_stub(account_type="ResourceAccount", status=status)
    result = await _close_aliyun(fake)

    assert result.ok is True
    assert result.already_closed is True
    assert "DeleteAccount" not in calls


@pytest.mark.asyncio
async def test_aliyun_close_idempotent_on_not_in_directory() -> None:
    driver = _aliyun_driver()

    async def fake_acs3(**kwargs):
        raise RuntimeError(
            "Aliyun GetAccount 404: AccountId.NotExist account not found",
        )

    with patch.object(AliyunCloudDriver, "_acs3_call", side_effect=fake_acs3):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="aliyun_account", external_id="5333"),
        )
    assert result.ok is True
    assert result.already_closed is True
    assert result.provider_state == "not_in_directory"


@pytest.mark.asyncio
async def test_alicloud_intl_subclass_inherits_the_real_closure() -> None:
    """The 海外 tenant is the one that actually has CloudAccount members.

    ``AlicloudIntlCloudDriver`` overrides BSS / ActionTrail / SSO but not
    the close path, and nothing covered that until this test — the whole
    prod finding lives on an ``alicloud_intl`` scope.
    """
    driver = AlicloudIntlCloudDriver(
        integration_id="int-alicloud-intl",
        config={"access_key_id": "LTAI-TEST", "access_key_secret": "secret"},
    )
    fake, calls = _rd_stub(account_type="CloudAccount")
    with patch.object(AliyunCloudDriver, "_acs3_call", side_effect=fake), \
         patch.object(aliyun_cloud, "_RD_PRECHECK_POLL_INTERVAL_SEC", 0):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="aliyun_account", external_id="5995487234252122",
            ),
        )

    assert result.ok is True
    assert result.provider_state == "deleted"
    assert "UpdateAccount" in calls
    assert "RemoveCloudAccount" not in calls


# ── Azure azure_subscription → cancel REST ─────────────────────────


def _azure_driver() -> AzureCloudDriver:
    return AzureCloudDriver(
        integration_id="int-azure",
        config={
            "tenant_id": "00000000-0000-0000-0000-000000000001",
            "client_id": "client-id",
            "client_secret": "client-secret",
        },
    )


class _AzureClientStub:
    """httpx.AsyncClient async-context-manager stub."""

    def __init__(self, *, get_status: int, post_status: int):
        self.get_status = get_status
        self.post_status = post_status
        self.posts: list[dict] = []

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        return False

    async def get(self, url, params=None, headers=None):
        if self.get_status == 200:
            return _Resp(200, {"state": "Enabled"})
        return _Resp(self.get_status, None)

    async def post(self, url, params=None, headers=None):
        self.posts.append({"url": url, "params": params})
        return _Resp(self.post_status, {})


class _Resp:
    def __init__(self, status_code: int, payload: dict | None) -> None:
        self.status_code = status_code
        self._payload = payload
        self.text = "" if payload is None else "{}"

    def json(self):
        return self._payload or {}


@pytest.mark.asyncio
async def test_azure_subscription_cancel_happy_path() -> None:
    driver = _azure_driver()
    stub = _AzureClientStub(get_status=200, post_status=202)
    with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="bearer")), \
         patch.object(AzureCloudDriver, "_http", return_value=stub):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="azure_subscription",
                external_id="00000000-0000-0000-0000-000000000123",
            ),
        )
    assert result.ok is True
    assert result.provider_state == "cancelled"
    assert any(
        "/cancel" in p["url"] for p in stub.posts
    )


@pytest.mark.asyncio
async def test_azure_subscription_cancel_idempotent_on_404() -> None:
    driver = _azure_driver()
    stub = _AzureClientStub(get_status=404, post_status=404)
    with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="bearer")), \
         patch.object(AzureCloudDriver, "_http", return_value=stub):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="azure_subscription",
                external_id="00000000-0000-0000-0000-000000000404",
            ),
        )
    assert result.ok is True
    assert result.already_closed is True
    assert result.provider_state == "not_found"


@pytest.mark.asyncio
async def test_azure_unknown_scope_type_raises_not_implemented() -> None:
    driver = _azure_driver()
    with pytest.raises(NotImplementedError):
        await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="aws_account", external_id="x"),
        )


# ── GCP gcp_project → projects.delete ─────────────────────────────


def _gcp_driver() -> GcpCloudDriver:
    return GcpCloudDriver(
        integration_id="int-gcp",
        config={
            "service_account_json": "{}",
            "organization_id": "0000",
        },
    )


@pytest.mark.asyncio
async def test_gcp_project_delete_happy_path() -> None:
    driver = _gcp_driver()
    crm = MagicMock()
    crm.projects.return_value.get.return_value.execute.return_value = {
        "state": "ACTIVE",
    }
    crm.projects.return_value.delete.return_value.execute.return_value = {
        "name": "operations/abc",
    }
    with patch.object(GcpCloudDriver, "_svc", return_value=crm):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="gcp_project", external_id="my-project"),
        )
    assert result.ok is True
    assert result.provider_state == "delete_requested"


@pytest.mark.asyncio
async def test_gcp_project_delete_idempotent_when_already_delete_requested() -> None:
    driver = _gcp_driver()
    crm = MagicMock()
    crm.projects.return_value.get.return_value.execute.return_value = {
        "state": "DELETE_REQUESTED",
    }
    with patch.object(GcpCloudDriver, "_svc", return_value=crm):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="gcp_project", external_id="my-project-x"),
        )
    assert result.ok is True
    assert result.already_closed is True
    assert result.provider_state == "delete_requested"
    crm.projects.return_value.delete.assert_not_called()


# ── Tencent / UCloud standalone_iam evacuation ────────────────────


@pytest.mark.asyncio
async def test_tencent_close_standalone_iam_evacuates_users() -> None:
    driver = TencentCloudDriver(
        integration_id="int-tc",
        config={"secret_id": "AKID", "secret_key": "SK", "region": "ap-shanghai"},
    )
    with patch.object(
        TencentCloudDriver, "revoke_account",
        AsyncMock(side_effect=[
            type("PR", (), {"ok": True, "raw_response": {}, "error_message": None})(),
            type("PR", (), {"ok": True, "raw_response": {}, "error_message": None})(),
        ]),
    ):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="standalone_iam", external_id="tc-tenant"),
            iam_account_ids=["leaver-a", "leaver-b"],
        )
    assert result.ok is True
    assert result.revoked_users == 2
    assert result.provider_state == "users_evacuated"


@pytest.mark.asyncio
async def test_tencent_close_partial_failure_surfaces_error() -> None:
    driver = TencentCloudDriver(
        integration_id="int-tc",
        config={"secret_id": "AKID", "secret_key": "SK", "region": "ap-shanghai"},
    )
    pr_cls = type("PR", (), {})
    ok = pr_cls()
    ok.ok = True
    ok.raw_response = {}
    ok.error_message = None
    bad = pr_cls()
    bad.ok = False
    bad.raw_response = None
    bad.error_message = "Tencent DeleteUser: PolicyAttached"

    with patch.object(
        TencentCloudDriver, "revoke_account",
        AsyncMock(side_effect=[ok, bad]),
    ):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="standalone_iam", external_id="tc-tenant"),
            iam_account_ids=["alice", "bob"],
        )
    assert result.ok is False
    assert result.error_code == "evacuation_partial"
    assert result.revoked_users == 1


@pytest.mark.asyncio
async def test_ucloud_close_standalone_iam_evacuates_users() -> None:
    driver = UCloudDriver(
        integration_id="int-uc",
        config={"public_key": "P", "private_key": "S", "project_id": "proj-1"},
    )
    pr_cls = type("PR", (), {})
    ok = pr_cls()
    ok.ok = True
    ok.raw_response = {}
    ok.error_message = None
    with patch.object(
        UCloudDriver, "revoke_account",
        AsyncMock(return_value=ok),
    ):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="standalone_iam", external_id="uc-tenant"),
            iam_account_ids=["leaver"],
        )
    assert result.ok is True
    assert result.revoked_users == 1


@pytest.mark.asyncio
async def test_tencent_unsupported_scope_type_raises() -> None:
    driver = TencentCloudDriver(
        integration_id="int-tc",
        config={"secret_id": "AKID", "secret_key": "SK", "region": "ap-shanghai"},
    )
    with pytest.raises(NotImplementedError):
        await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="aws_account", external_id="x"),
        )


# ── Azure cancel — provider 4xx / 5xx error mapping (Round-52) ────────


class _AzureCancelStub:
    """Stub that lets each test wire a custom cancel-POST status + body
    to exercise the parsed-ARM-error dispatch in close_scope_on_provider.

    Mirrors ``_AzureClientStub`` but with full control over the POST
    response body so the ARM error envelope parser
    (``_parse_arm_error``) sees realistic inputs.
    """

    def __init__(
        self,
        *,
        cancel_status: int,
        cancel_body: dict | None = None,
        cancel_headers: dict | None = None,
    ) -> None:
        self.cancel_status = cancel_status
        self.cancel_body = cancel_body
        self.cancel_headers = cancel_headers or {}

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        return False

    async def get(self, url, params=None, headers=None):
        return _Resp(200, {"state": "Enabled"})

    async def post(self, url, params=None, headers=None):
        return _BodyResp(
            self.cancel_status,
            self.cancel_body,
            headers=self.cancel_headers,
        )


class _BodyResp:
    """``_Resp`` variant that supports an arbitrary JSON body so tests
    can shape the ARM error payload (top-level OR nested ``error``)."""

    def __init__(
        self,
        status_code: int,
        payload: dict | None,
        *,
        headers: dict | None = None,
    ) -> None:
        import json as _json
        self.status_code = status_code
        self._payload = payload
        self.text = "" if payload is None else _json.dumps(payload)
        self.headers = headers or {}

    def json(self):
        if self._payload is None:
            raise ValueError("no body")
        return self._payload


@pytest.mark.asyncio
async def test_azure_cancel_400_resources_existing_maps_to_friendly_message() -> None:
    """The user-reported bug: Azure 400 ResourcesExistingOnSubscription
    used to dump the raw JSON body. Driver now classifies it as
    ``error_code='resources_existing'`` with an actionable remediation
    message AND preserves the ARM code in detail for audit."""
    driver = _azure_driver()
    stub = _AzureCancelStub(
        cancel_status=400,
        cancel_body={
            "error": {
                "code": "ResourcesExistingOnSubscription",
                "message": (
                    "There are existing resources associated with "
                    "subscription. Delete them before cancelling the "
                    "subscription"
                ),
            },
        },
    )
    with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \
         patch.object(AzureCloudDriver, "_http", return_value=stub):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="azure_subscription",
                external_id="00000000-0000-0000-0000-000000000123",
            ),
        )
    assert result.ok is False
    assert result.error_code == "resources_existing"
    assert "still contains resources" in (result.error_message or "")
    assert "az group" in (result.error_message or "")
    assert result.detail.get("arm_code") == "ResourcesExistingOnSubscription"


@pytest.mark.asyncio
async def test_azure_cancel_400_top_level_error_shape() -> None:
    """Some Azure ARM endpoints return the error code/message at the
    top level instead of nested under ``error``. The parser must
    handle both shapes — verifying the exact payload the user pasted."""
    driver = _azure_driver()
    stub = _AzureCancelStub(
        cancel_status=400,
        cancel_body={
            "code": "ResourcesExistingOnSubscription",
            "message": (
                "There are existing resources associated with "
                "subscription. Delete them before cancelling the "
                "subscription"
            ),
        },
    )
    with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \
         patch.object(AzureCloudDriver, "_http", return_value=stub):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="azure_subscription",
                external_id="00000000-0000-0000-0000-000000000124",
            ),
        )
    assert result.ok is False
    assert result.error_code == "resources_existing"


@pytest.mark.asyncio
async def test_azure_cancel_400_not_registered_treated_as_already_closed() -> None:
    """``SubscriptionNotInRegisteredState`` is a race against Azure
    flipping the subscription out from under us — treat as idempotent
    success so admin retries converge."""
    driver = _azure_driver()
    stub = _AzureCancelStub(
        cancel_status=400,
        cancel_body={
            "error": {
                "code": "SubscriptionNotInRegisteredState",
                "message": "Subscription is not in a registered state",
            },
        },
    )
    with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \
         patch.object(AzureCloudDriver, "_http", return_value=stub):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="azure_subscription",
                external_id="00000000-0000-0000-0000-000000000125",
            ),
        )
    assert result.ok is True
    assert result.already_closed is True
    assert result.provider_state == "not_registered"


@pytest.mark.asyncio
async def test_azure_cancel_400_pending_charges_maps_friendly() -> None:
    driver = _azure_driver()
    stub = _AzureCancelStub(
        cancel_status=400,
        cancel_body={
            "error": {
                "code": "PendingCharges",
                "message": "Subscription has outstanding charges",
            },
        },
    )
    with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \
         patch.object(AzureCloudDriver, "_http", return_value=stub):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="azure_subscription",
                external_id="00000000-0000-0000-0000-000000000126",
            ),
        )
    assert result.ok is False
    assert result.error_code == "pending_charges"
    assert "outstanding billing" in (result.error_message or "")


@pytest.mark.asyncio
async def test_azure_cancel_401_maps_to_auth_failed() -> None:
    driver = _azure_driver()
    stub = _AzureCancelStub(cancel_status=401, cancel_body=None)
    with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \
         patch.object(AzureCloudDriver, "_http", return_value=stub):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="azure_subscription",
                external_id="00000000-0000-0000-0000-000000000127",
            ),
        )
    assert result.ok is False
    assert result.error_code == "auth_failed"


@pytest.mark.asyncio
async def test_azure_cancel_429_maps_to_throttled() -> None:
    driver = _azure_driver()
    stub = _AzureCancelStub(
        cancel_status=429, cancel_body=None,
        cancel_headers={"Retry-After": "30"},
    )
    with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \
         patch.object(AzureCloudDriver, "_http", return_value=stub):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="azure_subscription",
                external_id="00000000-0000-0000-0000-000000000128",
            ),
        )
    assert result.ok is False
    assert result.error_code == "throttled"
    assert result.detail.get("retry_after") == "30"


@pytest.mark.asyncio
async def test_azure_cancel_500_maps_to_transient_failure() -> None:
    driver = _azure_driver()
    stub = _AzureCancelStub(cancel_status=503, cancel_body=None)
    with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \
         patch.object(AzureCloudDriver, "_http", return_value=stub):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="azure_subscription",
                external_id="00000000-0000-0000-0000-000000000129",
            ),
        )
    assert result.ok is False
    assert result.error_code == "transient_failure"


@pytest.mark.asyncio
async def test_azure_cancel_400_unknown_code_keeps_generic_with_classified_code() -> None:
    """A 400 with a code we don't recognise should still be dispatched
    to ``cancel_precondition_failed`` (NOT ``cancel_failed``) and carry
    the symbolic code in audit detail so future occurrences are easy
    to grep."""
    driver = _azure_driver()
    stub = _AzureCancelStub(
        cancel_status=400,
        cancel_body={
            "error": {
                "code": "SomeFutureUnknownCode",
                "message": "this is unfamiliar",
            },
        },
    )
    with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \
         patch.object(AzureCloudDriver, "_http", return_value=stub):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="azure_subscription",
                external_id="00000000-0000-0000-0000-000000000130",
            ),
        )
    assert result.ok is False
    assert result.error_code == "cancel_precondition_failed"
    assert "SomeFutureUnknownCode" in (result.error_message or "")
    assert result.detail.get("arm_code") == "SomeFutureUnknownCode"


# ── GCP projects.delete 400 mapping (Round-52) ──────────────────────


def _gcp_http_error(status: int, body: dict) -> Exception:
    """Build a googleapiclient ``HttpError``-ish object that our driver's
    detail-parser can read. We don't import the real class because the
    driver's branch only touches ``.resp.status`` and ``.content``."""
    import json as _json

    class _Resp:
        def __init__(self, s):
            self.status = s
    err = Exception("simulated GCP error")
    err.resp = _Resp(status)  # type: ignore[attr-defined]
    err.content = _json.dumps(body).encode("utf-8")  # type: ignore[attr-defined]
    return err


@pytest.mark.asyncio
async def test_gcp_project_delete_400_lien_maps_friendly() -> None:
    driver = _gcp_driver()
    from app.integrations.drivers import gcp_cloud

    crm = MagicMock()
    crm.projects.return_value.get.return_value.execute.return_value = {
        "state": "ACTIVE",
    }
    lien_err = _gcp_http_error(400, {
        "error": {
            "code": 400,
            "message": "Cannot delete project because it has liens.",
            "details": [
                {"@type": "type.googleapis.com/google.rpc.PreconditionFailure",
                 "violations": [{"type": "LIEN", "subject": "p", "description": "d"}]},
            ],
        },
    })
    crm.projects.return_value.delete.return_value.execute.side_effect = lien_err

    with patch.object(GcpCloudDriver, "_svc", return_value=crm), \
         patch.object(gcp_cloud, "HttpError", Exception):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="gcp_project", external_id="my-project"),
        )
    assert result.ok is False
    assert result.error_code == "liens_exist"
    assert "lien" in (result.error_message or "").lower()


@pytest.mark.asyncio
async def test_gcp_project_delete_400_billing_maps_friendly() -> None:
    driver = _gcp_driver()
    from app.integrations.drivers import gcp_cloud

    crm = MagicMock()
    crm.projects.return_value.get.return_value.execute.return_value = {
        "state": "ACTIVE",
    }
    # New shape (Round-52 follow-up): the dispatch now reads the
    # symbolic ``ErrorInfo.reason`` instead of message free-text. This
    # is the canonical GCP shape for a "billing account still linked"
    # rejection.
    billing_err = _gcp_http_error(400, {
        "error": {
            "code": 400,
            "message": "Project is still attached to a billing account.",
            "details": [
                {"@type": "type.googleapis.com/google.rpc.ErrorInfo",
                 "reason": "BILLING_ACCOUNT_LINKED"},
            ],
        },
    })
    crm.projects.return_value.delete.return_value.execute.side_effect = billing_err

    with patch.object(GcpCloudDriver, "_svc", return_value=crm), \
         patch.object(gcp_cloud, "HttpError", Exception):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="gcp_project", external_id="my-project-y"),
        )
    assert result.ok is False
    assert result.error_code == "billing_attached"


@pytest.mark.asyncio
async def test_gcp_project_delete_lien_message_with_billing_word_still_lien() -> None:
    """Regression: if the GCP message free-text contains the word
    "billing" (e.g. "lien from Cloud Billing prevents delete") but the
    structured ``reason`` says LIEN, the dispatcher must trust the
    enum, not the message substring. Pre-Round-52 this would have
    mis-routed to ``billing_attached``."""
    driver = _gcp_driver()
    from app.integrations.drivers import gcp_cloud

    crm = MagicMock()
    crm.projects.return_value.get.return_value.execute.return_value = {
        "state": "ACTIVE",
    }
    err = _gcp_http_error(400, {
        "error": {
            "code": 400,
            "message": "Cannot delete: lien from Cloud Billing prevents delete.",
            "details": [
                {"@type": "type.googleapis.com/google.rpc.PreconditionFailure",
                 "violations": [{"type": "LIEN", "subject": "p", "description": "d"}]},
            ],
        },
    })
    crm.projects.return_value.delete.return_value.execute.side_effect = err

    with patch.object(GcpCloudDriver, "_svc", return_value=crm), \
         patch.object(gcp_cloud, "HttpError", Exception):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="gcp_project", external_id="my-project-z"),
        )
    assert result.ok is False
    assert result.error_code == "liens_exist"


@pytest.mark.asyncio
async def test_gcp_project_delete_412_precondition_maps_to_400_path() -> None:
    """GCP returns 412 for a subset of precondition failures (etag
    mismatches, some lien states). Driver treats 412 the same as 400."""
    driver = _gcp_driver()
    from app.integrations.drivers import gcp_cloud

    crm = MagicMock()
    crm.projects.return_value.get.return_value.execute.return_value = {
        "state": "ACTIVE",
    }
    err = _gcp_http_error(412, {
        "error": {
            "code": 412,
            "message": "Precondition failed.",
            "details": [
                {"@type": "type.googleapis.com/google.rpc.PreconditionFailure",
                 "violations": [{"type": "LIEN", "subject": "p", "description": "d"}]},
            ],
        },
    })
    crm.projects.return_value.delete.return_value.execute.side_effect = err

    with patch.object(GcpCloudDriver, "_svc", return_value=crm), \
         patch.object(gcp_cloud, "HttpError", Exception):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="gcp_project", external_id="my-project-412"),
        )
    assert result.ok is False
    assert result.error_code == "liens_exist"


@pytest.mark.asyncio
async def test_gcp_project_delete_429_maps_to_throttled() -> None:
    driver = _gcp_driver()
    from app.integrations.drivers import gcp_cloud

    crm = MagicMock()
    crm.projects.return_value.get.return_value.execute.return_value = {
        "state": "ACTIVE",
    }
    err = _gcp_http_error(429, {"error": {"code": 429, "message": "rate limited"}})
    crm.projects.return_value.delete.return_value.execute.side_effect = err

    with patch.object(GcpCloudDriver, "_svc", return_value=crm), \
         patch.object(gcp_cloud, "HttpError", Exception):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="gcp_project", external_id="my-project-throttled"),
        )
    assert result.ok is False
    assert result.error_code == "throttled"


@pytest.mark.asyncio
async def test_gcp_project_delete_5xx_maps_to_transient() -> None:
    driver = _gcp_driver()
    from app.integrations.drivers import gcp_cloud

    crm = MagicMock()
    crm.projects.return_value.get.return_value.execute.return_value = {
        "state": "ACTIVE",
    }
    err = _gcp_http_error(503, {"error": {"code": 503, "message": "service unavailable"}})
    crm.projects.return_value.delete.return_value.execute.side_effect = err

    with patch.object(GcpCloudDriver, "_svc", return_value=crm), \
         patch.object(gcp_cloud, "HttpError", Exception):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="gcp_project", external_id="my-project-503"),
        )
    assert result.ok is False
    assert result.error_code == "transient_failure"


# ── Azure additional code coverage (Round-52 scan follow-up) ───────


@pytest.mark.asyncio
async def test_azure_cancel_400_subscription_must_have_owner() -> None:
    driver = _azure_driver()
    stub = _AzureCancelStub(
        cancel_status=400,
        cancel_body={
            "error": {
                "code": "SubscriptionMustHaveOwner",
                "message": "Subscription must have at least one Owner",
            },
        },
    )
    with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \
         patch.object(AzureCloudDriver, "_http", return_value=stub):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="azure_subscription",
                external_id="00000000-0000-0000-0000-000000000131",
            ),
        )
    assert result.ok is False
    assert result.error_code == "missing_owner"
    assert "Owner" in (result.error_message or "")


@pytest.mark.asyncio
async def test_azure_cancel_400_billing_profile_not_active() -> None:
    driver = _azure_driver()
    stub = _AzureCancelStub(
        cancel_status=400,
        cancel_body={
            "error": {
                "code": "BillingProfileNotActive",
                "message": "Billing profile is not active.",
            },
        },
    )
    with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \
         patch.object(AzureCloudDriver, "_http", return_value=stub):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="azure_subscription",
                external_id="00000000-0000-0000-0000-000000000132",
            ),
        )
    assert result.ok is False
    assert result.error_code == "pending_charges"


@pytest.mark.asyncio
async def test_azure_cancel_400_outstanding_charges() -> None:
    driver = _azure_driver()
    stub = _AzureCancelStub(
        cancel_status=400,
        cancel_body={
            "error": {
                "code": "OutstandingCharges",
                "message": "Subscription has outstanding charges",
            },
        },
    )
    with patch.object(AzureCloudDriver, "_token", AsyncMock(return_value="b")), \
         patch.object(AzureCloudDriver, "_http", return_value=stub):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(
                scope_type="azure_subscription",
                external_id="00000000-0000-0000-0000-000000000133",
            ),
        )
    assert result.ok is False
    assert result.error_code == "pending_charges"


# ── Abandoning the pre-check items is what makes the close finish ──
#
# Prod, 2026-09-01: account 5995487234252122 was submitted with 16
# ``AbandonableChecks`` and NO ``AbandonableCheckId``. Aliyun accepted
# the DeleteAccount, then killed it at ``CheckFailed`` on exactly those
# 16 items. Withholding them did not protect anything — ``DeleteAccount``
# destroys every resource in the member account anyway — it just made
# the closure unfinishable and pushed the work back onto the admin.


@pytest.mark.asyncio
async def test_aliyun_abandons_every_abandonable_check() -> None:
    fake, calls = _rd_stub(
        account_type="ResourceAccount",
        abandonable=[
            {"CheckId": "SP_oss_oss_intl", "CheckName": "oss_intl",
             "Description": "instance number:1"},
            {"CheckId": "SP_pai_pai_intl", "Description": "instance number:1"},
        ],
    )
    result = await _close_aliyun(fake)

    assert result.ok is True
    sent = _delete_params(calls)
    assert sent["AbandonableCheckId"] == ["SP_oss_oss_intl", "SP_pai_pai_intl"]
    # And it is on the record what was waved through on the admin's behalf.
    assert result.detail["abandoned_check_ids"] == [
        "SP_oss_oss_intl", "SP_pai_pai_intl",
    ]
    assert any("oss_intl" in x for x in result.detail["abandoned_checks"])


@pytest.mark.asyncio
async def test_aliyun_sends_no_abandon_param_when_there_is_nothing_to_abandon() -> None:
    fake, calls = _rd_stub(account_type="ResourceAccount")
    result = await _close_aliyun(fake)

    assert result.ok is True
    assert "AbandonableCheckId" not in _delete_params(calls)
    assert result.detail["abandoned_check_ids"] == []


@pytest.mark.asyncio
async def test_aliyun_skips_abandonable_entries_with_no_check_id() -> None:
    """An entry Aliyun returns without a ``CheckId`` cannot be abandoned.
    Inventing one would fail the whole DeleteAccount on a malformed
    parameter instead of leaving that single item unabandoned."""
    fake, calls = _rd_stub(
        account_type="ResourceAccount",
        abandonable=[
            {"CheckId": "SP_fc_fc_intl", "Description": "instance number:1"},
            {"Description": "instance number:3"},
        ],
    )
    await _close_aliyun(fake)

    assert _delete_params(calls)["AbandonableCheckId"] == ["SP_fc_fc_intl"]


@pytest.mark.asyncio
async def test_acs3_sends_list_params_as_a_json_array() -> None:
    """``AbandonableCheckId`` must reach the wire as a JSON array.

    Measured against prod, 2026-09-02. This assertion previously demanded
    the classic Aliyun RPC repeated form (``AbandonableCheckId.1`` /
    ``.2``) and passed — while Aliyun silently ignored the parameter and
    killed the deletion at ``CheckFailed`` on the very items being
    abandoned. Re-sent as a JSON array (the form the Aliyun CLI uses),
    the same account went ``Deleting`` with an empty ``FailReasonList``.
    **A green test here proves the encoder does what it was told, not
    that the provider accepts it** — this one is pinned to an observed
    provider response, not to a convention.
    """
    captured: dict[str, str] = {}

    class _Resp:
        status_code = 200
        text = "{}"

        def json(self):
            return {}

    class _Client:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *exc):
            return False

        async def request(self, method, url, headers=None, content=None):
            captured["url"] = url
            return _Resp()

    driver = _aliyun_driver()
    with patch.object(aliyun_cloud.httpx, "AsyncClient", lambda **kw: _Client()):
        await driver._acs3_call(
            host="resourcemanager.aliyuncs.com", version="2020-03-31",
            action="DeleteAccount",
            params={"AccountId": "5111", "AbandonableCheckId": ["A", "B"]},
        )
    from urllib.parse import parse_qs, urlparse
    q = parse_qs(urlparse(captured["url"]).query)
    assert q["AbandonableCheckId"] == ['["A","B"]']
    assert not any(k.startswith("AbandonableCheckId.") for k in q)


# ── Resuming after a deletion that died ───────────────────────────

@pytest.mark.asyncio
async def test_aliyun_dead_deletion_forces_a_fresh_check() -> None:
    """``GetAccount`` cannot tell you a deletion died — prod's member sat
    at ``SwitchSuccess`` while its deletion was ``CheckFailed``. Ask the
    deletion task, and start a NEW check: re-using the dead one's result
    reproduces the failure exactly."""
    fake, calls = _rd_stub(
        account_type="ResourceAccount",
        deletion_task_status="CheckFailed",
        check_task_exists=True,
    )
    result = await _close_aliyun(fake)

    assert result.ok is True
    # A fresh check was started even though a completed one existed.
    assert "CheckAccountDelete" in calls
    assert "DeleteAccount" in calls


@pytest.mark.asyncio
@pytest.mark.parametrize("in_flight", ["Checking", "Deleting"])
async def test_aliyun_deletion_already_running_is_not_resubmitted(
    in_flight: str,
) -> None:
    fake, calls = _rd_stub(
        account_type="ResourceAccount", deletion_task_status=in_flight,
    )
    result = await _close_aliyun(fake)

    assert result.ok is False
    assert result.error_code == "deletion_precheck_running"
    assert "DeleteAccount" not in calls


@pytest.mark.asyncio
async def test_aliyun_already_deleted_is_idempotent() -> None:
    fake, calls = _rd_stub(
        account_type="ResourceAccount", deletion_task_status="Success",
    )
    result = await _close_aliyun(fake)

    assert result.ok is True
    assert result.already_closed is True
    assert result.provider_state == "deleted"
    assert "DeleteAccount" not in calls


# ── Aliyun DeleteAccount error classification (Round-52 follow-up) ──
#
# These anchor on Aliyun's symbolic error codes rather than loose
# substrings. Each drives the full close conversation and fails only at
# ``DeleteAccount``, so they also prove the pre-check stage does not
# swallow or reshape a provider rejection.


@pytest.mark.asyncio
async def test_aliyun_delete_account_unsettled_bill_anchored_code() -> None:
    """Anchored on the real Aliyun symbolic code
    ``Member.NotAllowedDelete.HasUnsettledBill`` (not the loose ``"Bill"``
    substring the pre-Round-52 form used)."""
    fake, _ = _rd_stub(
        account_type="ResourceAccount",
        delete_error=(
            "Aliyun resourcemanager 400 "
            "Member.NotAllowedDelete.HasUnsettledBill: settle bill first"
        ),
    )
    result = await _close_aliyun(fake, external_id="5335")
    assert result.ok is False
    assert result.error_code == "pending_charges"


@pytest.mark.asyncio
async def test_aliyun_delete_account_unrelated_bill_word_does_not_misroute() -> None:
    """Regression: pre-Round-52 the matcher was ``"Bill" in msg`` and
    would have mis-routed ANY error mentioning "Bill" anywhere. Now
    the matcher only fires on anchored symbolic codes — a free-text
    message containing the substring "Bill" without an Aliyun
    billing code must fall through to the generic ``delete_failed``."""
    fake, _ = _rd_stub(
        account_type="ResourceAccount",
        delete_error=(
            "Aliyun resourcemanager 400 SomeUnknownCode: "
            "Cannot fulfill Billable operation"
        ),
    )
    result = await _close_aliyun(fake, external_id="5336")
    assert result.ok is False
    # Falls through to generic — NOT pending_charges.
    assert result.error_code == "delete_failed"


@pytest.mark.asyncio
async def test_aliyun_delete_account_resources_existing_maps_friendly() -> None:
    fake, _ = _rd_stub(
        account_type="ResourceAccount",
        delete_error=(
            "Aliyun resourcemanager 400 "
            "Member.NotAllowedDelete.HasResource: release resources first"
        ),
    )
    result = await _close_aliyun(fake, external_id="5337")
    assert result.ok is False
    assert result.error_code == "resources_existing"


@pytest.mark.asyncio
async def test_aliyun_delete_account_real_name_auth_maps_friendly() -> None:
    fake, _ = _rd_stub(
        account_type="ResourceAccount",
        delete_error=(
            "Aliyun resourcemanager 400 "
            "Member.NotAllowedToDoAction.NotPassRealNameAuth: verify first"
        ),
    )
    result = await _close_aliyun(fake, external_id="5338")
    assert result.ok is False
    assert result.error_code == "not_real_name_authed"


@pytest.mark.asyncio
async def test_aliyun_precheck_provider_error_is_a_result_not_an_exception() -> None:
    """A provider failure inside the pre-check must come back as a
    CloseScopeResult; a bare RuntimeError would reach the service as an
    opaque 500 instead of an actionable 502."""
    driver = _aliyun_driver()

    async def fake(**kwargs):
        if kwargs["action"] == "GetAccount":
            return {"Account": {"Type": "ResourceAccount",
                                "Status": "CreateSuccess"}}
        raise RuntimeError("Aliyun CheckAccountDelete 500: internal error")

    with patch.object(AliyunCloudDriver, "_acs3_call", side_effect=fake), \
         patch.object(aliyun_cloud, "_RD_PRECHECK_POLL_INTERVAL_SEC", 0):
        result = await driver.close_scope_on_provider(
            CloudScopeRef(scope_type="aliyun_account", external_id="5339"),
        )
    assert result.ok is False
    assert result.error_code == "deletion_precheck_failed"