"""UCloud ``list_scope_grants`` pagination-truncation regression (#1655).

Prod risk: ``ListUsers`` / ``ListPoliciesForUser`` terminated on
``offset >= total`` where ``total`` came from ``TotalCount``. When UCloud
returned a FULL page but ``TotalCount`` was absent / 0 (a real degenerate
shape), ``offset (100) >= total (0)`` was True → the loop broke after page 1,
truncating the roster. The grants-sweep then reads the missing tail as
"cloud removed those grants" and mass-revokes them.

Fix: terminate on a SHORT page (``len(page) < limit``) or an empty page, and
only trust ``offset >= total`` when ``total`` is positive; a page ceiling
guards a degenerate infinite loop.
"""
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.ucloud_cloud import UCloudDriver


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


def _scope() -> CloudScopeRef:
    return CloudScopeRef(
        scope_type="standalone_iam",
        external_id="org-z2uz3o",
        display_name="UCloud",
    )


@pytest.mark.asyncio
async def test_list_scope_grants_keeps_paging_when_total_count_absent() -> None:
    """A full first page of users with TotalCount absent/0 must NOT truncate
    the roster after page 1 — the driver must keep paging until a short page."""
    # Page 1: 100 users (full), TotalCount absent. Page 2: 1 user (short),
    # terminal. Every user holds exactly one policy (a single short page).
    page1_users = [
        {"UserName": f"u{i}", "Email": f"u{i}@pingcap.com", "UserId": str(i)}
        for i in range(100)
    ]
    page2_users = [
        {"UserName": "utail", "Email": "utail@pingcap.com", "UserId": "999"},
    ]

    async def _fake_call(action: str, params: dict[str, Any], **_kw) -> dict:
        if action == "ListUsers":
            offset = params["Offset"]
            if offset == 0:
                # Full page, NO TotalCount key at all (degenerate shape).
                return {"Users": page1_users}
            if offset == 100:
                return {"Users": page2_users}
            return {"Users": []}
        if action == "ListPoliciesForUser":
            return {
                "Policies": [
                    {"PolicyName": "ReadOnly", "PolicyURN": "ucs:iam::ucs:policy/RO"},
                ],
                "TotalCount": 1,
            }
        return {}

    drv = _make_driver()
    with patch.object(UCloudDriver, "_call", new=AsyncMock(side_effect=_fake_call)):
        grants = await drv.list_scope_grants(_scope())

    # All 101 users were paged (not truncated to the first 100), each yielding
    # one discovered grant.
    emails = {g.principal_email for g in grants}
    assert len(grants) == 101
    assert "utail@pingcap.com" in emails


@pytest.mark.asyncio
async def test_list_scope_grants_policies_keep_paging_when_total_absent() -> None:
    """A single user whose ListPoliciesForUser returns a full first page with
    TotalCount absent must have the policy tail paged, not truncated."""
    page1_policies = [
        {"PolicyName": f"P{i}", "PolicyURN": f"ucs:iam::ucs:policy/P{i}"}
        for i in range(100)
    ]
    page2_policies = [
        {"PolicyName": "Ptail", "PolicyURN": "ucs:iam::ucs:policy/Ptail"},
    ]

    async def _fake_call(action: str, params: dict[str, Any], **_kw) -> dict:
        if action == "ListUsers":
            if params["Offset"] == 0:
                return {
                    "Users": [
                        {"UserName": "solo", "Email": "solo@pingcap.com", "UserId": "1"},
                    ],
                }
            return {"Users": []}
        if action == "ListPoliciesForUser":
            offset = params["Offset"]
            if offset == 0:
                return {"Policies": page1_policies}  # full page, no TotalCount
            if offset == 100:
                return {"Policies": page2_policies}
            return {"Policies": []}
        return {}

    drv = _make_driver()
    with patch.object(UCloudDriver, "_call", new=AsyncMock(side_effect=_fake_call)):
        grants = await drv.list_scope_grants(_scope())

    role_keys = {g.role_key for g in grants}
    assert len(grants) == 101
    assert "Ptail" in role_keys