"""AliCloud Intl grants — Cloud SSO ``ListAccessAssignments`` is the
authoritative source, NOT RAM users + attached policies.

Regression context (2026-04-26): screenshot showed 4 users on the
``alicloud-apac-alicloud-presales`` sub-account via Cloud SSO
(shuai.feng=admin, dunbin.shi/brian.niu/yingkun.yang=poweruser).
Helix DB had 25 grants on the same sub-account, all derived from
RAM users on the master — completely different (and wrong) people.

Pin the contract: ``list_scope_grants`` calls Cloud SSO, expands
groups, emits one ``DiscoveredGrant`` per (user, AccessConfig).
"""
from __future__ import annotations

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

import pytest

from app.integrations.driver_base import CloudScopeRef
from app.integrations.drivers.alicloud_intl_cloud import AlicloudIntlCloudDriver


def _make_driver() -> AlicloudIntlCloudDriver:
    """Build a driver instance with stub config so we can patch
    ``_acs3_call`` directly."""
    d = AlicloudIntlCloudDriver(
        integration_id="x", config={
            "access_key_id": "AK", "access_key_secret": "SK",
        },
    )
    return d


@pytest.mark.asyncio
async def test_screenshot_assignments_match_emitted_grants():
    """The exact 4-user response from the prod probe must round-trip
    into 4 DiscoveredGrant records with the right emails + access
    configs."""
    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method, params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-test"}]}
        if action == "ListAccessAssignments":
            return {
                "AccessAssignments": [
                    {
                        "AccessConfigurationName": "admin",
                        "AccessConfigurationId": "ac-admin",
                        "PrincipalType": "User",
                        "PrincipalId": "u-shuai",
                        "PrincipalName": "shuai.feng@pingcap.com",
                        "TargetType": "RD-Account",
                        "TargetId": "5984872860690848",
                        "CreateTime": "2026-03-07T05:49:17Z",
                    },
                    {
                        "AccessConfigurationName": "poweruser",
                        "AccessConfigurationId": "ac-power",
                        "PrincipalType": "User",
                        "PrincipalId": "u-dunbin",
                        "PrincipalName": "dunbin.shi@pingcap.com",
                        "TargetType": "RD-Account",
                        "TargetId": "5984872860690848",
                    },
                    {
                        "AccessConfigurationName": "poweruser",
                        "AccessConfigurationId": "ac-power",
                        "PrincipalType": "User",
                        "PrincipalId": "u-brian",
                        "PrincipalName": "brian.niu@pingcap.com",
                        "TargetType": "RD-Account",
                        "TargetId": "5984872860690848",
                    },
                    {
                        "AccessConfigurationName": "poweruser",
                        "AccessConfigurationId": "ac-power",
                        "PrincipalType": "User",
                        "PrincipalId": "u-yingkun",
                        "PrincipalName": "yingkun.yang@pingcap.com",
                        "TargetType": "RD-Account",
                        "TargetId": "5984872860690848",
                    },
                ],
                "NextToken": None,
            }
        return {}

    scope = CloudScopeRef(
        scope_type="aliyun_account",
        external_id="5984872860690848",
        display_name="alicloud-apac-alicloud-presales",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        grants = await driver.list_scope_grants(scope)

    assert len(grants) == 4
    by_email = {g.principal_email: g for g in grants}
    assert by_email["shuai.feng@pingcap.com"].role_key == "admin"
    assert by_email["dunbin.shi@pingcap.com"].role_key == "poweruser"
    assert by_email["brian.niu@pingcap.com"].role_key == "poweruser"
    assert by_email["yingkun.yang@pingcap.com"].role_key == "poweruser"
    # Source flag preserved for audit
    assert all(
        g.raw and g.raw.get("source") == "cloudsso" for g in grants
    )


@pytest.mark.asyncio
async def test_group_assignments_expand_to_member_users():
    """Cloud SSO supports Group-type assignments (e.g. ADSyncAdmins).
    The driver must expand groups into individual member grants so
    Helix's per-person view matches reality."""
    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method, params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-test"}]}
        if action == "ListAccessAssignments":
            return {
                "AccessAssignments": [
                    {
                        "AccessConfigurationName": "admin",
                        "AccessConfigurationId": "ac-admin",
                        "PrincipalType": "Group",
                        "PrincipalId": "g-devs",
                        "PrincipalName": "dev-control-plane",
                        "TargetType": "RD-Account",
                        "TargetId": "scope-1",
                    },
                ],
                "NextToken": None,
            }
        if action == "ListGroupMembers":
            return {
                "GroupMembers": [
                    {"UserId": "u-1", "Email": "alice@pingcap.com",
                     "UserName": "alice@pingcap.com"},
                    {"UserId": "u-2", "Email": "bob@pingcap.com",
                     "UserName": "bob@pingcap.com"},
                ],
                "NextToken": None,
            }
        return {}

    scope = CloudScopeRef(
        scope_type="aliyun_account", external_id="scope-1",
        display_name="scope-1",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        grants = await driver.list_scope_grants(scope)

    emails = {g.principal_email for g in grants}
    assert emails == {"alice@pingcap.com", "bob@pingcap.com"}
    assert all(g.role_key == "admin" for g in grants)
    # Group context preserved on the raw payload so audit can see
    # which group an admin came in via.
    for g in grants:
        assert g.raw["GroupName"] == "dev-control-plane"


@pytest.mark.asyncio
async def test_no_directory_falls_back_to_parent_ram():
    """Tenants without Cloud SSO provisioned (legacy AliCloud setups)
    must still get something — fall back to the parent's RAM-based
    discovery so we don't return 0 grants for everything."""
    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method, params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": []}
        return {}

    # Parent's list_scope_grants is also called via _acs3_call to RAM
    # endpoints. Stub it directly so we don't need to mock RAM too.
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)), \
         patch(
             "app.integrations.drivers.aliyun_cloud.AliyunCloudDriver."
             "list_scope_grants",
             new=AsyncMock(return_value=["RAM-fallback-marker"]),
         ):
        scope = CloudScopeRef(
            scope_type="aliyun_account", external_id="x",
            display_name="x",
        )
        grants = await driver.list_scope_grants(scope)
    assert grants == ["RAM-fallback-marker"]


@pytest.mark.asyncio
async def test_paginates_assignments():
    """RD scopes can have >page-size assignments — pin NextToken
    pagination."""
    driver = _make_driver()
    # Round-5 harmonisation: paginator now stops on IsTruncated=false
    # OR missing NextToken (defensive). Real Aliyun always returns
    # IsTruncated; the mock now matches that.
    pages = [
        {"AccessAssignments": [
            {"AccessConfigurationName": "admin", "AccessConfigurationId": "a",
             "PrincipalType": "User", "PrincipalId": f"u{i}",
             "PrincipalName": f"user{i}@pingcap.com",
             "TargetType": "RD-Account", "TargetId": "s"}
            for i in range(5)
        ], "IsTruncated": True, "NextToken": "tk"},
        {"AccessAssignments": [
            {"AccessConfigurationName": "poweruser", "AccessConfigurationId": "p",
             "PrincipalType": "User", "PrincipalId": f"u{i}",
             "PrincipalName": f"user{i}@pingcap.com",
             "TargetType": "RD-Account", "TargetId": "s"}
            for i in range(5, 8)
        ], "IsTruncated": False, "NextToken": None},
    ]
    page_iter = iter(pages)

    async def fake_acs3(*, host, version, action, method, params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d"}]}
        if action == "ListAccessAssignments":
            return next(page_iter)
        return {}

    scope = CloudScopeRef(
        scope_type="aliyun_account", external_id="s", display_name="s",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        grants = await driver.list_scope_grants(scope)
    assert len(grants) == 8


# ── revoke_access (CloudSSO) regression — round 2026-05-13 ──
#
# Prod incident: offboarding huohao@pingcap.com cycled 58 times with
# ``Aliyun DetachPolicyFromUser 400: InvalidParameter.PolicyName.
# InvalidChars`` because the inherited RAM-based revoke parsed an
# SSO external_ref ``sso::ac-xxx::u-yyy::<account>`` as
# ``user_name='sso'`` + ``PolicyName=':<rest>'``. These pin the
# CloudSSO removal path so the same shape can't regress.


@pytest.mark.asyncio
async def test_revoke_access_sso_user_calls_delete_access_assignment():
    """User-direct SSO ref → DeleteAccessAssignment with full param set."""
    from app.integrations.driver_base import CloudAccessContext, CloudScopeRef
    driver = _make_driver()
    captured: list[dict[str, Any]] = []

    async def fake_acs3(*, host, version, action, method, params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-test"}]}
        if action == "DeleteAccessAssignment":
            captured.append({"params": params, "method": method})
            return {"Task": {"TaskId": "t-1"}}
        raise AssertionError(f"unexpected action {action!r}")

    ctx = CloudAccessContext(
        principal_email="huohao@pingcap.com",
        principal_display_name="huohao",
        scope=CloudScopeRef(
            scope_type="aliyun_account",
            external_id="5752453686349026",
            display_name="dbaas-staging",
        ),
        role_key="poweruser",
        existing_external_ref=(
            "sso::ac-00gfoixsnlbb56v9mm13::u-002ilbgmmocsv7orlt76"
            "::5752453686349026"
        ),
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.revoke_access(ctx)
    assert result.ok is True
    assert captured == [{
        "method": "POST",
        "params": {
            "DirectoryId": "d-test",
            "AccessConfigurationId": "ac-00gfoixsnlbb56v9mm13",
            "TargetType": "RD-Account",
            "TargetId": "5752453686349026",
            "PrincipalType": "User",
            "PrincipalId": "u-002ilbgmmocsv7orlt76",
            "DeprovisionStrategy":
                "DeprovisionForLastAccessAssignmentOnAccount",
        },
    }]


@pytest.mark.asyncio
async def test_revoke_access_sso_group_expansion_removes_member_only():
    """Group-expanded ref → RemoveUserFromGroup (NOT DeleteAccessAssignment).

    Removing the assignment would revoke for every group member; the
    expected behavior on per-person offboarding is to drop only this
    one user from the SSO group.
    """
    from app.integrations.driver_base import CloudAccessContext, CloudScopeRef
    driver = _make_driver()
    captured: list[dict[str, Any]] = []

    async def fake_acs3(*, host, version, action, method, params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-1"}]}
        if action == "RemoveUserFromGroup":
            captured.append(params)
            return {}
        raise AssertionError(f"unexpected action {action!r}")

    ctx = CloudAccessContext(
        principal_email="x@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="acct-1", display_name="x",
        ),
        role_key="admin",
        existing_external_ref=(
            "sso::ac-1::u-1::acct-1::group::g-1"
        ),
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.revoke_access(ctx)
    assert result.ok is True
    assert captured == [{
        "DirectoryId": "d-1", "GroupId": "g-1", "UserId": "u-1",
    }]


@pytest.mark.asyncio
async def test_revoke_access_non_sso_ref_falls_through_to_parent():
    """RAM-policy refs (no ``sso::`` prefix) must keep using the
    parent's RAM detach path — don't break the legacy code path."""
    from app.integrations.driver_base import CloudAccessContext, CloudScopeRef
    driver = _make_driver()
    parent_called: list[CloudAccessContext] = []

    async def fake_parent_revoke(self, ctx):
        from app.integrations.driver_base import CloudGrantResult
        parent_called.append(ctx)
        return CloudGrantResult(ok=True, external_ref=ctx.existing_external_ref)

    ctx = CloudAccessContext(
        principal_email="x@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="a", display_name="x",
        ),
        role_key="System:AliyunOSSReadOnlyAccess",
        existing_external_ref="xuser::System:AliyunOSSReadOnlyAccess",
    )
    with patch(
        "app.integrations.drivers.aliyun_cloud.AliyunCloudDriver.revoke_access",
        new=fake_parent_revoke,
    ):
        result = await driver.revoke_access(ctx)
    assert result.ok is True
    assert len(parent_called) == 1
    assert parent_called[0].existing_external_ref == (
        "xuser::System:AliyunOSSReadOnlyAccess"
    )


@pytest.mark.asyncio
async def test_revoke_access_sso_already_absent_treated_as_success():
    """``AccessAssignmentNotExist`` from Aliyun → idempotent success."""
    from app.integrations.driver_base import CloudAccessContext, CloudScopeRef
    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method, params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d"}]}
        if action == "DeleteAccessAssignment":
            raise RuntimeError(
                "Aliyun DeleteAccessAssignment 404: AccessAssignmentNotExist"
            )
        raise AssertionError(action)

    ctx = CloudAccessContext(
        principal_email="x@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="t", display_name="x",
        ),
        role_key="admin",
        existing_external_ref="sso::ac::u::t",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.revoke_access(ctx)
    assert result.ok is True
    assert result.already_absent is True


@pytest.mark.asyncio
async def test_revoke_access_sso_no_directory_returns_already_absent():
    """SSO directory vanished → assignment can't exist; don't 502
    offboarding on a missing IdP."""
    from app.integrations.driver_base import CloudAccessContext, CloudScopeRef
    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method, params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": []}
        raise AssertionError(action)

    ctx = CloudAccessContext(
        principal_email="x@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="t", display_name="x",
        ),
        role_key="admin",
        existing_external_ref="sso::ac::u::t",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.revoke_access(ctx)
    assert result.ok is True
    assert result.already_absent is True


@pytest.mark.asyncio
async def test_revoke_access_sso_malformed_ref_rejected():
    """An ``sso::`` ref missing required segments must NOT silently
    fall through to the parent (where it would corrupt PolicyName)."""
    from app.integrations.driver_base import CloudAccessContext, CloudScopeRef
    driver = _make_driver()

    ctx = CloudAccessContext(
        principal_email="x@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="t", display_name="x",
        ),
        role_key="admin",
        existing_external_ref="sso::ac::u",  # missing target_id
    )
    # _acs3_call must not be called at all — malformed ref short-circuits.
    with patch.object(
        driver, "_acs3_call",
        new=AsyncMock(side_effect=AssertionError("must not call")),
    ):
        result = await driver.revoke_access(ctx)
    assert result.ok is False
    assert result.error_message and "Malformed" in result.error_message


# ── CloudSSO list_scope_roles + grant_access (PR #565 follow-up) ──


@pytest.mark.asyncio
async def test_list_scope_roles_returns_cloudsso_access_configurations():
    """In CloudSSO mode the role catalog is ``ListAccessConfigurations``,
    not ``ListPolicies``. ``role_key`` matches the convention
    ``list_scope_grants`` uses (AccessConfigurationName) so a granted
    role round-trips through the request flow."""
    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method, params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-test"}]}
        if action == "ListAccessConfigurations":
            return {
                "AccessConfigurations": [
                    {
                        "AccessConfigurationId": "ac-admin",
                        "AccessConfigurationName": "admin",
                        "Description": "Full admin",
                    },
                    {
                        "AccessConfigurationId": "ac-power",
                        "AccessConfigurationName": "poweruser",
                        "Description": None,
                    },
                    {
                        "AccessConfigurationId": "ac-dbaas",
                        "AccessConfigurationName": "DBaaS-Prod-Admin-Role",
                    },
                ],
                "NextToken": None,
            }
        raise AssertionError(f"unexpected action {action}")

    scope = CloudScopeRef(
        scope_type="aliyun_account", external_id="5984872860690848",
        display_name="alicloud-apac-alicloud-presales",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        roles = await driver.list_scope_roles(scope)

    keys = {r.role_key for r in roles}
    assert keys == {"admin", "poweruser", "DBaaS-Prod-Admin-Role"}
    # All sourced from CloudSSO
    assert all(r.source == "cloudsso" for r in roles)
    # All promoted (admin-defined permission sets — show on default dropdown)
    assert all(r.is_promoted for r in roles)
    # raw carries id for grant_access lookup
    by_name = {r.role_key: r for r in roles}
    assert by_name["poweruser"].raw["AccessConfigurationId"] == "ac-power"


@pytest.mark.asyncio
async def test_list_scope_roles_no_directory_falls_back_to_parent_ram():
    """Single-account intl tenant without CloudSSO directory must fall
    back to the parent's RAM ListPolicies enumeration (with the
    PR #565 AliyunReserved filter still applied)."""
    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": []}
        if action == "ListPolicies":
            return {"Policies": {"Policy": [
                {"PolicyName": "ReadOnlyAccess", "Description": "ro"},
                # Reserved namespace must still be filtered (PR #565).
                {"PolicyName": "AliyunReservedSSO-foo-foo",
                 "Description": "reserved"},
            ]}}
        return {}

    scope = CloudScopeRef(
        scope_type="aliyun_account", external_id="acct", display_name="x",
    )
    # Parent ``list_scope_roles`` calls ``load_default_roles`` which
    # reads ``platform_settings`` — table doesn't exist in this unit
    # test (no DB fixture). Patch to no-op; we only care about whether
    # the RAM live-merge path filters AliyunReserved*.
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)), \
         patch(
            "app.integrations.drivers._cloud_defaults.load_default_roles",
            new=AsyncMock(return_value=[]),
         ):
        roles = await driver.list_scope_roles(scope)

    keys = {r.role_key for r in roles}
    # Parent path emits "Type:Name" — AliyunReserved* filtered.
    assert "System:ReadOnlyAccess" in keys
    assert not any("AliyunReserved" in k for k in keys)


@pytest.mark.asyncio
async def test_grant_access_creates_cloudsso_assignment_and_returns_sso_ref():
    """The happy path: mohangjie applies for poweruser, dengshuan
    approves, Helix provisioning calls CreateAccessAssignment with the
    resolved (user_id, access_config_id, target_id) triple."""
    from app.integrations.driver_base import CloudAccessContext

    driver = _make_driver()
    create_calls: list[dict[str, Any]] = []

    async def fake_acs3(*, host, version, action, method, params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-test"}]}
        if action == "ListAccessConfigurations":
            return {"AccessConfigurations": [
                {"AccessConfigurationId": "ac-power",
                 "AccessConfigurationName": "poweruser"},
            ]}
        if action == "ListUsers":
            return {"Users": [
                {"UserId": "u-mohan", "UserName": "mohangjie@pingcap.com"},
            ]}
        if action == "CreateAccessAssignment":
            create_calls.append(dict(params or {}))
            return {"AccessAssignmentTask": {"Status": "InProgress",
                                             "TaskId": "task-xyz"}}
        raise AssertionError(f"unexpected action {action}")

    ctx = CloudAccessContext(
        principal_email="mohangjie@pingcap.com",
        principal_display_name="Mohangjie",
        scope=CloudScopeRef(
            scope_type="aliyun_account",
            external_id="5092239846523038",
            display_name="autonomous-dev",
        ),
        role_key="poweruser",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.grant_access(ctx)

    assert result.ok is True
    assert result.external_ref == "sso::ac-power::u-mohan::5092239846523038"
    # Detail keys snake_case to match
    # ``lifecycle_service._ALLOWED_EXTERNAL_DETAIL_KEYS``. Round-1
    # code review caught the prior PascalCase emit being silently
    # filtered to ``external_detail={}``.
    assert result.detail["access_configuration_name"] == "poweruser"
    assert result.detail["access_configuration_id"] == "ac-power"
    assert result.detail["principal_id"] == "u-mohan"
    assert result.detail["target_id"] == "5092239846523038"
    assert result.detail["target_type"] == "RD-Account"
    assert len(create_calls) == 1
    assert create_calls[0]["DirectoryId"] == "d-test"
    assert create_calls[0]["AccessConfigurationId"] == "ac-power"
    assert create_calls[0]["TargetId"] == "5092239846523038"
    assert create_calls[0]["TargetType"] == "RD-Account"
    assert create_calls[0]["PrincipalType"] == "User"
    assert create_calls[0]["PrincipalId"] == "u-mohan"


@pytest.mark.asyncio
async def test_grant_access_already_exists_treated_as_idempotent():
    """If the assignment already exists (admin set it up manually
    before Helix request landed), grant_access returns ``ok=True
    already_exists=True`` rather than failing the request."""
    from app.integrations.driver_base import CloudAccessContext

    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method, params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-test"}]}
        if action == "ListAccessConfigurations":
            return {"AccessConfigurations": [
                {"AccessConfigurationId": "ac-admin",
                 "AccessConfigurationName": "admin"},
            ]}
        if action == "ListUsers":
            return {"Users": [
                {"UserId": "u-x", "UserName": "x@pingcap.com"},
            ]}
        if action == "CreateAccessAssignment":
            raise RuntimeError(
                "EntityAlreadyExists.AccessAssignment: "
                "the assignment already exists"
            )
        raise AssertionError(action)

    ctx = CloudAccessContext(
        principal_email="x@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="t", display_name="x",
        ),
        role_key="admin",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.grant_access(ctx)

    assert result.ok is True
    assert result.already_exists is True
    assert result.external_ref == "sso::ac-admin::u-x::t"


@pytest.mark.asyncio
async def test_grant_access_unknown_role_returns_clean_error():
    """If the requested AccessConfiguration name is no longer in the
    directory (renamed / deleted), grant_access must surface a clear
    error message — not crash, not call CreateAccessAssignment."""
    from app.integrations.driver_base import CloudAccessContext

    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method, params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-test"}]}
        if action == "ListAccessConfigurations":
            return {"AccessConfigurations": [
                {"AccessConfigurationId": "ac-admin",
                 "AccessConfigurationName": "admin"},
            ]}
        if action == "CreateAccessAssignment":
            raise AssertionError("must not call when role unknown")
        return {}

    ctx = CloudAccessContext(
        principal_email="x@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="t", display_name="x",
        ),
        role_key="ghost-role",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.grant_access(ctx)

    assert result.ok is False
    assert "AccessConfiguration" in (result.error_message or "")
    assert "ghost-role" in (result.error_message or "")


@pytest.mark.asyncio
async def test_grant_access_unknown_user_returns_clean_error():
    """If the requester isn't provisioned in the CloudSSO directory
    (SAML/SCIM lag for a brand-new hire), surface a clean error
    instead of an opaque CreateAccessAssignment 400."""
    from app.integrations.driver_base import CloudAccessContext

    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method, params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-test"}]}
        if action == "ListAccessConfigurations":
            return {"AccessConfigurations": [
                {"AccessConfigurationId": "ac-admin",
                 "AccessConfigurationName": "admin"},
            ]}
        if action == "ListUsers":
            return {"Users": []}  # not provisioned yet
        if action == "CreateAccessAssignment":
            raise AssertionError("must not call when user unknown")
        return {}

    ctx = CloudAccessContext(
        principal_email="newhire@pingcap.com",
        principal_display_name="New",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="t", display_name="x",
        ),
        role_key="admin",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.grant_access(ctx)

    assert result.ok is False
    assert "newhire@pingcap.com" in (result.error_message or "")
    assert "SAML" in (result.error_message or "") or "SCIM" in (result.error_message or "")


@pytest.mark.asyncio
async def test_grant_access_no_directory_falls_back_to_parent_ram():
    """If the tenant has no CloudSSO directory at all (legacy
    single-account intl tenant), grant_access falls through to the
    parent RAM path. The PR #565 reserved-policy guard in the parent
    still protects against AliyunReserved* leakage."""
    from app.integrations.driver_base import CloudAccessContext

    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": []}
        if action == "CreateUser":
            return {"User": {"UserName": "x"}}
        if action == "AttachPolicyToUser":
            return {}
        raise AssertionError(action)

    ctx = CloudAccessContext(
        principal_email="x@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="t", display_name="x",
        ),
        role_key="System:ReadOnlyAccess",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.grant_access(ctx)

    # Parent RAM path runs (CreateUser + AttachPolicyToUser succeeded).
    assert result.ok is True


@pytest.mark.asyncio
async def test_grant_access_rejects_aliyun_reserved_role_via_parent_guard():
    """If somehow a legacy ``cmdb_cloud_roles`` row carries the
    parent's role_key shape (``Custom:AliyunReservedSSO-*``) and lands
    in the intl driver, the PR #565 parent guard short-circuits with
    a clear message — even when no SSO directory exists."""
    from app.integrations.driver_base import CloudAccessContext

    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method, params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": []}
        if action == "AttachPolicyToUser":
            raise AssertionError("must not reach RAM attach")
        return {}

    ctx = CloudAccessContext(
        principal_email="x@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="t", display_name="x",
        ),
        role_key="Custom:AliyunReservedSSO-poweruser-poweruser",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.grant_access(ctx)

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


# ── Round-1 code-review fixes — transport-error + cache + is_custom ──


@pytest.mark.asyncio
async def test_list_scope_roles_emits_is_custom_true_for_cloudsso_configs():
    """``is_custom`` MUST be True on CloudSSO permission sets so the
    role-catalog ``reset_role_catalog_to_default`` endpoint
    (lifecycle_router.py) doesn't demote every CloudSSO role to
    ``is_promoted=False`` — that endpoint computes
    ``target = bool(r.is_custom) OR r.role_key in default_keys`` and
    the cloud.aliyun.default_policies setting only lists RAM-style
    'System:' keys.
    """
    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-test"}]}
        if action == "ListAccessConfigurations":
            return {"AccessConfigurations": [
                {"AccessConfigurationId": "ac-power",
                 "AccessConfigurationName": "poweruser"},
            ]}
        raise AssertionError(action)

    scope = CloudScopeRef(
        scope_type="aliyun_account", external_id="t", display_name="x",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        roles = await driver.list_scope_roles(scope)

    assert len(roles) == 1
    assert roles[0].is_custom is True
    assert roles[0].is_promoted is True


@pytest.mark.asyncio
async def test_list_scope_roles_transient_failure_raises_not_fallback():
    """Round-4 logic review: ListAccessConfigurations transient
    failure must RAISE rather than fall back to parent RAM. The
    sync framework's per-scope try/except puts the scope into
    ``errors`` and skips it; falling back to RAM would emit
    ``Type:Name`` keys that pass the 50% drop-tolerance floor and
    cause mark-and-sweep to soft-delete the (un-touched) CloudSSO
    rows on the next sync, producing a mixed-shape catalog.
    """
    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-test"}]}
        if action == "ListAccessConfigurations":
            raise RuntimeError("ServiceUnavailable: cloudsso 503")
        if action == "ListPolicies":
            raise AssertionError("must not fall back to parent RAM")
        return {}

    scope = CloudScopeRef(
        scope_type="aliyun_account", external_id="t", display_name="x",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)), \
         patch(
            "app.integrations.drivers._cloud_defaults.load_default_roles",
            new=AsyncMock(return_value=[]),
         ), \
         pytest.raises(RuntimeError, match="ServiceUnavailable"):
        await driver.list_scope_roles(scope)


@pytest.mark.asyncio
async def test_list_access_configurations_does_not_cache_empty():
    """Empty result must NOT poison the cache. Round-1 code review:
    a transient empty payload would otherwise lock every subsequent
    scope in the same driver-instance lifetime to ``[]``.
    """
    driver = _make_driver()
    call_count = {"n": 0}

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListAccessConfigurations":
            call_count["n"] += 1
            # First call returns empty (transient); second call
            # returns the real list. Cache must NOT have stored the
            # empty result — second call must hit the API again.
            if call_count["n"] == 1:
                return {"AccessConfigurations": []}
            return {"AccessConfigurations": [
                {"AccessConfigurationId": "ac-power",
                 "AccessConfigurationName": "poweruser"},
            ]}
        raise AssertionError(action)

    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        first = await driver._sso_list_access_configurations("d-test")
        second = await driver._sso_list_access_configurations("d-test")

    assert first == []
    assert len(second) == 1
    assert call_count["n"] == 2  # api hit both times — empty not cached


@pytest.mark.asyncio
async def test_list_access_configurations_caches_non_empty():
    """Non-empty success caches; second call short-circuits."""
    driver = _make_driver()
    call_count = {"n": 0}

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListAccessConfigurations":
            call_count["n"] += 1
            return {"AccessConfigurations": [
                {"AccessConfigurationId": "ac-x", "AccessConfigurationName": "x"},
            ]}
        raise AssertionError(action)

    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        a = await driver._sso_list_access_configurations("d-test")
        b = await driver._sso_list_access_configurations("d-test")

    assert a == b
    assert call_count["n"] == 1  # cache hit on second call


@pytest.mark.asyncio
async def test_grant_access_listusers_transient_error_returns_clean_message():
    """Round-1 code review: prior implementation swallowed RuntimeError
    in _sso_resolve_user_id and emitted "SAML/SCIM provisioning"
    error — misrouting ops investigation. Now a transport failure
    must produce a clean transient-error message.
    """
    from app.integrations.driver_base import CloudAccessContext

    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-test"}]}
        if action == "ListAccessConfigurations":
            return {"AccessConfigurations": [
                {"AccessConfigurationId": "ac-x", "AccessConfigurationName": "x"},
            ]}
        if action == "ListUsers":
            raise RuntimeError("ServiceUnavailable: cloudsso 503")
        raise AssertionError(action)

    ctx = CloudAccessContext(
        principal_email="someone@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="t", display_name="x",
        ),
        role_key="x",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.grant_access(ctx)

    assert result.ok is False
    msg = (result.error_message or "").lower()
    assert "temporarily unavailable" in msg or "retry" in msg
    # Critically: must NOT misroute to SAML/SCIM
    assert "saml" not in msg
    assert "scim" not in msg


@pytest.mark.asyncio
async def test_grant_access_list_access_configs_transient_error_returns_clean_message():
    """Same class as above but for ListAccessConfigurations failure.
    Must NOT misroute to "role renamed in console" — that's the
    user-error path; this is the transient-error path.
    """
    from app.integrations.driver_base import CloudAccessContext

    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-test"}]}
        if action == "ListAccessConfigurations":
            raise RuntimeError("ServiceUnavailable: cloudsso 503")
        raise AssertionError(action)

    ctx = CloudAccessContext(
        principal_email="x@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="t", display_name="x",
        ),
        role_key="poweruser",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.grant_access(ctx)

    assert result.ok is False
    msg = (result.error_message or "").lower()
    assert "temporarily unavailable" in msg or "retry" in msg
    # Must NOT misroute to "config renamed in console"
    assert "renamed" not in msg
    assert "deleted in the aliyun cloudsso console" not in msg


@pytest.mark.asyncio
async def test_grant_access_no_directory_with_cloudsso_role_key_returns_clean_error():
    """If the directory lookup transient-fails AND the role_key is a
    bare CloudSSO name (no ':'), grant_access must NOT fall through
    to the parent RAM path. Parent would reject with "role_key must
    be Type:Name" — a confusing error for a CloudSSO-populated
    catalog. Round-1 code review: the fallback asymmetry between
    revoke (defensive ``sso::`` guard) and grant (no guard) was
    flagged.
    """
    from app.integrations.driver_base import CloudAccessContext

    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListDirectories":
            raise RuntimeError("ServiceUnavailable: cloudsso 503")
        # If grant_access falls through to parent RAM, AttachPolicyToUser
        # would be called — assert that's NOT the case.
        if action == "AttachPolicyToUser":
            raise AssertionError(
                "must not fall through to parent RAM for CloudSSO role_key"
            )
        return {}

    ctx = CloudAccessContext(
        principal_email="x@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="t", display_name="x",
        ),
        role_key="poweruser",  # bare CloudSSO name, no ':'
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.grant_access(ctx)

    assert result.ok is False
    msg = (result.error_message or "").lower()
    assert "transient" in msg or "unreachable" in msg or "retry" in msg


@pytest.mark.asyncio
async def test_revoke_access_listdirectories_transient_does_NOT_mark_already_absent():
    """Round-2 code review caught a LATENT pre-existing bug exposed
    by the new transport-error semantics: revoke_access on a
    ListDirectories transient failure used to silently return
    ``ok=True already_absent=True`` — a false success that would
    drop a still-live CloudSSO assignment from Helix's records
    while leaving it active cloud-side.

    Now: transport failure → ok=False with retry message, so the
    offboarding flow retries instead of declaring victory.
    """
    from app.integrations.driver_base import CloudAccessContext

    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListDirectories":
            raise RuntimeError("ServiceUnavailable: cloudsso 503")
        # If revoke proceeds past directory resolution, we'd see
        # DeleteAccessAssignment — assert that's NOT the case for
        # a transient ListDirectories failure.
        if action == "DeleteAccessAssignment":
            raise AssertionError(
                "must not proceed to DeleteAccessAssignment on "
                "ListDirectories transient failure"
            )
        raise AssertionError(action)

    ctx = CloudAccessContext(
        principal_email="x@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="t", display_name="x",
        ),
        role_key="poweruser",
        existing_external_ref="sso::ac-power::u-x::t",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.revoke_access(ctx)

    assert result.ok is False
    # CRITICAL — must NOT mark already_absent on a transient failure
    assert result.already_absent is False
    msg = (result.error_message or "").lower()
    assert "retry" in msg or "unavailable" in msg


@pytest.mark.asyncio
async def test_revoke_access_no_directory_genuinely_returns_already_absent():
    """Sanity check: when ListDirectories succeeds but returns an
    empty list (legitimate no-SSO tenant), revoke_access still
    returns ``ok=True already_absent=True`` — the assignment
    cannot exist on a tenant with no directory."""
    from app.integrations.driver_base import CloudAccessContext

    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": []}  # genuinely empty success
        raise AssertionError(action)

    ctx = CloudAccessContext(
        principal_email="x@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="t", display_name="x",
        ),
        role_key="poweruser",
        existing_external_ref="sso::ac::u::t",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.revoke_access(ctx)

    assert result.ok is True
    assert result.already_absent is True


@pytest.mark.asyncio
async def test_list_scope_grants_listdirectories_transient_raises_not_fallback():
    """Round-4 logic review: ListDirectories transient failure
    during grants discovery must RAISE rather than fall back to
    parent RAM. Fall-back returns near-zero rows on a CloudSSO
    tenant (RAM users are mostly service accounts); the
    reconciler would silently flip live CloudSSO grants to
    ``status='revoked' revoke_reason='drift_removed'`` for
    scopes with fewer than ~10 active grants (50%-floor doesn't
    save them). Raising lets the sync framework's per-scope
    try/except skip drift-sweep cleanly.
    """
    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListDirectories":
            raise RuntimeError("ServiceUnavailable: cloudsso 503")
        if action in ("ListUsers", "ListPoliciesForUser"):
            raise AssertionError("must not fall back to parent RAM")
        return {}

    scope = CloudScopeRef(
        scope_type="aliyun_account", external_id="t", display_name="x",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)), \
         pytest.raises(RuntimeError, match="ServiceUnavailable"):
        await driver.list_scope_grants(scope)


@pytest.mark.asyncio
async def test_grant_access_ram_shape_rolekey_on_cloudsso_tenant_returns_shape_mismatch():
    """Round-4 logic review: a stale/legacy RAM-shape role_key
    ('System:ReadOnlyAccess') on a CloudSSO-managed tenant must
    surface a clear shape-mismatch error, not the misleading
    "AccessConfiguration not found — renamed in console" message
    that the resolver's strict name match would otherwise produce.
    """
    from app.integrations.driver_base import CloudAccessContext

    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-test"}]}
        # If we proceed past the shape check we'd hit
        # ListAccessConfigurations / ListUsers / AttachPolicyToUser
        # — none should fire.
        if action in (
            "ListAccessConfigurations",
            "ListUsers",
            "AttachPolicyToUser",
            "CreateAccessAssignment",
        ):
            raise AssertionError(
                f"must not reach {action} on RAM-shape role_key + "
                "CloudSSO tenant"
            )
        return {}

    ctx = CloudAccessContext(
        principal_email="x@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="t", display_name="x",
        ),
        # RAM-shape role_key on a tenant that DOES have CloudSSO
        # directory provisioned — must NOT silently RAM-attach.
        role_key="System:ReadOnlyAccess",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.grant_access(ctx)

    assert result.ok is False
    msg = (result.error_message or "").lower()
    assert "ram-shape" in msg or "shape" in msg or "stale" in msg
    # CRITICAL: must NOT misroute to "renamed in console"
    assert "renamed" not in msg


@pytest.mark.asyncio
async def test_grant_access_listdirectories_transient_with_ram_role_key_returns_transient():
    """Even a RAM-shape role_key (`Custom:X`) must NOT fall through
    to parent RAM grant when ListDirectories transient-fails. We
    cannot know whether this tenant is CloudSSO or RAM-only
    without a successful directory lookup, so the safe default is
    to retry rather than make a RAM attempt that could create a
    stray RAM user on a CloudSSO-managed tenant. Round-2 fix."""
    from app.integrations.driver_base import CloudAccessContext

    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListDirectories":
            raise RuntimeError("ServiceUnavailable: cloudsso 503")
        if action in ("CreateUser", "AttachPolicyToUser"):
            raise AssertionError(
                "must not attempt RAM grant on ListDirectories transient"
            )
        return {}

    ctx = CloudAccessContext(
        principal_email="x@pingcap.com",
        principal_display_name="x",
        scope=CloudScopeRef(
            scope_type="aliyun_account", external_id="t", display_name="x",
        ),
        role_key="System:ReadOnlyAccess",  # RAM-shape
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        result = await driver.grant_access(ctx)

    assert result.ok is False
    msg = (result.error_message or "").lower()
    assert "retry" in msg or "unavailable" in msg


@pytest.mark.asyncio
async def test_resolve_user_id_paginates_via_is_truncated():
    """Defensive pagination on ListUsers — if Aliyun ever ignores the
    Filter or the result set spans pages, the loop must consume
    every page rather than miss the user on page 2+. Round-1 code
    review flagged the absent IsTruncated handling.
    """
    driver = _make_driver()
    call_count = {"n": 0}

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListUsers":
            call_count["n"] += 1
            if call_count["n"] == 1:
                return {
                    "Users": [{"UserId": "u-other", "UserName": "other@x.com"}],
                    "IsTruncated": True,
                    "NextToken": "ntk-2",
                }
            return {
                "Users": [{"UserId": "u-target", "UserName": "target@x.com"}],
                "IsTruncated": False,
            }
        return {}

    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        uid = await driver._sso_resolve_user_id("d-test", "target@x.com")

    assert uid == "u-target"
    assert call_count["n"] == 2


# ── validate_request_role (submit-time shape pre-flight) ──────────────
# Closes the gap where a stale RAM-shape catalog row passed submit,
# routed through the owner's approval, and only failed at grant time
# (the approver, not the requester, saw the toast). The pre-flight
# mirrors grant_access's shape decision tree without the actual grant.


@pytest.mark.asyncio
async def test_validate_request_role_rejects_ram_shape_on_cloudsso_tenant():
    """RAM-shape key + CloudSSO directory present → reject at submit."""
    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-test"}]}
        raise AssertionError(f"must not call {action} during pre-flight")

    scope = CloudScopeRef(
        scope_type="aliyun_account", external_id="t", display_name="x",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        msg = await driver.validate_request_role(scope, "System:ReadOnlyAccess")
    assert msg is not None
    low = msg.lower()
    assert "ram-shape" in low or "stale" in low


@pytest.mark.asyncio
async def test_validate_request_role_allows_cloudsso_name_on_cloudsso_tenant():
    """Bare CloudSSO permission-set name + directory present → allowed."""
    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListDirectories":
            return {"Directories": [{"DirectoryId": "d-test"}]}
        raise AssertionError(f"must not call {action} during pre-flight")

    scope = CloudScopeRef(
        scope_type="aliyun_account", external_id="t", display_name="x",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        msg = await driver.validate_request_role(scope, "DBaaS-DevUser-Role")
    assert msg is None


@pytest.mark.asyncio
async def test_validate_request_role_fails_open_on_transient():
    """ListDirectories transient-fail → return None (grant path stays
    the authoritative gate; never block a legit request on infra blip)."""
    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        if action == "ListDirectories":
            raise RuntimeError("ServiceUnavailable: cloudsso 503")
        raise AssertionError(f"must not call {action} during pre-flight")

    scope = CloudScopeRef(
        scope_type="aliyun_account", external_id="t", display_name="x",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        msg = await driver.validate_request_role(scope, "System:ReadOnlyAccess")
    assert msg is None


@pytest.mark.asyncio
async def test_validate_request_role_non_aliyun_account_scope_is_permissive():
    """Non-``aliyun_account`` scope types (legacy RAM single-account) are
    not CloudSSO-shaped — never reach the directory probe, always allow."""
    driver = _make_driver()

    async def fake_acs3(*, host, version, action, method="POST", params=None, body=None):
        raise AssertionError("must not call any API for non-aliyun_account")

    scope = CloudScopeRef(
        scope_type="aliyun_ram_user", external_id="t", display_name="x",
    )
    with patch.object(driver, "_acs3_call", new=AsyncMock(side_effect=fake_acs3)):
        msg = await driver.validate_request_role(scope, "System:ReadOnlyAccess")
    assert msg is None