"""Unit tests for the Aliyun cloud driver — lifecycle methods.

The AliyunCloudDriver uses ACS3-HMAC-SHA256 signing and hits several
product hosts (resourcemanager, ram, business). All HTTP mocked via
respx. These tests cover the methods added in this PR:

  * list_iam_users (tenant-wide RAM enumeration)
  * revoke_account (delete-chain for offboarding kill switch)

Grant / revoke / billing already covered by integration-level tests.
"""
from __future__ import annotations

import urllib.parse
from datetime import date
from unittest.mock import AsyncMock, patch

import httpx
import pytest
import respx

from app.integrations.driver_base import (
    CloudAccessContext,
    CloudAccountRevokeContext,
    CloudScopeRef,
)
from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver

_CFG = {
    "access_key_id": "LTAI-test-ak",
    "access_key_secret": "test-sk",
    "account_mode": "sso",
    "region": "cn-shanghai",
}


def _driver() -> AliyunCloudDriver:
    return AliyunCloudDriver(integration_id="aliyun-test", config=_CFG)


def _ram_response(action: str, payload: dict) -> httpx.Response:
    return httpx.Response(200, json=payload)


# ── list_iam_users ───────────────────────────────────────────────


@pytest.mark.asyncio
@respx.mock
async def test_list_iam_users_paginates() -> None:
    # First page: truncated, second page: final.
    page1 = {
        "Users": {"User": [
            {"UserId": "1", "UserName": "alice", "DisplayName": "Alice",
             "Email": "alice@example.com", "CreateDate": "2025-01-01T00:00:00Z"},
        ]},
        "IsTruncated": True,
        "Marker": "next",
    }
    page2 = {
        "Users": {"User": [
            {"UserId": "2", "UserName": "bob", "DisplayName": "Bob",
             "Email": "bob@example.com"},
        ]},
        "IsTruncated": False,
    }
    seq = {"i": 0}

    def _route(request: httpx.Request) -> httpx.Response:
        if request.headers.get("x-acs-action") != "ListUsers":
            return httpx.Response(500, text="unexpected action")
        resp = page1 if seq["i"] == 0 else page2
        seq["i"] += 1
        return httpx.Response(200, json=resp)

    respx.mock.post(
        url__regex=r"^https://ram\.aliyuncs\.com/.*",
    ).mock(side_effect=_route)

    users = await _driver().list_iam_users()
    assert [u["user_name"] for u in users] == ["alice", "bob"]
    assert users[0]["email"] == "alice@example.com"


# ── revoke_account (happy path: full delete chain) ───────────────


@pytest.mark.asyncio
@respx.mock
async def test_revoke_account_delete_chain() -> None:
    # Build a scripted sequence keyed on x-acs-action header.
    calls: list[str] = []

    def _route(request: httpx.Request) -> httpx.Response:
        action = request.headers.get("x-acs-action", "")
        calls.append(action)
        if action == "ListPoliciesForUser":
            return httpx.Response(200, json={
                "Policies": {"Policy": [
                    {"PolicyName": "AdministratorAccess", "PolicyType": "System"},
                ]},
            })
        if action == "ListAccessKeys":
            return httpx.Response(200, json={
                "AccessKeys": {"AccessKey": [
                    {"AccessKeyId": "AK1"},
                ]},
            })
        if action == "ListGroupsForUser":
            return httpx.Response(200, json={
                "Groups": {"Group": [{"GroupName": "dev"}]},
            })
        # Everything else returns 200 with empty payload (detach, delete_ak,
        # delete_login_profile, unbind_mfa, remove_from_group, delete_user).
        return httpx.Response(200, json={})

    respx.mock.post(url__regex=r"^https://ram\.aliyuncs\.com/.*").mock(
        side_effect=_route,
    )

    result = await _driver().revoke_account(CloudAccountRevokeContext(
        provider="aliyun", account_id="alice@example.com",
    ))
    assert result.ok, result.error_message
    # The literal handle wins when it exists. This assertion used to read
    # "alice" — deriving the local part unconditionally is exactly the bug
    # `account_id_candidates` was introduced to fix: on the Tencent tenant 17
    # of 22 live users have `@` in the name, so the revoke looked up a user
    # that never existed and reported "already absent" while the real one kept
    # AdministratorAccess. Here ListPoliciesForUser answers for the address,
    # so the address IS the user and it must be the one deleted.
    assert result.external_ref == "alice@example.com"
    # The delete chain must have visited DeleteUser after the prerequisites.
    assert "DeleteUser" in calls
    assert calls.index("DeleteUser") > calls.index("DetachPolicyFromUser")


@pytest.mark.asyncio
@respx.mock
async def test_revoke_account_falls_back_to_local_part_when_address_absent() -> None:
    """The other half of the candidate contract.

    Tenants whose RAM users genuinely are named after the local part
    (``li.da``, ``xiaowu.wang``) must keep working — so a literal handle that
    does not exist has to fall through to the derived one rather than
    reporting the user absent. Without this case the test above would pass on
    an implementation that only ever tries the literal value.
    """
    probed: list[str] = []

    def _route(request: httpx.Request) -> httpx.Response:
        if request.headers.get("x-acs-action", "") == "ListPoliciesForUser":
            user = request.url.params.get("UserName", "")
            probed.append(user)
            if user == "alice@example.com":
                return httpx.Response(
                    404,
                    text='{"Code": "EntityNotExist.User", "Message": "no such user"}',
                )
            return httpx.Response(200, json={"Policies": {"Policy": []}})
        return httpx.Response(200, json={})

    respx.mock.post(url__regex=r"^https://ram\.aliyuncs\.com/.*").mock(
        side_effect=_route,
    )

    result = await _driver().revoke_account(CloudAccountRevokeContext(
        provider="aliyun", account_id="alice@example.com",
    ))
    assert result.ok, result.error_message
    assert result.external_ref == "alice"
    # Literal first, derived second — the order is the contract, not just the
    # set: probing the derived name first is what caused the prod no-op.
    assert probed == ["alice@example.com", "alice"]


@pytest.mark.asyncio
@respx.mock
async def test_revoke_account_already_absent_user() -> None:
    # First call (ListPoliciesForUser) returns EntityNotExist.User → we
    # treat this as idempotent success without walking the chain.
    def _route(request: httpx.Request) -> httpx.Response:
        if request.headers.get("x-acs-action") == "ListPoliciesForUser":
            return httpx.Response(
                404,
                text='{"Code": "EntityNotExist.User", "Message": "user not found"}',
            )
        return httpx.Response(500, text="unexpected call to " + request.headers.get("x-acs-action", ""))

    respx.mock.post(url__regex=r"^https://ram\.aliyuncs\.com/.*").mock(
        side_effect=_route,
    )
    result = await _driver().revoke_account(CloudAccountRevokeContext(
        provider="aliyun", account_id="ghost@example.com",
    ))
    assert result.ok
    assert "already absent" in (result.raw_response or {}).get("note", "")


def test_revoke_account_requires_account_id() -> None:
    import asyncio
    result = asyncio.run(_driver().revoke_account(
        CloudAccountRevokeContext(provider="aliyun", account_id=""),
    ))
    assert not result.ok
    assert "account_id" in (result.error_message or "")


# ── capability metadata ──────────────────────────────────────────


def test_user_revoke_in_capabilities() -> None:
    meta = AliyunCloudDriver.metadata()
    assert "user_revoke" in meta.capabilities


# ── list_billing_daily — NextToken pagination ───────────────────


@pytest.mark.asyncio
@respx.mock
async def test_list_billing_daily_follows_next_token() -> None:
    """``DescribeInstanceBill`` response pages via ``NextToken``. Driver
    must loop until the cursor is empty or busy accounts silently lose
    services past the first 300 rows.

    Round-34: switched from ``QueryAccountBill`` to ``DescribeInstanceBill``
    because the former silently undercounts on RD Reseller masters.
    """
    call_count = {"n": 0}
    acct_id = "5962436752113096"

    def _route(request: httpx.Request) -> httpx.Response:
        action = request.headers.get("x-acs-action", "")
        if action != "DescribeInstanceBill":
            return httpx.Response(500, text=f"unexpected action={action}")
        call_count["n"] += 1
        # First call: one product + NextToken pointing at page 2.
        # ``DescribeInstanceBill`` returns ``Items`` as a flat list
        # (unlike ``QueryAccountBill`` which wraps in ``{"Item": [...]}``).
        if call_count["n"] == 1:
            return httpx.Response(200, json={
                "Data": {
                    "Items": [
                        {
                            "OwnerID": acct_id,
                            "ProductName": "ECS",
                            "PretaxAmount": 100.0,
                        },
                    ],
                    "NextToken": "cursor-2",
                },
            })
        return httpx.Response(200, json={
            "Data": {
                "Items": [
                    {
                        "OwnerID": acct_id,
                        "ProductName": "OSS",
                        "PretaxAmount": 42.5,
                    },
                ],
            },
        })

    respx.mock.post(
        url__regex=r"^https://business\.aliyuncs\.com/.*",
    ).mock(side_effect=_route)

    scope = CloudScopeRef(scope_type="aliyun_account", external_id=acct_id)
    rows = await _driver().list_billing_daily(
        date(2026, 4, 1), date(2026, 4, 1), scope=scope,
    )
    assert call_count["n"] == 2
    assert len(rows) == 1
    row = rows[0]
    assert row.currency == "CNY"
    assert row.total_cost_usd == pytest.approx(142.5)
    assert row.cost_by_service["ECS"] == pytest.approx(100.0)
    assert row.cost_by_service["OSS"] == pytest.approx(42.5)


# ── list_billing_daily — RD member filtering via client-side OwnerID ───


@pytest.mark.asyncio
@respx.mock
async def test_list_billing_daily_uses_describe_instance_bill() -> None:
    """Round-34 pin: ``list_billing_daily`` MUST call
    ``DescribeInstanceBill`` (the line-item endpoint), not
    ``QueryAccountBill`` (the per-account-summary endpoint).

    Prod evidence (alicloud-intl-primary, 2026-05-18 audit): the
    integration runs in RD **Reseller** mode (master pays for every
    sub-account's resources). ``QueryAccountBill BillOwnerId=<sub>``
    returns only the sub's intra-account residual fees (~$7/day) —
    every ECS / EBS / OSS / SLB line item is INVISIBLE because master
    paid it. ``DescribeInstanceBill`` returns the full RD-consolidated
    line-item bill with each row carrying ``BillAccountID`` (master,
    payer) and ``OwnerID`` (sub, resource owner). Round-33 patched
    ``BillOwnerId`` on the wrong API; Round-34 swaps to the right API.

    When called with a sub-account scope, the driver must:
      1. Hit ``DescribeInstanceBill``, not ``QueryAccountBill``.
      2. Filter the returned items client-side by ``OwnerID`` to keep
         only the requested sub's rows.
    """
    seen_action: dict[str, str] = {}

    def _route(request: httpx.Request) -> httpx.Response:
        action = request.headers.get("x-acs-action", "")
        seen_action["last"] = action
        if action != "DescribeInstanceBill":
            return httpx.Response(500, text=f"unexpected action={action}")
        return httpx.Response(200, json={
            "Data": {
                "Items": [
                    # Target sub's rows.
                    {
                        "OwnerID": "5901860086223871",
                        "BillAccountID": "5962436752113096",
                        "ProductName": "ECS",
                        "PretaxAmount": 100.0,
                    },
                    {
                        "OwnerID": "5901860086223871",
                        "BillAccountID": "5962436752113096",
                        "ProductName": "OSS",
                        "PretaxAmount": 200.0,
                    },
                    # Different sub — must be filtered out.
                    {
                        "OwnerID": "5108739859105002",
                        "BillAccountID": "5962436752113096",
                        "ProductName": "ECS",
                        "PretaxAmount": 999.0,
                    },
                ],
            },
        })

    respx.mock.post(
        url__regex=r"^https://business\.aliyuncs\.com/.*",
    ).mock(side_effect=_route)

    scope = CloudScopeRef(
        scope_type="aliyun_account",
        external_id="5901860086223871",
    )
    rows = await _driver().list_billing_daily(
        date(2026, 4, 1), date(2026, 4, 1), scope=scope,
    )
    assert seen_action.get("last") == "DescribeInstanceBill", (
        f"list_billing_daily must call DescribeInstanceBill (line-item "
        f"endpoint that includes resource owner). Pre-Round-34 it called "
        f"QueryAccountBill which on an RD Reseller master returns only "
        f"the sub's intra-account residual fees. saw={seen_action!r}"
    )
    assert len(rows) == 1
    assert rows[0].total_cost_usd == pytest.approx(300.0)  # 100 + 200, NOT 1299
    assert rows[0].cost_by_service["ECS"] == pytest.approx(100.0)
    assert rows[0].cost_by_service["OSS"] == pytest.approx(200.0)
    # 5108739859105002 row was correctly filtered.
    assert "5108739859105002" not in str(rows[0].cost_by_service)


# ── list_scopes — STS-resolved external_id (no UUID fallback) ───


@pytest.mark.asyncio
@respx.mock
async def test_iam_mode_scope_uses_sts_account_id() -> None:
    """In IAM mode, list_scopes must call STS GetCallerIdentity and
    use the returned AccountId as ``external_id`` — not the integration
    UUID. Mirrors the AWS-driver fix for production incident
    2026-04-23 (ghost cloud_scope rows keyed by integration UUID)."""
    from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver

    cfg = dict(_CFG, account_mode="iam")
    d = AliyunCloudDriver(integration_id="aliyun-test", config=cfg)

    respx.mock.post(
        url__regex=r"^https://sts\.aliyuncs\.com/.*",
    ).mock(return_value=httpx.Response(200, json={
        "AccountId": "1234567890123456",
        "Arn": "acs:ram::1234567890123456:root",
        "UserId": "1234567890123456",
    }))

    scopes = await d.list_scopes()
    assert len(scopes) == 1
    assert scopes[0].external_id == "1234567890123456"
    assert scopes[0].scope_type == "standalone_iam"
    assert scopes[0].raw["aliyun_account_id"] == "1234567890123456"


@pytest.mark.asyncio
@respx.mock
async def test_iam_mode_raises_when_sts_fails() -> None:
    """STS unreachable → ProviderAuthError (no UUID fallback).
    Sync job fails loudly; no scope row is written."""
    from app.domains.cmdb.providers.exceptions import ProviderAuthError
    from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver

    cfg = dict(_CFG, account_mode="iam")
    d = AliyunCloudDriver(integration_id="aliyun-test", config=cfg)

    respx.mock.post(
        url__regex=r"^https://sts\.aliyuncs\.com/.*",
    ).mock(return_value=httpx.Response(403, text="InvalidAccessKeyId.NotFound"))

    with pytest.raises(ProviderAuthError, match="GetCallerIdentity failed"):
        await d.list_scopes()


@pytest.mark.asyncio
@respx.mock
async def test_iam_mode_rejects_non_canonical_account_id() -> None:
    """STS returning a non-numeric AccountId is rejected (could
    happen if the upstream API regresses)."""
    from app.domains.cmdb.providers.exceptions import ProviderAuthError
    from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver

    cfg = dict(_CFG, account_mode="iam")
    d = AliyunCloudDriver(integration_id="aliyun-test", config=cfg)

    respx.mock.post(
        url__regex=r"^https://sts\.aliyuncs\.com/.*",
    ).mock(return_value=httpx.Response(200, json={
        "AccountId": "not-a-number",
    }))

    with pytest.raises(ProviderAuthError, match="non-canonical"):
        await d.list_scopes()


@pytest.mark.asyncio
@respx.mock
async def test_no_rd_falls_through_to_sts_account_id() -> None:
    """When ResourceDirectory ListAccounts fails (single-account
    tenant), the no-rd branch must resolve the real account id via
    STS rather than substituting the integration UUID."""
    from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver

    d = AliyunCloudDriver(integration_id="aliyun-test", config=_CFG)

    respx.mock.post(
        url__regex=r"^https://resourcemanager\.aliyuncs\.com/.*",
    ).mock(return_value=httpx.Response(
        400, text='{"Code":"ResourceDirectoryNotInUse","Message":"no rd"}',
    ))
    respx.mock.post(
        url__regex=r"^https://sts\.aliyuncs\.com/.*",
    ).mock(return_value=httpx.Response(200, json={
        "AccountId": "5101000000000000",
    }))

    scopes = await d.list_scopes()
    assert len(scopes) == 1
    assert scopes[0].external_id == "5101000000000000"
    assert scopes[0].raw["note"] == "no-rd"
    assert scopes[0].raw["aliyun_account_id"] == "5101000000000000"


@pytest.mark.parametrize(
    "error",
    [
        "Aliyun ListAccounts 403: Forbidden.RAM",
        "Aliyun ListAccounts 429: Throttling.User",
        "Aliyun ListAccounts 503: ServiceUnavailable",
        "Aliyun ListAccounts transport timeout",
    ],
)
@pytest.mark.asyncio
async def test_list_scopes_does_not_misclassify_first_page_failure_as_no_rd(
    error: str,
) -> None:
    driver = _driver()
    driver._own_account_id = AsyncMock(return_value="5101000000000000")  # type: ignore[method-assign]
    driver._acs3_call = AsyncMock(side_effect=RuntimeError(error))  # type: ignore[method-assign]

    with pytest.raises(RuntimeError, match=error):
        await driver.list_scopes()


@pytest.mark.asyncio
async def test_list_scopes_never_falls_back_after_a_successful_first_page() -> None:
    driver = _driver()
    driver._own_account_id = AsyncMock(return_value="5101000000000000")  # type: ignore[method-assign]
    driver._acs3_call = AsyncMock(  # type: ignore[method-assign]
        side_effect=[
            {
                "Accounts": {
                    "Account": [
                        {
                            "AccountId": f"510100000000{i:04d}",
                            "Status": "CreateSuccess",
                        }
                        for i in range(100)
                    ],
                },
                "TotalCount": 101,
            },
            RuntimeError("Aliyun ListAccounts 400: ResourceDirectoryNotInUse"),
        ],
    )

    with pytest.raises(RuntimeError, match="ResourceDirectoryNotInUse"):
        await driver.list_scopes()


@pytest.mark.asyncio
async def test_list_scopes_rejects_repeated_account_across_pages() -> None:
    driver = _driver()
    driver._own_account_id = AsyncMock(return_value="5101999999999999")  # type: ignore[method-assign]
    first_page = [
        {"AccountId": f"510100000000{i:04d}", "Status": "CreateSuccess"}
        for i in range(100)
    ]
    driver._acs3_call = AsyncMock(side_effect=[  # type: ignore[method-assign]
        {"Accounts": {"Account": first_page}, "TotalCount": 101},
        {"Accounts": {"Account": [first_page[0]]}, "TotalCount": 101},
    ])

    with pytest.raises(RuntimeError, match="repeated AccountId"):
        await driver.list_scopes()


@pytest.mark.asyncio
async def test_list_scopes_rejects_empty_page_before_total() -> None:
    driver = _driver()
    driver._own_account_id = AsyncMock(return_value="5101999999999999")  # type: ignore[method-assign]
    first_page = [
        {"AccountId": f"510100000000{i:04d}", "Status": "CreateSuccess"}
        for i in range(100)
    ]
    driver._acs3_call = AsyncMock(side_effect=[  # type: ignore[method-assign]
        {"Accounts": {"Account": first_page}, "TotalCount": 101},
        {"Accounts": {"Account": []}, "TotalCount": 101},
    ])

    with pytest.raises(RuntimeError, match="empty page"):
        await driver.list_scopes()


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("total", "last_page_size"),
    [(150, 1), (101, 2)],
)
async def test_list_scopes_rejects_final_page_total_mismatch(
    total: int,
    last_page_size: int,
) -> None:
    """The final page must make cumulative unique rows equal TotalCount."""
    driver = _driver()
    driver._own_account_id = AsyncMock(  # type: ignore[method-assign]
        return_value="5101999999999999",
    )
    first_page = [
        {"AccountId": f"510100000000{i:04d}", "Status": "CreateSuccess"}
        for i in range(100)
    ]
    last_page = [
        {"AccountId": f"510100000001{i:04d}", "Status": "CreateSuccess"}
        for i in range(last_page_size)
    ]
    driver._acs3_call = AsyncMock(side_effect=[  # type: ignore[method-assign]
        {"Accounts": {"Account": first_page}, "TotalCount": total},
        {"Accounts": {"Account": last_page}, "TotalCount": total},
    ])

    with pytest.raises(RuntimeError, match="cardinality disagrees"):
        await driver.list_scopes()


@pytest.mark.asyncio
async def test_list_scopes_rejects_nonempty_page_with_zero_total() -> None:
    driver = _driver()
    driver._own_account_id = AsyncMock(  # type: ignore[method-assign]
        return_value="5101999999999999",
    )
    driver._acs3_call = AsyncMock(return_value={  # type: ignore[method-assign]
        "Accounts": {"Account": [{
            "AccountId": "5101000000000001",
            "Status": "CreateSuccess",
        }]},
        "TotalCount": 0,
    })

    with pytest.raises(RuntimeError, match="cardinality disagrees"):
        await driver.list_scopes()


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "total_count",
    [True, False, 1.9, "1", "1.0", None, {"value": 1}],
)
async def test_list_scopes_rejects_nonintegral_total_count(
    total_count: object,
) -> None:
    driver = _driver()
    driver._own_account_id = AsyncMock(  # type: ignore[method-assign]
        return_value="5101999999999999",
    )
    driver._acs3_call = AsyncMock(return_value={  # type: ignore[method-assign]
        "Accounts": {"Account": []},
        "TotalCount": total_count,
    })

    with pytest.raises(RuntimeError, match="non-integral or negative"):
        await driver.list_scopes()


@pytest.mark.asyncio
@pytest.mark.parametrize("payload", [{"raw": "gateway html"}, {}])
async def test_list_scopes_rejects_missing_authority_fields(payload) -> None:
    driver = _driver()
    driver._own_account_id = AsyncMock(  # type: ignore[method-assign]
        return_value="5101999999999999",
    )
    driver._acs3_call = AsyncMock(return_value=payload)  # type: ignore[method-assign]

    with pytest.raises(RuntimeError, match="omitted Accounts or TotalCount"):
        await driver.list_scopes()


@pytest.mark.asyncio
async def test_list_scopes_fails_closed_when_master_identity_is_unavailable() -> None:
    driver = _driver()
    driver._own_account_id = AsyncMock(  # type: ignore[method-assign]
        side_effect=RuntimeError("STS transient failure"),
    )
    driver._acs3_call = AsyncMock()  # type: ignore[method-assign]

    with pytest.raises(RuntimeError, match="STS transient failure"):
        await driver.list_scopes()
    driver._acs3_call.assert_not_awaited()


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "account_id",
    [None, "", "123456789", "not-a-number", {"nested": "1234567890123456"}],
)
async def test_list_scopes_rejects_noncanonical_member_account_id(
    account_id: object,
) -> None:
    """RD members must meet the same identity contract as the STS master."""
    driver = _driver()
    driver._own_account_id = AsyncMock(  # type: ignore[method-assign]
        return_value="5101999999999999",
    )
    driver._acs3_call = AsyncMock(return_value={  # type: ignore[method-assign]
        "Accounts": {"Account": [{
            "AccountId": account_id,
            "Status": "CreateSuccess",
        }]},
        "TotalCount": 1,
    })

    with pytest.raises(RuntimeError, match="non-canonical AccountId"):
        await driver.list_scopes()


@pytest.mark.asyncio
async def test_list_scopes_rejects_total_change_during_pagination() -> None:
    driver = _driver()
    driver._own_account_id = AsyncMock(return_value="5101999999999999")  # type: ignore[method-assign]
    first_page = [
        {"AccountId": f"510100000000{i:04d}", "Status": "CreateSuccess"}
        for i in range(100)
    ]
    driver._acs3_call = AsyncMock(side_effect=[  # type: ignore[method-assign]
        {"Accounts": {"Account": first_page}, "TotalCount": 101},
        {
            "Accounts": {"Account": [{
                "AccountId": "5101000000009999",
                "Status": "CreateSuccess",
            }]},
            "TotalCount": 102,
        },
    ])

    with pytest.raises(RuntimeError, match="changed TotalCount"):
        await driver.list_scopes()


@pytest.mark.asyncio
async def test_list_scopes_rejects_page_ceiling() -> None:
    driver = _driver()
    driver._own_account_id = AsyncMock(return_value="5101999999999999")  # type: ignore[method-assign]

    async def _page(**kwargs):
        page = kwargs["params"]["PageNumber"]
        return {
            "Accounts": {"Account": [
                {
                    "AccountId": f"5101{page:04d}{i:08d}",
                    "Status": "CreateSuccess",
                }
                for i in range(100)
            ]},
            "TotalCount": 301,
        }

    driver._acs3_call = AsyncMock(side_effect=_page)  # type: ignore[method-assign]
    with (
        patch("app.integrations.drivers.aliyun_cloud._MAX_SCOPE_PAGES", 2),
        pytest.raises(RuntimeError, match="2-page safety ceiling"),
    ):
        await driver.list_scopes()


# ── AliyunReserved* policy filter (PR #565) ─────────────────────


@pytest.mark.asyncio
@respx.mock
async def test_list_scope_roles_filters_aliyun_reserved_sso_policies() -> None:
    """CloudSSO-enabled accounts auto-create ``Custom`` RAM policies
    named ``AliyunReservedSSO-<config>-<config>``. Aliyun rejects
    ``AttachPolicyToUser`` on them with ``NotSupport.AliyunReservedPolicy``,
    so they must never reach the request dropdown.
    """
    d = _driver()
    ref = CloudScopeRef(scope_type="aliyun_account", external_id="acc-1")

    def _handler(request: httpx.Request) -> httpx.Response:
        action = request.headers.get("x-acs-action")
        if action == "ListPolicies":
            qs = urllib.parse.parse_qs(request.url.query.decode())
            ptype = (qs.get("PolicyType") or [""])[0]
            if ptype == "Custom":
                return httpx.Response(200, json={
                    "Policies": {"Policy": [
                        {"PolicyName": "CompanyDevPolicy", "Description": "real custom"},
                        # Three flavors of the reserved namespace.
                        {"PolicyName": "AliyunReservedSSO-poweruser-poweruser",
                         "Description": "cloudsso mirror"},
                        {"PolicyName": "AliyunReservedSSO-admin-admin",
                         "Description": "cloudsso mirror"},
                        {"PolicyName": "AliyunReservedFoo", "Description": "future reserved"},
                    ]},
                })
            # PolicyType=System — return minimal set
            return httpx.Response(200, json={"Policies": {"Policy": [
                {"PolicyName": "ReadOnlyAccess", "Description": "rw"},
            ]}})
        return httpx.Response(404)

    respx.mock.post(url__regex=r"^https://ram\.aliyuncs\.com/.*").mock(side_effect=_handler)

    # ``list_scope_roles`` calls ``load_default_roles`` which queries
    # ``platform_settings``; in this driver-level unit test the table
    # doesn't exist (no DB fixture). Patch it to a no-op — we only
    # care about whether the reserved-prefix filter kicks in for the
    # live API portion of the merge.
    with patch(
        "app.integrations.drivers._cloud_defaults.load_default_roles",
        return_value=[],
    ):
        roles = await d.list_scope_roles(ref)
    names = {r.role_key for r in roles}

    assert "Custom:CompanyDevPolicy" in names
    assert "System:ReadOnlyAccess" in names
    # AliyunReservedSSO-* CloudSSO mirrors are the only documented
    # unattachable Aliyun namespace — filter must be narrow enough
    # to skip ONLY them, leaving any other ``AliyunReserved...`` name
    # that happens to be a legitimate Custom policy alone.
    assert not any(r.role_key.endswith("AliyunReservedSSO-poweruser-poweruser") for r in roles)
    assert not any(r.role_key.endswith("AliyunReservedSSO-admin-admin") for r in roles)
    # AliyunReservedFoo does NOT start with ``AliyunReservedSSO-`` so
    # it must pass through (an admin-created Custom policy with this
    # name is attachable per Aliyun's API — was the false-positive
    # surfaced in Round-1 code review of the original broad prefix).
    assert "Custom:AliyunReservedFoo" in names


@pytest.mark.asyncio
@respx.mock
async def test_list_scope_roles_raises_on_unchanged_marker() -> None:
    """Regression (#1655): ListPolicies returning ``IsTruncated=True`` with an
    unchanged/empty Marker is a degenerate response — it must RAISE rather
    than spin forever (pinning the worker) or silently truncate the catalog."""
    d = _driver()
    ref = CloudScopeRef(scope_type="aliyun_account", external_id="acc-1")

    def _handler(request: httpx.Request) -> httpx.Response:
        if request.headers.get("x-acs-action") == "ListPolicies":
            # Always truncated, always the same marker → would loop forever
            # without the guard.
            return httpx.Response(200, json={
                "Policies": {"Policy": [
                    {"PolicyName": "SomePolicy", "Description": "x"},
                ]},
                "IsTruncated": True,
                "Marker": "STUCK",
            })
        return httpx.Response(404)

    respx.mock.post(url__regex=r"^https://ram\.aliyuncs\.com/.*").mock(side_effect=_handler)
    with patch(
        "app.integrations.drivers._cloud_defaults.load_default_roles",
        return_value=[],
    ), pytest.raises(RuntimeError, match="unchanged/empty Marker"):
        await d.list_scope_roles(ref)


@pytest.mark.asyncio
async def test_grant_access_rejects_aliyun_reserved_policy() -> None:
    """Defense in depth: even if a legacy ``cmdb_cloud_roles`` row
    still carries an ``AliyunReserved*`` policy name, ``grant_access``
    must short-circuit with a clear message before hitting Aliyun's
    raw 400.
    """
    d = _driver()
    ref = CloudScopeRef(scope_type="aliyun_account", external_id="acc-1")
    ctx = CloudAccessContext(
        principal_email="someone@example.com",
        principal_display_name="Someone",
        scope=ref,
        role_key="Custom:AliyunReservedSSO-poweruser-poweruser",
    )

    result = await d.grant_access(ctx)

    assert result.ok is False
    assert "reserved" in (result.error_message or "").lower()
    assert "CloudSSO" in (result.error_message or "")


# ── create_scope — ParentFolderId is optional (defaults to RD Root) ──


@pytest.mark.asyncio
async def test_create_scope_resolves_root_folder_when_unset() -> None:
    """Regression (2026-06-03): the self-service "申请新阿里云子账号"
    flow permanently failed with "missing rd_parent_folder_id" because
    the driver hard-raised on a blank folder id. Aliyun's
    ``CreateResourceAccount`` treats ``ParentFolderId`` as OPTIONAL.
    Prod verified that neither aliyun-primary nor alicloud-intl-primary
    ever had the field configured, so the guard blocked every request.
    ``create_scope`` must now resolve the RD Root folder via
    ``GetResourceDirectory`` and send it as ``ParentFolderId``.
    """
    d = AliyunCloudDriver(integration_id="aliyun-test", config=dict(_CFG))
    captured: dict = {}

    async def _fake_acs3(self, *, host, version, action, params):
        if action == "GetResourceDirectory":
            return {"ResourceDirectory": {
                "RootFolderId": "fd-root-001",
                "ResourceDirectoryId": "rd-xyz",
                "MasterAccountId": "5962436752113096",
            }}
        captured.update({"action": action, **params})
        return {"Account": {"AccountId": "5999000000000001",
                            "DisplayName": params["DisplayName"]}}

    with patch.object(AliyunCloudDriver, "_acs3_call", _fake_acs3):
        scope = await d.create_scope(
            name="Team Datalake",
            description="self-service",
            owner_email="owner@example.com",
            budget_usd_monthly=1000.0,
        )

    assert captured["action"] == "CreateResourceAccount"
    assert captured["ParentFolderId"] == "fd-root-001"  # resolved Root
    assert captured["DisplayName"] == "Team Datalake"  # readable, kept
    # AccountNamePrefix is alphanumeric-only (space stripped, not dashed)
    assert captured["AccountNamePrefix"] == "teamdatalake"
    assert captured["ResellAccountType"] == "ResellAccount"
    assert scope.external_id == "5999000000000001"
    assert scope.raw["rd_parent_folder_id"] == "fd-root-001"
    assert scope.raw["rd_parent_folder_source"] == "resolved_root"
    # 结算账号 defaults to the RD 管理账号 (see the payer tests below).
    assert captured["PayerAccountId"] == "5962436752113096"


@pytest.mark.asyncio
async def test_create_scope_raises_when_not_rd_master() -> None:
    """If the credentials are NOT the RD master, GetResourceDirectory
    returns no RootFolderId — fail loud with an actionable message
    rather than sending a blank ParentFolderId."""
    d = AliyunCloudDriver(integration_id="aliyun-test", config=dict(_CFG))

    async def _fake_acs3(self, *, host, version, action, params):
        if action == "GetResourceDirectory":
            return {"ResourceDirectory": {}}  # member account → no root
        raise AssertionError("CreateResourceAccount must not be reached")

    with patch.object(AliyunCloudDriver, "_acs3_call", _fake_acs3):
        with pytest.raises(RuntimeError, match="master account"):
            await d.create_scope(
                name="Team X", description=None,
                owner_email="o@example.com", budget_usd_monthly=10.0,
            )


@pytest.mark.asyncio
async def test_create_scope_sends_parent_folder_when_set() -> None:
    """When the admin DOES configure a 资源夹, the account is filed
    under it — ``ParentFolderId`` is sent verbatim."""
    cfg = dict(_CFG, rd_parent_folder_id="fd-abc123")
    d = AliyunCloudDriver(integration_id="aliyun-test", config=cfg)
    captured: dict = {}

    async def _fake_acs3(self, *, host, version, action, params):
        if action == "GetResourceDirectory":
            # Still consulted for the 结算账号 (管理账号代付) even when the
            # 资源夹 is configured.
            return {"ResourceDirectory": {"RootFolderId": "r-Root01",
                                          "MasterAccountId":
                                              "5962436752113096"}}
        captured.update(params)
        return {"Account": {"AccountId": "5999000000000002",
                            "DisplayName": params["DisplayName"]}}

    with patch.object(AliyunCloudDriver, "_acs3_call", _fake_acs3):
        scope = await d.create_scope(
            name="Team Edge",
            description=None,
            owner_email="owner@example.com",
            budget_usd_monthly=500.0,
        )

    assert captured["ParentFolderId"] == "fd-abc123"
    assert scope.raw["rd_parent_folder_id"] == "fd-abc123"


# ── create_scope — 结算账号 (PayerAccountId) ─────────────────────────


def _rd_payer_fake(captured: dict, calls: list[str], account_id: str,
                   *, master_id: str | None = "5962436752113096"):
    """Fake ``_acs3_call`` recording actions + CreateResourceAccount
    params, with a configurable RD ``MasterAccountId``."""

    async def _fake_acs3(self, *, host, version, action, params):
        calls.append(action)
        if action == "GetResourceDirectory":
            rd: dict = {"RootFolderId": "r-Root01"}
            if master_id is not None:
                rd["MasterAccountId"] = master_id
            return {"ResourceDirectory": rd}
        captured.update(params)
        return {"Account": {"AccountId": account_id,
                            "DisplayName": params["DisplayName"]}}

    return _fake_acs3


@pytest.mark.asyncio
async def test_create_scope_defaults_payer_to_rd_master() -> None:
    """Regression (2026-08-07, prod): Aliyun's CreateResourceAccount
    treats ``PayerAccountId`` as optional and defaults to 「新成员自主付款」.
    Helix never sent it, so every self-service 阿里云子账号 came out
    self-paying — caught on ``data-integration-team-internal-alicloud``
    (AccountId 5083883504403318, created 2026-07-08), whose 当前结算账号
    read 自主付款 and had to be flipped to 「使用管理账号为新成员付款」 by hand.
    Default must be the RD 管理账号.
    """
    d = AliyunCloudDriver(integration_id="aliyun-test", config=dict(_CFG))
    captured: dict = {}
    calls: list[str] = []

    with patch.object(AliyunCloudDriver, "_acs3_call",
                      _rd_payer_fake(captured, calls, "5999000000000010")):
        scope = await d.create_scope(
            name="Team Payer", description=None,
            owner_email="o@example.com", budget_usd_monthly=1.0,
        )

    assert captured["PayerAccountId"] == "5962436752113096"
    assert scope.raw["payer_account_id"] == "5962436752113096"
    assert scope.raw["payer_source"] == "resolved_master"
    # RootFolderId + MasterAccountId come from ONE cached call.
    assert calls.count("GetResourceDirectory") == 1


@pytest.mark.asyncio
async def test_create_scope_payer_explicit_member_account() -> None:
    """``rd_payer_account_id=<AccountId>`` → console
    「使用已有成员为新成员付款」."""
    cfg = dict(_CFG, rd_payer_account_id="5083883504403318")
    d = AliyunCloudDriver(integration_id="aliyun-test", config=cfg)
    captured: dict = {}
    calls: list[str] = []

    with patch.object(AliyunCloudDriver, "_acs3_call",
                      _rd_payer_fake(captured, calls, "5999000000000011")):
        scope = await d.create_scope(
            name="Team Payer2", description=None,
            owner_email="o@example.com", budget_usd_monthly=1.0,
        )

    assert captured["PayerAccountId"] == "5083883504403318"
    assert scope.raw["payer_source"] == "config"


@pytest.mark.asyncio
async def test_create_scope_payer_self_opt_out_sends_no_param() -> None:
    """``rd_payer_account_id=self`` is the only way to get the old
    自主付款 behaviour — and it must be an explicit operator choice."""
    cfg = dict(_CFG, rd_payer_account_id="self")
    d = AliyunCloudDriver(integration_id="aliyun-test", config=cfg)
    captured: dict = {}
    calls: list[str] = []

    with patch.object(AliyunCloudDriver, "_acs3_call",
                      _rd_payer_fake(captured, calls, "5999000000000012")):
        scope = await d.create_scope(
            name="Team Payer3", description=None,
            owner_email="o@example.com", budget_usd_monthly=1.0,
        )

    assert "PayerAccountId" not in captured
    assert scope.raw["payer_account_id"] is None
    assert scope.raw["payer_source"] == "config_self"


@pytest.mark.asyncio
async def test_create_scope_raises_when_master_account_id_missing() -> None:
    """No MasterAccountId and no explicit override → fail loud instead of
    silently creating a self-paying member (the 2026-08-07 bug)."""
    d = AliyunCloudDriver(integration_id="aliyun-test", config=dict(_CFG))
    captured: dict = {}
    calls: list[str] = []

    with patch.object(AliyunCloudDriver, "_acs3_call",
                      _rd_payer_fake(captured, calls, "5999000000000013",
                                     master_id=None)):
        with pytest.raises(RuntimeError, match="MasterAccountId"):
            await d.create_scope(
                name="Team Payer4", description=None,
                owner_email="o@example.com", budget_usd_monthly=1.0,
            )

    assert "CreateResourceAccount" not in calls


@pytest.mark.asyncio
async def test_create_scope_rejects_non_numeric_payer_config() -> None:
    """A typo'd 结算账号 must fail loudly, not fall through to 自主付款."""
    cfg = dict(_CFG, rd_payer_account_id="sa@pingcap.com")
    d = AliyunCloudDriver(integration_id="aliyun-test", config=cfg)
    captured: dict = {}
    calls: list[str] = []

    with patch.object(AliyunCloudDriver, "_acs3_call",
                      _rd_payer_fake(captured, calls, "5999000000000014")):
        with pytest.raises(RuntimeError, match="rd_payer_account_id"):
            await d.create_scope(
                name="Team Payer5", description=None,
                owner_email="o@example.com", budget_usd_monthly=1.0,
            )

    assert "CreateResourceAccount" not in calls


@pytest.mark.asyncio
async def test_create_scope_strips_brand_and_punctuation_from_prefix() -> None:
    """Regression (2026-06-03, prod): the live 2020-03-31
    CreateResourceAccount rejected ``AccountNamePrefix`` containing the
    Alibaba brand token ``aliyun`` with
    ``IllegalParameter.Account.AccountName``. PROVEN empirically against
    prod: prefix ``devofflineafterworldaliyun`` was refused; the
    identical request with only ``aliyun`` removed
    (``devofflineafterworld``) created the account (id 5660180459278784).
    So the prefix drops both punctuation (alnum-only) AND Alibaba brand
    tokens; DisplayName keeps the readable, ``aliyun``-bearing name.
    """
    d = AliyunCloudDriver(integration_id="aliyun-test", config=dict(_CFG))
    captured: dict = {}

    async def _fake_acs3(self, *, host, version, action, params):
        if action == "GetResourceDirectory":
            return {"ResourceDirectory": {"RootFolderId": "r-Root01",
                                   "MasterAccountId": "5962436752113096"}}
        captured.update(params)
        return {"Account": {"AccountId": "5999000000000003",
                            "DisplayName": params["DisplayName"]}}

    with patch.object(AliyunCloudDriver, "_acs3_call", _fake_acs3):
        await d.create_scope(
            name="dev-offline-afterworld-aliyun",
            description=None,
            owner_email="owner@example.com",
            budget_usd_monthly=100.0,
        )

    # Hyphens dropped (alnum-only) AND 'aliyun' stripped → matches the
    # prefix that actually succeeded in prod.
    assert captured["AccountNamePrefix"] == "devofflineafterworld"
    assert "aliyun" not in captured["AccountNamePrefix"]
    assert "-" not in captured["AccountNamePrefix"]
    # DisplayName keeps the brand word + hyphens (accepted by Aliyun).
    assert captured["DisplayName"] == "dev-offline-afterworld-aliyun"


@pytest.mark.asyncio
async def test_create_scope_prefix_all_brand_falls_back_to_helix() -> None:
    """A name that reduces to nothing after brand+punctuation stripping
    (e.g. 'Aliyun') must not send an empty/too-short prefix — pad to a
    safe default so CreateResourceAccount still gets a legal prefix."""
    d = AliyunCloudDriver(integration_id="aliyun-test", config=dict(_CFG))
    captured: dict = {}

    async def _fake_acs3(self, *, host, version, action, params):
        if action == "GetResourceDirectory":
            return {"ResourceDirectory": {"RootFolderId": "r-Root01",
                                   "MasterAccountId": "5962436752113096"}}
        captured.update(params)
        return {"Account": {"AccountId": "5999000000000005",
                            "DisplayName": params["DisplayName"]}}

    with patch.object(AliyunCloudDriver, "_acs3_call", _fake_acs3):
        await d.create_scope(
            name="Aliyun", description=None,
            owner_email="o@example.com", budget_usd_monthly=1.0,
        )

    assert captured["AccountNamePrefix"] == "helix"
    assert captured["DisplayName"] == "Aliyun"  # DisplayName untouched


@pytest.mark.asyncio
async def test_create_scope_prefix_caps_at_37_and_strips_spaces() -> None:
    """AccountNamePrefix is bounded to Aliyun's 37-char limit and drops
    spaces/punctuation (the old slug capped at 50 and dashed spaces)."""
    d = AliyunCloudDriver(integration_id="aliyun-test", config=dict(_CFG))
    captured: dict = {}

    async def _fake_acs3(self, *, host, version, action, params):
        if action == "GetResourceDirectory":
            return {"ResourceDirectory": {"RootFolderId": "r-Root01",
                                   "MasterAccountId": "5962436752113096"}}
        captured.update(params)
        return {"Account": {"AccountId": "5999000000000004",
                            "DisplayName": params["DisplayName"]}}

    long_name = "Acme Global Platform Engineering Sandbox Team Alpha"
    with patch.object(AliyunCloudDriver, "_acs3_call", _fake_acs3):
        await d.create_scope(
            name=long_name, description=None,
            owner_email="o@example.com", budget_usd_monthly=1.0,
        )

    prefix = captured["AccountNamePrefix"]
    assert len(prefix) <= 37
    assert prefix.isalnum() and prefix.islower()
    assert prefix == "acmeglobalplatformengineeringsandboxteamalpha"[:37]


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("name", "expected_prefix"),
    [
        # Digit-leading names pass the upstream alnum validator + FE rule
        # but Aliyun rejects an AccountNamePrefix not starting with a
        # letter — prepend "h".
        ("123", "h123"),
        ("01-datalake", "h01datalake"),
        ("2024-prod", "h2024prod"),
        # Brand-stripping can turn a letter-leading name into a
        # digit-leading slug; the check runs AFTER stripping.
        ("aliyun123", "h123"),
        # Letter-leading names are untouched.
        ("datalake", "datalake"),
    ],
)
async def test_create_scope_prefix_forces_letter_lead(
    name, expected_prefix,
) -> None:
    """Finding #4 (2026-06-04): AccountNamePrefix must start with a
    lowercase letter or CreateResourceAccount fails with
    IllegalParameter.Account.AccountName."""
    d = AliyunCloudDriver(integration_id="aliyun-test", config=dict(_CFG))
    captured: dict = {}

    async def _fake_acs3(self, *, host, version, action, params):
        if action == "GetResourceDirectory":
            return {"ResourceDirectory": {"RootFolderId": "r-Root01",
                                   "MasterAccountId": "5962436752113096"}}
        captured.update(params)
        return {"Account": {"AccountId": "5999000000000006",
                            "DisplayName": params["DisplayName"]}}

    with patch.object(AliyunCloudDriver, "_acs3_call", _fake_acs3):
        await d.create_scope(
            name=name, description=None,
            owner_email="o@example.com", budget_usd_monthly=1.0,
        )

    prefix = captured["AccountNamePrefix"]
    assert prefix == expected_prefix
    assert prefix[0].isascii() and prefix[0].isalpha()