"""Regression tests for Tencent Cloud CAM grant / revoke driver paths.

These tests pin the **group-aware** revoke contract — the corrected
2026-06-02 model after the 2026-06-01 incident on lifu.wu@ /
chujie@:

  * lifu.wu@'s 9 grants and chujie@'s 4 grants on Tencent were ALL
    GROUP-INHERITED (CAM ``管理员`` / ``账单计费`` / ``EE`` groups).
    The pre-fix driver called ``DetachUserPolicy`` against them →
    Tencent returns RequestId but does nothing (group-inherited
    attachments aren't undone by per-policy detach) → cloud-side
    permissions survived 2 months despite Helix marking the grants
    revoked.

  * Two distinct CAM identifier fields:
      - ``Uin``: sub-account UIN (plain number), used by
        ``DetachUserPolicy.DetachUin`` / ``AttachUserPolicy.AttachUin``
        / ``ListAttachedUserAllPolicies.TargetUin``.
      - ``Uid``: CAM-internal id (different number), used by
        ``ListGroupsForUser.Uid`` / ``RemoveUserFromGroup.Info[].Uid``.
    Mixing the two surfaces as ``ResourceNotFound.UserNotExist`` even
    when the user plainly exists; the driver MUST harvest both via
    GetUser and route to the right field.

  * SDK 3.1.79's ``DetachUserPolicyRequest`` / ``AttachUserPolicyRequest``
    only carry ``PolicyId`` + ``DetachUin`` / ``AttachUin`` respectively.
    Setting ``DetachUserName`` / ``AttachUserName`` becomes an ad-hoc
    Python attribute that serializes mangled (``EtachUserName`` —
    leading char dropped) and Tencent rejects with
    ``UnknownParameter``. Never set them.

Invariants pinned below:
  0. list_scope_roles fails closed on any incomplete CAM policy page.
  1. revoke_access on a group-inherited policy → RemoveUserFromGroup
     (Uid, GroupId) per source group; DetachUserPolicy is NEVER fired.
  2. revoke_access on a direct-attach policy → DetachUserPolicy(Uin,
     PolicyId) with NO DetachUserName.
  3. revoke_access on a not-attached policy → already_absent without
     mutating the cloud.
  4. revoke_access on a missing user → already_absent without listing.
  5. revoke_account chain uses Uid (NOT Uin) on ListGroupsForUser +
     RemoveUserFromGroup.
  6. grant_access (AttachUserPolicy) carries AttachUin from AddUser
     response or GetUser fallback, NEVER AttachUserName.
"""
from __future__ import annotations

from unittest.mock import patch

import pytest

from app.domains.cmdb.providers.exceptions import ProviderAuthError
from app.integrations.driver_base import (
    CloudAccessContext,
    CloudAccountRevokeContext,
    CloudScopeRef,
)
from app.integrations.drivers.tencent_cloud import TencentCloudDriver

_CFG = {
    "secret_id": "AK-test",
    "secret_key": "SK-test",
    "region": "ap-guangzhou",
    "account_mode": "iam",
}


def _driver() -> TencentCloudDriver:
    return TencentCloudDriver(integration_id="tc-test", config=_CFG)


class _Resp:
    """Tiny response holder — every attr is whatever we feed it."""

    def __init__(self, **fields):
        self.__dict__.update(fields)


def _scope() -> CloudScopeRef:
    return CloudScopeRef(
        scope_type="standalone_iam",
        external_id="123456",
        display_name="tencent-test",
    )


# ── account identity / credential-safe probe ────────────────────────


@pytest.mark.asyncio
async def test_list_scopes_keys_subuser_credentials_to_owner_uin() -> None:
    driver = _driver()
    with patch.object(
        driver,
        "_fetch_account_identity_sync",
        return_value=("credential-subuser-uin", "stable-owner-uin", 1250001),
    ):
        scopes = await driver.list_scopes()

    assert len(scopes) == 1
    assert scopes[0].external_id == "stable-owner-uin"
    assert scopes[0].raw == {
        "mode": "iam",
        "owner_uin": "stable-owner-uin",
        "credential_uin": "credential-subuser-uin",
    }


def test_account_identity_missing_owner_uin_fails_closed() -> None:
    driver = _driver()
    with (
        patch.object(
            driver, "_cam_client",
            return_value=_Resp(GetUserAppId=lambda _req: None),
        ),
        patch.object(
            driver, "_call",
            return_value=_Resp(Uin="subuser-uin", OwnerUin="", AppId=1),
        ),
        pytest.raises(ProviderAuthError, match="omitted Uin or OwnerUin"),
    ):
        driver._fetch_account_identity_sync()


def test_connection_redacts_tencent_named_secrets() -> None:
    secret = "plain-format-secret-value"
    driver = _driver()
    with patch.object(
        driver,
        "_fetch_account_identity_sync",
        side_effect=RuntimeError(
            f"SecretKey={secret} Signature=0123456789abcdef",
        ),
    ):
        result = driver._test_connection_sync()

    assert result.ok is False
    assert secret not in result.message
    assert "0123456789abcdef" not in result.message
    assert "<redacted>" in result.message


def test_account_identity_error_log_is_redacted(caplog) -> None:
    secret = "plain-format-secret-value"
    driver = _driver()
    with (
        patch.object(
            driver, "_cam_client",
            return_value=_Resp(GetUserAppId=lambda _req: None),
        ),
        patch.object(
            driver, "_call",
            side_effect=RuntimeError(f"secret_key={secret}"),
        ),
        pytest.raises(ProviderAuthError),
    ):
        driver._fetch_account_identity_sync()

    assert secret not in caplog.text
    assert "<redacted>" in caplog.text


# ── list_scope_roles (authoritative pagination) ─────────────────────


def test_list_policies_mid_pagination_failure_is_not_partial_success() -> None:
    """A healthy page 1 followed by a failed page 2 must raise.

    Returning the 200-row prefix would let role sync treat the missing tail
    as authoritative and soft-delete real CAM policies from Helix.
    """
    from tencentcloud.common.exception.tencent_cloud_sdk_exception import (
        TencentCloudSDKException,
    )

    first_page = [
        _Resp(
            PolicyName=f"policy-{index}",
            PolicyId=index,
            Type=1,
            Description="",
        )
        for index in range(200)
    ]
    requested_pages: list[int] = []

    def _call(_fn, req):
        requested_pages.append(req.Page)
        if req.Page == 1:
            return _Resp(List=first_page, TotalNum=220)
        raise TencentCloudSDKException(
            "RequestLimitExceeded", "throttled", "request-id",
        )

    driver = _driver()
    with (
        patch.object(
            driver, "_cam_client",
            return_value=_Resp(ListPolicies=lambda _req: None),
        ),
        patch.object(driver, "_call", side_effect=_call),
    ):
        with pytest.raises(RuntimeError, match=r"ListPolicies page=2 failed"):
            driver._list_policies_sync(_scope(), set())

    assert requested_pages == [1, 2]


def test_list_policies_premature_empty_page_is_not_partial_success() -> None:
    """An empty page cannot terminate while TotalNum says rows remain."""
    first_page = [
        _Resp(
            PolicyName=f"policy-{index}", PolicyId=index,
            Type=1, Description="",
        )
        for index in range(200)
    ]

    def _call(_fn, req):
        if req.Page == 1:
            return _Resp(List=first_page, TotalNum=220)
        return _Resp(List=[], TotalNum=220)

    driver = _driver()
    with (
        patch.object(
            driver, "_cam_client",
            return_value=_Resp(ListPolicies=lambda _req: None),
        ),
        patch.object(driver, "_call", side_effect=_call),
    ):
        with pytest.raises(RuntimeError, match="empty page"):
            driver._list_policies_sync(_scope(), set())


def test_list_policies_complete_two_page_snapshot_succeeds() -> None:
    """The fail-closed guards must preserve a valid multi-page catalog."""
    pages = {
        1: [
            _Resp(
                PolicyName=f"policy-{index}", PolicyId=index,
                Type=1, Description="",
            )
            for index in range(200)
        ],
        2: [
            _Resp(
                PolicyName=f"policy-{index}", PolicyId=index,
                Type=1, Description="",
            )
            for index in range(200, 220)
        ],
    }

    def _call(_fn, req):
        return _Resp(List=pages[req.Page], TotalNum=220)

    driver = _driver()
    with (
        patch.object(
            driver, "_cam_client",
            return_value=_Resp(ListPolicies=lambda _req: None),
        ),
        patch.object(driver, "_call", side_effect=_call),
    ):
        roles = driver._list_policies_sync(_scope(), set())

    assert len(roles) == 220
    assert len({role.role_key for role in roles}) == 220


def test_list_policies_repeated_page_is_not_partial_success() -> None:
    """An ignored Page parameter must not inflate observed role count."""
    first_page = [
        _Resp(
            PolicyName=f"policy-{index}", PolicyId=index,
            Type=1, Description="",
        )
        for index in range(200)
    ]

    def _call(_fn, req):
        return _Resp(
            List=first_page if req.Page == 1 else [first_page[0]],
            TotalNum=201,
        )

    driver = _driver()
    with (
        patch.object(
            driver, "_cam_client",
            return_value=_Resp(ListPolicies=lambda _req: None),
        ),
        patch.object(driver, "_call", side_effect=_call),
    ):
        with pytest.raises(RuntimeError, match="overlapping/repeated"):
            driver._list_policies_sync(_scope(), set())


def test_list_policies_page_ceiling_is_not_partial_success() -> None:
    """A corrupt TotalNum cannot pin a worker or return its prefix."""
    def _call(_fn, req):
        return _Resp(
            List=[
                _Resp(
                    PolicyName=f"policy-{req.Page}-{index}",
                    PolicyId=req.Page * 1000 + index,
                    Type=1,
                    Description="",
                )
                for index in range(200)
            ],
            TotalNum=601,
        )

    driver = _driver()
    with (
        patch.object(
            driver, "_cam_client",
            return_value=_Resp(ListPolicies=lambda _req: None),
        ),
        patch.object(driver, "_call", side_effect=_call),
        patch("app.integrations.drivers.tencent_cloud._MAX_CAM_POLICY_PAGES", 2),
    ):
        with pytest.raises(RuntimeError, match="2-page safety ceiling"):
            driver._list_policies_sync(_scope(), set())


def test_list_policies_rejects_total_change_during_pagination() -> None:
    first_page = [
        _Resp(
            PolicyName=f"policy-{index}", PolicyId=index,
            Type=1, Description="",
        )
        for index in range(200)
    ]

    def _call(_fn, req):
        if req.Page == 1:
            return _Resp(List=first_page, TotalNum=201)
        return _Resp(List=[_Resp(
            PolicyName="policy-200", PolicyId=200, Type=1, Description="",
        )], TotalNum=202)

    driver = _driver()
    with (
        patch.object(
            driver, "_cam_client",
            return_value=_Resp(ListPolicies=lambda _req: None),
        ),
        patch.object(driver, "_call", side_effect=_call),
    ):
        with pytest.raises(RuntimeError, match="changed TotalNum"):
            driver._list_policies_sync(_scope(), set())


def test_list_policies_rejects_short_nonfinal_page() -> None:
    driver = _driver()
    with (
        patch.object(
            driver, "_cam_client",
            return_value=_Resp(ListPolicies=lambda _req: None),
        ),
        patch.object(driver, "_call", return_value=_Resp(
            List=[_Resp(
                PolicyName="policy-1", PolicyId=1, Type=1, Description="",
            )],
            TotalNum=201,
        )),
    ):
        with pytest.raises(RuntimeError, match="short page"):
            driver._list_policies_sync(_scope(), set())


def test_list_policies_rejects_missing_policy_name() -> None:
    driver = _driver()
    with (
        patch.object(
            driver, "_cam_client",
            return_value=_Resp(ListPolicies=lambda _req: None),
        ),
        patch.object(driver, "_call", return_value=_Resp(
            List=[_Resp(PolicyName="", PolicyId=1, Type=1, Description="")],
            TotalNum=1,
        )),
    ):
        with pytest.raises(RuntimeError, match="without PolicyName"):
            driver._list_policies_sync(_scope(), set())


def test_list_policies_rejects_more_unique_rows_than_total() -> None:
    driver = _driver()
    with (
        patch.object(
            driver, "_cam_client",
            return_value=_Resp(ListPolicies=lambda _req: None),
        ),
        patch.object(driver, "_call", return_value=_Resp(
            List=[
                _Resp(PolicyName="p1", PolicyId=1, Type=1, Description=""),
                _Resp(PolicyName="p2", PolicyId=2, Type=1, Description=""),
            ],
            TotalNum=1,
        )),
    ):
        with pytest.raises(RuntimeError, match="2 unique rows.*TotalNum=1"):
            driver._list_policies_sync(_scope(), set())


def test_list_scope_grants_rejects_short_attached_policy_page() -> None:
    driver = _driver()
    user = _Resp(
        Name="alice@example.com",
        Email="alice@example.com",
        Uin=123,
    )

    def _call(_fn, req):
        if req.__class__.__name__ == "ListUsersRequest":
            return _Resp(Data=[user])
        return _Resp(
            PolicyList=[_Resp(PolicyName="p1", PolicyId=1)],
            TotalNum=2,
        )

    with (
        patch.object(
            driver, "_cam_client",
            return_value=_Resp(
                ListUsers=lambda _req: None,
                ListAttachedUserAllPolicies=lambda _req: None,
            ),
        ),
        patch.object(driver, "_call", side_effect=_call),
    ):
        with pytest.raises(RuntimeError, match="short page"):
            driver._list_scope_grants_sync(_scope())


def test_list_scope_grants_deduplicates_same_page_attachment_paths() -> None:
    """CAM can list one effective policy for both direct and group paths."""
    driver = _driver()
    user = _Resp(Name="alice@example.com", Email="alice@example.com", Uin=123)

    def _call(_fn, req):
        if req.__class__.__name__ == "ListUsersRequest":
            return _Resp(Data=[user])
        assert req.AttachType == 0
        return _Resp(
            PolicyList=[
                _Resp(PolicyName="p1", PolicyId=1),
                _Resp(PolicyName="p1", PolicyId=1),
            ],
            TotalNum=2,
        )

    with (
        patch.object(
            driver, "_cam_client",
            return_value=_Resp(
                ListUsers=lambda _req: None,
                ListAttachedUserAllPolicies=lambda _req: None,
            ),
        ),
        patch.object(driver, "_call", side_effect=_call),
    ):
        grants = driver._list_scope_grants_sync(_scope())

    assert [(grant.role_key, grant.external_ref) for grant in grants] == [
        ("p1", "alice@example.com::1::p1"),
    ]


def test_list_scope_grants_emits_bare_handle_for_user_without_email() -> None:
    """A CAM user with no ``Email`` and no ``@`` in its name must not take
    the tenant's whole snapshot down with it.

    Prod 2026-08-03 → 2026-08-12: ``cmdb.grants.tencent-primary`` failed on
    every one of 18 consecutive runs because the service account
    ``tiup-server`` has neither. The Tencent members list froze at its
    2026-08-01 state while the page kept presenting it as current. The
    partial-snapshot guard this replaced was aimed at a snapshot that
    OMITS a user (which the drift sweep reads as "access revoked") — but
    emitting the bare handle keeps the snapshot complete, and an
    unattributable principal is then counted + named by the resolver
    instead of crashing the run.
    """
    driver = _driver()
    users = [
        _Resp(Name="tiup-server", Email="", Uin=100036767862),
        _Resp(Name="zhuoran.li", Email=None, Uin=100040362773),
        _Resp(Name="Lilin@PingCAP.com", Email="", Uin=100040386673),
    ]

    def _call(_fn, req):
        if req.__class__.__name__ == "ListUsersRequest":
            return _Resp(Data=users)
        return _Resp(
            PolicyList=[_Resp(PolicyName="p1", PolicyId=1)],
            TotalNum=1,
        )

    with (
        patch.object(
            driver, "_cam_client",
            return_value=_Resp(
                ListUsers=lambda _req: None,
                ListAttachedUserAllPolicies=lambda _req: None,
            ),
        ),
        patch.object(driver, "_call", side_effect=_call),
    ):
        grants = driver._list_scope_grants_sync(_scope())

    # Every user survives — no user is silently dropped either.
    assert [g.principal_email for g in grants] == [
        "tiup-server", "zhuoran.li", "lilin@pingcap.com",
    ]


def test_list_scope_grants_rejects_user_without_uin() -> None:
    driver = _driver()
    with (
        patch.object(
            driver, "_cam_client",
            return_value=_Resp(ListUsers=lambda _req: None),
        ),
        patch.object(driver, "_call", return_value=_Resp(Data=[_Resp(
            Name="alice@example.com",
            Email="alice@example.com",
            Uin=None,
        )])),
    ):
        with pytest.raises(RuntimeError, match="without a valid Uin"):
            driver._list_scope_grants_sync(_scope())


# ── revoke_access (group-inherited path) ─────────────────────────────


@pytest.mark.asyncio
async def test_revoke_access_group_inherited_calls_remove_from_group() -> None:
    """A policy attached via group → RemoveUserFromGroup(Uid, GroupId),
    NO DetachUserPolicy. Covers the 2026-06-01 lifu.wu@ root cause."""
    detach_calls = 0
    remove_calls: list[dict] = []

    class _StubClient:
        def GetUser(self, req):  # noqa: N802
            return _Resp(Uid=22001053, Uin=100042853879)

        def ListAttachedUserAllPolicies(self, req):  # noqa: N802
            assert req.TargetUin == 100042853879
            # AttachType=2 → group-inherited; we return the matching
            # policy with its source group.
            if req.AttachType == 2:
                return _Resp(PolicyList=[
                    _Resp(
                        PolicyId=1,
                        PolicyName="AdministratorAccess",
                        Groups=[{"GroupId": 322158, "GroupName": "管理员"}],
                    ),
                ], TotalNum=1)
            # AttachType=1 would be direct attach — not exercised here.
            raise AssertionError(
                "direct-attach list should NOT be hit when the policy "
                "is found group-inherited",
            )

        def RemoveUserFromGroup(self, req):  # noqa: N802
            remove_calls.append({
                "Info": list(req.Info),
            })
            return _Resp()

        def DetachUserPolicy(self, req):  # noqa: N802
            nonlocal detach_calls
            detach_calls += 1
            return _Resp()

    driver = _driver()
    with patch.object(
        TencentCloudDriver, "_cam_client", return_value=_StubClient(),
    ):
        result = await driver.revoke_access(CloudAccessContext(
            principal_email="lifu.wu@pingcap.com",
            principal_display_name="Lifu Wu",
            scope=_scope(),
            role_key="AdministratorAccess",
            role_display_name="AdministratorAccess",
            existing_external_ref="lifu.wu@pingcap.com::1::AdministratorAccess",
        ))

    assert result.ok, result.error_message
    assert detach_calls == 0, (
        "DetachUserPolicy must NOT fire on group-inherited grants"
    )
    assert len(remove_calls) == 1
    assert remove_calls[0]["Info"] == [{"Uid": 22001053, "GroupId": 322158}]


@pytest.mark.asyncio
async def test_revoke_access_group_with_multiple_source_groups() -> None:
    """A policy that's group-inherited from MORE THAN ONE group on the
    same user → RemoveUserFromGroup fires for every source group."""
    remove_calls: list[int] = []

    class _StubClient:
        def GetUser(self, req):  # noqa: N802
            return _Resp(Uid=22001053, Uin=100042853879)

        def ListAttachedUserAllPolicies(self, req):  # noqa: N802
            if req.AttachType == 2:
                return _Resp(PolicyList=[
                    _Resp(
                        PolicyId=99,
                        PolicyName="X",
                        Groups=[
                            {"GroupId": 100, "GroupName": "A"},
                            {"GroupId": 200, "GroupName": "B"},
                        ],
                    ),
                ], TotalNum=1)
            return _Resp(PolicyList=[], TotalNum=0)

        def RemoveUserFromGroup(self, req):  # noqa: N802
            remove_calls.append(req.Info[0]["GroupId"])
            return _Resp()

    driver = _driver()
    with patch.object(
        TencentCloudDriver, "_cam_client", return_value=_StubClient(),
    ):
        result = await driver.revoke_access(CloudAccessContext(
            principal_email="multi@pingcap.com",
            principal_display_name="Multi",
            scope=_scope(),
            role_key="99",
            role_display_name="X",
            existing_external_ref="multi@pingcap.com::99::X",
        ))

    assert result.ok
    assert sorted(remove_calls) == [100, 200]


@pytest.mark.asyncio
async def test_revoke_access_finds_group_policy_on_second_page() -> None:
    remove_calls: list[int] = []

    class _StubClient:
        def GetUser(self, req):  # noqa: N802
            return _Resp(Uid=22, Uin=100)

        def ListAttachedUserAllPolicies(self, req):  # noqa: N802
            assert req.AttachType == 2
            if req.Page == 1:
                return _Resp(
                    PolicyList=[
                        _Resp(
                            PolicyId=index,
                            PolicyName=f"other-{index}",
                            Groups=[],
                        )
                        for index in range(200)
                    ],
                    TotalNum=201,
                )
            return _Resp(
                PolicyList=[_Resp(
                    PolicyId=999,
                    PolicyName="late-policy",
                    Groups=[{"GroupId": 77, "GroupName": "late"}],
                )],
                TotalNum=201,
            )

        def RemoveUserFromGroup(self, req):  # noqa: N802
            remove_calls.append(req.Info[0]["GroupId"])
            return _Resp()

    driver = _driver()
    with patch.object(driver, "_cam_client", return_value=_StubClient()):
        result = await driver.revoke_access(CloudAccessContext(
            principal_email="late@pingcap.com",
            principal_display_name="Late",
            scope=_scope(),
            role_key="999",
            role_display_name="late-policy",
            existing_external_ref="late@pingcap.com::999::late-policy",
        ))

    assert result.ok
    assert remove_calls == [77]


@pytest.mark.asyncio
async def test_revoke_access_second_page_failure_is_not_already_absent() -> None:
    from tencentcloud.common.exception.tencent_cloud_sdk_exception import (
        TencentCloudSDKException,
    )

    class _StubClient:
        def GetUser(self, req):  # noqa: N802
            return _Resp(Uid=22, Uin=100)

        def ListAttachedUserAllPolicies(self, req):  # noqa: N802
            if req.Page == 1:
                return _Resp(
                    PolicyList=[
                        _Resp(
                            PolicyId=index,
                            PolicyName=f"other-{index}",
                            Groups=[],
                        )
                        for index in range(200)
                    ],
                    TotalNum=201,
                )
            raise TencentCloudSDKException(
                "RequestLimitExceeded", "SecretKey=leaked-secret-value", "rid",
            )

    driver = _driver()
    with patch.object(driver, "_cam_client", return_value=_StubClient()):
        result = await driver.revoke_access(CloudAccessContext(
            principal_email="late@pingcap.com",
            principal_display_name="Late",
            scope=_scope(),
            role_key="999",
            role_display_name="late-policy",
            existing_external_ref="late@pingcap.com::999::late-policy",
        ))

    assert result.ok is False
    assert result.already_absent is False
    assert "leaked-secret-value" not in (result.error_message or "")
    assert "page=2" in (result.error_message or "")


@pytest.mark.asyncio
async def test_revoke_policy_lookup_failure_is_not_already_absent() -> None:
    from tencentcloud.common.exception.tencent_cloud_sdk_exception import (
        TencentCloudSDKException,
    )

    class _StubClient:
        def ListPolicies(self, req):  # noqa: N802
            raise TencentCloudSDKException(
                "InternalError", "provider unavailable", "rid",
            )

    driver = _driver()
    with patch.object(driver, "_cam_client", return_value=_StubClient()):
        result = await driver.revoke_access(CloudAccessContext(
            principal_email="lookup@pingcap.com",
            principal_display_name="Lookup",
            scope=_scope(),
            role_key="named-policy",
            role_display_name="named-policy",
            existing_external_ref=None,
        ))

    assert result.ok is False
    assert result.already_absent is False
    assert "ListPolicies" in (result.error_message or "")


# ── revoke_access (direct-attach path) ───────────────────────────────


@pytest.mark.asyncio
async def test_revoke_access_direct_attach_calls_detach_user_policy() -> None:
    """A directly-attached policy → DetachUserPolicy(Uin, PolicyId) with
    NO ad-hoc DetachUserName field set."""
    detach_req_seen: dict = {}
    remove_calls = 0

    class _StubClient:
        def GetUser(self, req):  # noqa: N802
            return _Resp(Uid=22001053, Uin=100042853879)

        def ListAttachedUserAllPolicies(self, req):  # noqa: N802
            if req.AttachType == 2:
                return _Resp(PolicyList=[], TotalNum=0)  # not group-inherited
            if req.AttachType == 1:
                return _Resp(  # direct
                    PolicyList=[_Resp(PolicyId=42, PolicyName="Y")],
                    TotalNum=1,
                )
            return _Resp(PolicyList=[], TotalNum=0)

        def DetachUserPolicy(self, req):  # noqa: N802
            detach_req_seen["PolicyId"] = req.PolicyId
            detach_req_seen["DetachUin"] = req.DetachUin
            # Verify the buggy field isn't set
            detach_req_seen["has_DetachUserName"] = hasattr(
                req, "DetachUserName",
            ) and "DetachUserName" in req.__dict__
            return _Resp()

        def RemoveUserFromGroup(self, req):  # noqa: N802
            nonlocal remove_calls
            remove_calls += 1
            return _Resp()

    driver = _driver()
    with patch.object(
        TencentCloudDriver, "_cam_client", return_value=_StubClient(),
    ):
        result = await driver.revoke_access(CloudAccessContext(
            principal_email="direct@pingcap.com",
            principal_display_name="Direct",
            scope=_scope(),
            role_key="42",
            role_display_name="Y",
            existing_external_ref="direct@pingcap.com::42::Y",
        ))

    assert result.ok
    assert remove_calls == 0
    assert detach_req_seen["PolicyId"] == 42
    assert detach_req_seen["DetachUin"] == 100042853879
    assert detach_req_seen["has_DetachUserName"] is False, (
        "DetachUserName must NEVER be set — SDK 3.1.79 has no such field; "
        "ad-hoc attrs serialize mangled as 'EtachUserName' and the API "
        "rejects with UnknownParameter."
    )


# ── revoke_access (idempotent already_absent paths) ──────────────────


@pytest.mark.asyncio
async def test_revoke_access_user_missing_is_already_absent() -> None:
    """GetUser → UserNotExist → already_absent, no other calls."""
    from tencentcloud.common.exception.tencent_cloud_sdk_exception import (
        TencentCloudSDKException,
    )
    extra_calls = 0

    class _StubClient:
        def GetUser(self, req):  # noqa: N802
            raise TencentCloudSDKException(
                "UserNotExist", "user gone", "rid",
            )

        def __getattr__(self, name):
            def _f(*a, **k):
                nonlocal extra_calls
                extra_calls += 1
            return _f

    driver = _driver()
    with patch.object(
        TencentCloudDriver, "_cam_client", return_value=_StubClient(),
    ):
        result = await driver.revoke_access(CloudAccessContext(
            principal_email="ghost@pingcap.com",
            principal_display_name="Ghost",
            scope=_scope(),
            role_key="1",
            role_display_name="X",
            existing_external_ref="ghost::1::X",
        ))

    assert result.ok and result.already_absent
    assert extra_calls == 0


@pytest.mark.asyncio
async def test_revoke_access_malformed_user_identity_fails_closed() -> None:
    """A successful GetUser response without both identifiers is not absence."""
    extra_calls = 0

    class _StubClient:
        def GetUser(self, req):  # noqa: N802
            return _Resp(Uid=123, Uin=None)

        def __getattr__(self, name):
            def _f(*a, **k):
                nonlocal extra_calls
                extra_calls += 1
            return _f

    driver = _driver()
    with patch.object(
        TencentCloudDriver, "_cam_client", return_value=_StubClient(),
    ):
        result = await driver.revoke_access(CloudAccessContext(
            principal_email="ambiguous@pingcap.com",
            principal_display_name="Ambiguous",
            scope=_scope(),
            role_key="1",
            role_display_name="X",
            existing_external_ref="ambiguous::1::X",
        ))

    assert result.ok is False
    assert result.already_absent is False
    assert "missing identifiers" in (result.error_message or "")
    assert extra_calls == 0


@pytest.mark.asyncio
async def test_revoke_access_not_attached_anywhere_is_already_absent() -> None:
    """Policy doesn't appear in either group-inherited or direct list
    → already_absent without any detach / remove API call."""
    detach_calls = 0
    remove_calls = 0

    class _StubClient:
        def GetUser(self, req):  # noqa: N802
            return _Resp(Uid=1, Uin=2)

        def ListAttachedUserAllPolicies(self, req):  # noqa: N802
            return _Resp(PolicyList=[], TotalNum=0)

        def DetachUserPolicy(self, req):  # noqa: N802
            nonlocal detach_calls
            detach_calls += 1
            return _Resp()

        def RemoveUserFromGroup(self, req):  # noqa: N802
            nonlocal remove_calls
            remove_calls += 1
            return _Resp()

    driver = _driver()
    with patch.object(
        TencentCloudDriver, "_cam_client", return_value=_StubClient(),
    ):
        result = await driver.revoke_access(CloudAccessContext(
            principal_email="x@pingcap.com",
            principal_display_name="X",
            scope=_scope(),
            role_key="999",
            role_display_name="X",
            existing_external_ref="x::999::X",
        ))

    assert result.ok and result.already_absent
    assert detach_calls == 0
    assert remove_calls == 0


# ── revoke_account (delete chain uses Uid for group ops) ─────────────


@pytest.mark.asyncio
async def test_revoke_account_uses_uid_not_uin_for_group_ops() -> None:
    """``ListGroupsForUser`` + ``RemoveUserFromGroup`` MUST be called
    with Uid (CAM-internal id), NOT Uin (sub-account UIN). Threading
    Uin where Uid is wanted returns ResourceNotFound.UserNotExist."""
    seen: dict = {}

    class _StubClient:
        def GetUser(self, req):  # noqa: N802
            return _Resp(Uid=22001053, Uin=100042853879)

        def ListAttachedUserAllPolicies(self, req):  # noqa: N802
            return _Resp(PolicyList=[], TotalNum=0)

        def ListGroupsForUser(self, req):  # noqa: N802
            seen["ListGroupsForUser.Uid"] = req.Uid
            return _Resp(GroupInfo=[_Resp(GroupId=999, GroupName="G")], TotalNum=1)

        def RemoveUserFromGroup(self, req):  # noqa: N802
            seen["RemoveUserFromGroup.Info"] = list(req.Info)
            return _Resp()

        def ListAccessKeys(self, req):  # noqa: N802
            return _Resp(AccessKeys=[])

        def UpdateUser(self, req):  # noqa: N802
            return _Resp()

        def DeleteUser(self, req):  # noqa: N802
            return _Resp()

    driver = _driver()
    with patch.object(
        TencentCloudDriver, "_cam_client", return_value=_StubClient(),
    ):
        result = await driver.revoke_account(CloudAccountRevokeContext(
            provider="tencent_cloud",
            account_id="lifu.wu@pingcap.com",
        ))

    assert result.ok, result.error_message
    assert seen["ListGroupsForUser.Uid"] == 22001053, (
        "ListGroupsForUser.Uid must be the CAM-internal Uid "
        "(22001053), NOT the sub-account Uin (100042853879)."
    )
    assert seen["RemoveUserFromGroup.Info"] == [
        {"Uid": 22001053, "GroupId": 999},
    ]


@pytest.mark.asyncio
async def test_revoke_account_page_cap_stops_before_delete_user() -> None:
    delete_calls = 0

    class _StubClient:
        def GetUser(self, req):  # noqa: N802
            return _Resp(Uid=22, Uin=100)

        def ListAttachedUserAllPolicies(self, req):  # noqa: N802
            return _Resp(
                PolicyList=[
                    _Resp(PolicyId=index, PolicyName=f"p-{index}")
                    for index in range(200)
                ],
                TotalNum=201,
            )

        def DeleteUser(self, req):  # noqa: N802
            nonlocal delete_calls
            delete_calls += 1
            return _Resp()

    driver = _driver()
    with (
        patch.object(driver, "_cam_client", return_value=_StubClient()),
        patch(
            "app.integrations.drivers.tencent_cloud."
            "_MAX_CAM_ATTACHED_POLICY_PAGES",
            1,
        ),
    ):
        result = await driver.revoke_account(CloudAccountRevokeContext(
            provider="tencent_cloud",
            account_id="leaver@pingcap.com",
        ))

    assert result.ok is False
    assert "safety ceiling" in (result.error_message or "")
    assert delete_calls == 0


# ── grant_access (AttachUserPolicy carries only AttachUin + PolicyId) ──


@pytest.mark.asyncio
async def test_grant_access_attach_request_has_no_attach_user_name() -> None:
    """AttachUserPolicyRequest must carry ONLY PolicyId + AttachUin.
    Setting AttachUserName becomes an ad-hoc attr that serializes
    mangled as 'EtachUserName' and Tencent rejects with
    UnknownParameter (parity with the Detach bug)."""
    captured: dict = {}

    class _StubClient:
        def AddUser(self, req):  # noqa: N802
            return _Resp(Uin=200099991111, Uid=12345)

        def AttachUserPolicy(self, req):  # noqa: N802
            captured["PolicyId"] = req.PolicyId
            captured["AttachUin"] = req.AttachUin
            captured["has_AttachUserName"] = "AttachUserName" in req.__dict__
            return _Resp()

    driver = _driver()
    with patch.object(
        TencentCloudDriver, "_cam_client", return_value=_StubClient(),
    ):
        result = await driver.grant_access(CloudAccessContext(
            principal_email="newhire@pingcap.com",
            principal_display_name="New Hire",
            scope=_scope(),
            role_key="123456",
            role_display_name="ReadOnlyAccess",
        ))

    assert result.ok, result.error_message
    assert captured["PolicyId"] == 123456
    assert captured["AttachUin"] == 200099991111
    assert captured["has_AttachUserName"] is False


@pytest.mark.asyncio
async def test_grant_access_includes_hris_mobile_for_new_cam_user() -> None:
    """CAM needs the mobile identity factor before its API-key flow works."""
    captured: dict = {}

    class _StubClient:
        def AddUser(self, req):  # noqa: N802
            captured["PhoneNum"] = req.PhoneNum
            captured["CountryCode"] = req.CountryCode
            return _Resp(Uin=200099991111, Uid=12345)

        def AttachUserPolicy(self, req):  # noqa: N802
            return _Resp()

    driver = _driver()
    with patch.object(
        TencentCloudDriver, "_cam_client", return_value=_StubClient(),
    ):
        result = await driver.grant_access(CloudAccessContext(
            principal_email="newhire@pingcap.com",
            principal_display_name="New Hire",
            scope=_scope(),
            role_key="123456",
            extra={
                "mobile_phone": "138 0013 8000",
                "mobile_country_code": "CN",
            },
        ))

    assert result.ok, result.error_message
    assert captured == {"PhoneNum": "13800138000", "CountryCode": "86"}


@pytest.mark.asyncio
async def test_grant_access_duplicate_sub_user_name_continues_to_attach_policy() -> None:
    """CAM's current duplicate-name code is an idempotent create result."""
    from tencentcloud.common.exception.tencent_cloud_sdk_exception import (
        TencentCloudSDKException,
    )

    captured: dict = {}

    class _StubClient:
        def AddUser(self, req):  # noqa: N802
            raise TencentCloudSDKException(
                "InvalidParameter.SubUserNameInUse",
                "The name already exists",
                "request-id",
            )

        def GetUser(self, req):  # noqa: N802
            assert req.Name == "zuowuhui"
            return _Resp(Uid=12345, Uin=200099991111)

        def AttachUserPolicy(self, req):  # noqa: N802
            captured["PolicyId"] = req.PolicyId
            captured["AttachUin"] = req.AttachUin
            return _Resp()

    driver = _driver()
    with patch.object(
        TencentCloudDriver, "_cam_client", return_value=_StubClient(),
    ):
        result = await driver.grant_access(CloudAccessContext(
            principal_email="zuowuhui@pingcap.com",
            principal_display_name="左武辉",
            scope=_scope(),
            role_key="123456",
            role_display_name="QCloudFinanceFullAccess",
        ))

    assert result.ok, result.error_message
    assert result.detail["created"] is False
    assert captured == {"PolicyId": 123456, "AttachUin": 200099991111}