"""UCloud driver ``revoke_access`` regression — 2026-05-14 (round 3).

Prod incident: offboarding huohao@pingcap.com cycled 58 times across
6 UCloud grants. Three iterations of this fix landed before the
correct shape was confirmed against UCloud's API docs.

Root cause (Round-23):
1. ``external_ref`` shape ``{user_name}::{policy_urn}::{policy_name}``
   contains ``::`` inside ``policy_urn`` (``ucs:iam::ucs:policy/X``).
   The previous parser took ``parts[1]`` and sent a truncated URN.
   Fixed by reconstructing URN as ``"::".join(parts[1:-1])``.
2. ``Scope`` IS required on Detach, but legacy code used the value
   ``"Account"`` — wrong. UCloud's current API expects ``"All"`` for
   tenant-wide attachments or ``"Specified"`` (requires ``ProjectID``)
   for project-scoped ones. The same fix corrects ``AttachPoliciesToUser``
   which had the same wrong ``"Account"`` value.

Investigation false alarms:
* PR #295 round 1: added ``Scope="Account"`` reading the literal
  RetCode=220 error text. That triggered RetCode=230 "Invalid Scope".
* PR #296 round 2: removed ``Scope`` entirely. That regressed to
  RetCode=220 again — Scope IS mandatory; the value just had to be
  correct. Round 3 (this PR) pins ``Scope="All"`` per the docs.

These tests pin: URN reconstruction, ``Scope="All"`` on both Attach
and Detach paths, plus the revoke_account delete-chain Scope.
"""
from __future__ import annotations

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

import pytest

from app.integrations.driver_base import CloudAccessContext, CloudScopeRef
from app.integrations.drivers.ucloud_cloud import UCloudDriver


def _make_driver() -> UCloudDriver:
    return UCloudDriver(
        integration_id="int-ucloud-test",
        config={"public_key": "PUB", "private_key": "PRIV"},
    )


def _ctx(external_ref: str, role_key: str = "AdministratorAccess") -> CloudAccessContext:
    return CloudAccessContext(
        principal_email="huohao@pingcap.com",
        principal_display_name="huohao",
        scope=CloudScopeRef(
            scope_type="standalone_iam",
            external_id="org-z2uz3o",
            display_name="UCloud",
        ),
        role_key=role_key,
        existing_external_ref=external_ref,
    )


@pytest.mark.asyncio
async def test_revoke_access_reconstructs_urn_with_double_colon() -> None:
    """Prod-shape external_ref must round-trip the full policy URN
    through to the DetachPoliciesFromUser call.

    ``"name-8pto9woh::ucs:iam::ucs:policy/UBillFullAccess::UBillFullAccess"``
    must become PolicyURNs.0=``ucs:iam::ucs:policy/UBillFullAccess``
    (NOT the truncated ``ucs:iam`` that prod was sending).
    """
    captured: list[dict[str, Any]] = []

    async def fake_call(action: str, params: dict[str, Any] | None = None, *, retryable: bool = False) -> dict[str, Any]:
        captured.append({"action": action, "params": params})
        return {}

    driver = _make_driver()
    ctx = _ctx(
        "name-8pto9woh::ucs:iam::ucs:policy/UBillFullAccess::UBillFullAccess",
        role_key="UBillFullAccess",
    )
    with patch.object(driver, "_call", new=AsyncMock(side_effect=fake_call)):
        result = await driver.revoke_access(ctx)

    assert result.ok is True
    assert len(captured) == 1
    call = captured[0]
    assert call["action"] == "DetachPoliciesFromUser"
    assert call["params"] == {
        "UserName": "name-8pto9woh",
        "PolicyURNs.0": "ucs:iam::ucs:policy/UBillFullAccess",
        "Scope": "All",
    }


@pytest.mark.asyncio
async def test_revoke_access_scope_must_be_all_not_account() -> None:
    """``DetachPoliciesFromUser`` requires ``Scope`` with value ``"All"``
    or ``"Specified"`` per UCloud's docs. ``"Account"`` is REJECTED with
    RetCode=230 "Invalid Scope" — pin the correct value.

    Regression guard for two earlier wrong attempts:
    * Round 1 (PR #295): added ``Scope="Account"`` → RetCode=230.
    * Round 2 (PR #296): removed Scope entirely → RetCode=220.
    Both surfaced as huohao@pingcap.com's offboarding cycling 58×.
    """
    captured: list[dict[str, Any]] = []

    async def fake_call(action: str, params: dict[str, Any] | None = None, *, retryable: bool = False) -> dict[str, Any]:
        captured.append(params or {})
        return {}

    driver = _make_driver()
    ctx = _ctx("u1::ucs:iam::ucs:policy/X::X", role_key="X")
    with patch.object(driver, "_call", new=AsyncMock(side_effect=fake_call)):
        await driver.revoke_access(ctx)

    assert captured[0].get("Scope") == "All", (
        f"Scope MUST be 'All' on DetachPoliciesFromUser; got {captured[0]!r}"
    )


@pytest.mark.asyncio
async def test_revoke_access_falls_back_to_resolve_urn_for_legacy_refs() -> None:
    """Legacy 2-part refs (pre-{user::urn::name} convention) — drop
    ``parts[1]`` into the URN lookup path."""
    captured_actions: list[str] = []

    async def fake_call(action: str, params: dict[str, Any] | None = None, *, retryable: bool = False) -> dict[str, Any]:
        captured_actions.append(action)
        if action == "ListPolicies":
            return {
                "Policies": [
                    {"PolicyName": "LegacyPolicy", "PolicyURN": "ucs:iam::ucs:policy/LegacyPolicy"},
                ],
                "TotalCount": 1,
            }
        return {}

    driver = _make_driver()
    ctx = _ctx("legacy-user::LegacyPolicy", role_key="LegacyPolicy")
    with patch.object(driver, "_call", new=AsyncMock(side_effect=fake_call)):
        result = await driver.revoke_access(ctx)

    assert result.ok is True
    # ListPolicies must have run because the URN wasn't in the ref.
    assert "ListPolicies" in captured_actions
    assert "DetachPoliciesFromUser" in captured_actions


@pytest.mark.asyncio
async def test_revoke_access_missing_urn_is_idempotent_success() -> None:
    """Policy gone from tenant → already_absent (idempotent success)."""
    async def fake_call(action: str, params: dict[str, Any] | None = None, *, retryable: bool = False) -> dict[str, Any]:
        if action == "ListPolicies":
            return {"Policies": [], "TotalCount": 0}
        raise AssertionError(f"unexpected {action}")

    driver = _make_driver()
    ctx = _ctx("u1::UnknownPolicy", role_key="UnknownPolicy")
    with patch.object(driver, "_call", new=AsyncMock(side_effect=fake_call)):
        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_translates_not_attached_to_already_absent() -> None:
    async def fake_call(action: str, params: dict[str, Any] | None = None, *, retryable: bool = False) -> dict[str, Any]:
        if action == "DetachPoliciesFromUser":
            raise RuntimeError("UCloud DetachPoliciesFromUser: NotAttached")
        return {}

    driver = _make_driver()
    ctx = _ctx("u1::ucs:iam::ucs:policy/A::A", role_key="A")
    with patch.object(driver, "_call", new=AsyncMock(side_effect=fake_call)):
        result = await driver.revoke_access(ctx)

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


@pytest.mark.asyncio
async def test_revoke_account_detach_chain_passes_scope() -> None:
    """``revoke_account`` (full IAM user delete on offboard) has the
    same DetachPoliciesFromUser dependency on ``Scope`` — pin it so
    the offboarding finish path doesn't relapse.
    """
    from app.integrations.driver_base import CloudAccountRevokeContext
    captured: list[dict[str, Any]] = []

    async def fake_call(action: str, params: dict[str, Any] | None = None, *, retryable: bool = False) -> dict[str, Any]:
        captured.append({"action": action, "params": params})
        if action == "ListPoliciesForUser":
            return {"Policies": [{
                "PolicyName": "AdministratorAccess",
                "PolicyURN": "ucs:iam::ucs:policy/AdministratorAccess",
            }]}
        if action == "ListAccessKeys":
            return {"AccessKeys": []}
        return {}

    driver = _make_driver()
    ctx = CloudAccountRevokeContext(provider="ucloud", account_id="someuser")
    with patch.object(driver, "_call", new=AsyncMock(side_effect=fake_call)):
        result = await driver.revoke_account(ctx)

    assert result.ok is True
    detach_calls = [c for c in captured if c["action"] == "DetachPoliciesFromUser"]
    assert len(detach_calls) == 1
    assert detach_calls[0]["params"] == {
        "UserName": "someuser",
        "PolicyURNs.0": "ucs:iam::ucs:policy/AdministratorAccess",
        "Scope": "All",
    }


@pytest.mark.asyncio
async def test_revoke_access_retcode_11021_is_already_absent() -> None:
    """UCloud ``RetCode=11021: Account not exist`` → already_absent.

    Surfaced 2026-06-01 on chujie@'s offboarding: the UCloud IAM user
    had been admin-deleted out-of-band, so ``DetachPoliciesFromUser``
    immediately returns ``RetCode=11021 Account not exist``. The driver
    used to return ``ok=False`` here, which left
    ``requires_manual_cloud_revoke=True`` on the grant and made
    ``sweep_pending_manual_cloud_revokes`` retry forever (no possible
    fix — the account literally no longer exists).
    """
    async def fake_call(action: str, params: dict[str, Any] | None = None, *, retryable: bool = False) -> dict[str, Any]:
        if action == "DetachPoliciesFromUser":
            raise RuntimeError(
                "UCloud DetachPoliciesFromUser RetCode=11021: Account not exist",
            )
        raise AssertionError(f"unexpected {action}")

    driver = _make_driver()
    ctx = _ctx(
        "deleted-user::ucs:iam::ucs:policy/AdministratorAccess::AdministratorAccess",
        role_key="AdministratorAccess",
    )
    with patch.object(driver, "_call", new=AsyncMock(side_effect=fake_call)):
        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_account_not_exist_string_is_already_absent() -> None:
    """Defence in depth — if UCloud surfaces the same condition via a
    different error string (no ``RetCode=11021`` prefix) the substring
    match on ``"Account not exist"`` still treats it as idempotent."""
    async def fake_call(action: str, params: dict[str, Any] | None = None, *, retryable: bool = False) -> dict[str, Any]:
        if action == "DetachPoliciesFromUser":
            raise RuntimeError("Account not exist (per UCloud)")
        raise AssertionError(f"unexpected {action}")

    driver = _make_driver()
    ctx = _ctx(
        "ghost::ucs:iam::ucs:policy/X::X", role_key="X",
    )
    with patch.object(driver, "_call", new=AsyncMock(side_effect=fake_call)):
        result = await driver.revoke_access(ctx)

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