"""Multi-sub-account fanout: AlicloudIntlProvider via STS AssumeRole.

The intl tenant has 17 member accounts in production. Single-account
discovery returned 0 servers / 0 domains / 0 IAM users because it only
hit the master account, where no resources live. This module locks in
the AssumeRole-fanout path:

  * ``_list_member_accounts`` paginates Resource Directory
    ``ListAccounts`` and falls back to single-account when the tenant
    isn't part of an RD.
  * ``_assume_role`` STS-AssumeRoles per sub-account, caches the temp
    credentials, and returns ``None`` on failure (the fanout silently
    skips that sub-account).
  * ``_impersonate`` swaps ``self._ak / _sk / _security_token`` for the
    duration of a context-manager block.
  * ``list_servers`` / ``list_iam_users`` etc. fan out across all
    member accounts AND tag every result with ``__sub_account_id`` so
    the UI can render per-sub-account totals.
"""
from __future__ import annotations

from unittest.mock import AsyncMock, patch

import pytest

from app.domains.cmdb.providers.alicloud_intl import (
    _DEFAULT_MEMBER_ROLE,
    AlicloudIntlProvider,
    _stamp_account_id,
)
from app.domains.cmdb.providers.base import (
    CloudServer,
    ProviderCapability,
    ProviderCredentials,
)


def _creds() -> ProviderCredentials:
    return ProviderCredentials(
        provider_id="test-intl",
        provider_type="alicloud_intl",
        account_alias="test-intl",
        fields={
            "access_key_id": "MASTER_AK",
            "access_key_secret": "MASTER_SK",
            "regions": "us-east-1",  # one region keeps the test deterministic
        },
    )


def test_iam_capability_now_supported() -> None:
    """Regression: prior to the multi-sub-account refactor the intl
    provider declared the same caps as 国内 (no IAM). With per-account
    AssumeRole'd RAM ListUsers, IAM is real — sync.py routes
    ``cmdb.iam_sync`` only when the provider declares it."""
    caps = AlicloudIntlProvider.supported_capabilities()
    assert ProviderCapability.IAM in caps
    assert ProviderCapability.SERVERS in caps  # didn't drop existing


def test_member_role_name_default() -> None:
    p = AlicloudIntlProvider(_creds())
    assert p._member_role_name() == _DEFAULT_MEMBER_ROLE


def test_member_role_name_override() -> None:
    creds = _creds()
    creds.fields["member_assume_role_name"] = "CustomCrossAccountRole"
    p = AlicloudIntlProvider(creds)
    assert p._member_role_name() == "CustomCrossAccountRole"


@pytest.mark.asyncio
async def test_list_member_accounts_paginates() -> None:
    """Two-page Resource Directory response. ``IsTruncated``-style
    pagination via ``TotalCount`` + ``PageNumber`` × ``PageSize``."""
    p = AlicloudIntlProvider(_creds())
    page1 = {
        "Accounts": {"Account": [
            {"AccountId": f"5{i:015d}", "DisplayName": f"acct-{i}",
             "Type": "ResourceAccount", "FolderId": "fd-1"}
            for i in range(100)
        ]},
        "TotalCount": 150,
    }
    page2 = {
        "Accounts": {"Account": [
            {"AccountId": f"5{i:015d}", "DisplayName": f"acct-{i}",
             "Type": "ResourceAccount", "FolderId": "fd-1"}
            for i in range(100, 150)
        ]},
        "TotalCount": 150,
    }
    calls: list[dict] = []

    async def fake_call(url, params):
        calls.append(dict(params))
        page = int(params["PageNumber"])
        return page1 if page == 1 else page2

    with patch.object(p, "_call", side_effect=fake_call):
        accts = await p._list_member_accounts()

    assert len(accts) == 150
    # Confirms second-page request was made — pagination didn't cut early.
    assert len(calls) == 2
    assert calls[1]["PageNumber"] == "2"


@pytest.mark.asyncio
async def test_list_member_accounts_falls_back_when_no_rd() -> None:
    """Tenants outside a Resource Directory raise either
    ``ResourceDirectoryNotInUse`` or a 403 RAM error — neither should
    crash the sync. The intl provider degrades to single-account mode
    by returning ``[]`` (callers treat empty list as "just call
    master")."""
    p = AlicloudIntlProvider(_creds())
    with patch.object(
        p, "_call",
        side_effect=RuntimeError("ResourceDirectoryNotInUse: …"),
    ):
        accts = await p._list_member_accounts()
    assert accts == []


@pytest.mark.asyncio
async def test_assume_role_caches_for_session() -> None:
    """STS rate limits matter when fanning out across 17 accounts × 14
    regions per resource type. The temp token must be reused across
    calls until it nears expiry."""
    p = AlicloudIntlProvider(_creds())
    sts_calls: list[dict] = []

    async def fake_call(url, params):
        sts_calls.append(dict(params))
        return {
            "Credentials": {
                "AccessKeyId": "TEMP_AK",
                "AccessKeySecret": "TEMP_SK",
                "SecurityToken": "TEMP_TOKEN",
            },
        }

    with patch.object(p, "_call", side_effect=fake_call):
        c1 = await p._assume_role("500000000000001")
        c2 = await p._assume_role("500000000000001")  # same account → cache hit

    assert c1 == ("TEMP_AK", "TEMP_SK", "TEMP_TOKEN")
    assert c1 == c2
    # Only one STS round-trip despite two ``_assume_role`` calls.
    assert len(sts_calls) == 1


@pytest.mark.asyncio
async def test_assume_role_returns_none_on_failure() -> None:
    """No trust policy on the sub-account → ``NoPermission.AssumeRole``.
    The fanout code uses ``None`` as the "skip this account" signal so
    one broken sub-account can't poison the whole sync."""
    p = AlicloudIntlProvider(_creds())
    with patch.object(
        p, "_call",
        side_effect=RuntimeError("NoPermission.AssumeRole: trust policy"),
    ):
        result = await p._assume_role("500000000000001")
    assert result is None


@pytest.mark.asyncio
async def test_impersonate_swaps_and_restores_credentials() -> None:
    """``_impersonate`` is the contract the fanout depends on: inside
    the block the provider must look like the sub-account; after the
    block it must look like the master again, even on exception."""
    p = AlicloudIntlProvider(_creds())

    async def fake_assume(_account_id):
        return ("TEMP_AK", "TEMP_SK", "TOKEN_X")

    with patch.object(p, "_assume_role", side_effect=fake_assume):
        async with p._impersonate("500000000000001") as scope_id:
            assert scope_id == "500000000000001"
            assert p._ak == "TEMP_AK"
            assert p._sk == "TEMP_SK"
            assert p._security_token == "TOKEN_X"

    # Restored.
    assert p._ak == "MASTER_AK"
    assert p._sk == "MASTER_SK"
    assert p._security_token is None


@pytest.mark.asyncio
async def test_impersonate_yields_none_when_assume_fails() -> None:
    """Caller pattern: ``async with _impersonate(id) as scope_id:`` then
    ``if scope_id is None: continue``. The credential swap MUST NOT
    happen when AssumeRole returned ``None`` — otherwise the next call
    inside the block would use stale half-swapped state."""
    p = AlicloudIntlProvider(_creds())
    p._security_token = None
    with patch.object(p, "_assume_role", AsyncMock(return_value=None)):
        async with p._impersonate("500000000000001") as scope_id:
            assert scope_id is None
            # Master credentials untouched.
            assert p._ak == "MASTER_AK"


@pytest.mark.asyncio
async def test_fanout_aggregates_across_accounts_and_skips_failures() -> None:
    """End-to-end shape: 3 sub-accounts, 1 fails AssumeRole, 2 succeed.
    Each successful sub-account contributes 1 server. Master also
    contributes 1. Total = 3 (1 master + 2 successful subs) and
    every server's ``raw["__sub_account_id"]`` is set."""
    p = AlicloudIntlProvider(_creds())

    members = [
        {"account_id": "AAA111", "display_name": "a", "type": "Resource", "folder_id": None},
        {"account_id": "BBB222", "display_name": "b", "type": "Resource", "folder_id": None},
        {"account_id": "CCC333", "display_name": "c", "type": "Resource", "folder_id": None},
    ]

    async def fake_members():
        return members

    # _assume_role: account BBB fails, others succeed.
    async def fake_assume(account_id):
        if account_id == "BBB222":
            return None
        return (f"AK_{account_id}", f"SK_{account_id}", f"TOK_{account_id}")

    # AliyunProvider.list_servers will be called 3 times (master + AAA + CCC).
    # Each time produce one server tagged with the *currently swapped* AK
    # so the fanout's ``_stamp_account_id`` actually has something to
    # observe.
    async def fake_list_servers(_self):
        return [
            CloudServer(
                instance_id=f"i-{_self._ak}",
                name="srv",
                status="running",
                region="us-east-1",
                zone=None,
                vcpu=1,
                memory_mb=1024,
                public_ips=[],
                private_ips=["10.0.0.1"],
                image_id=None,
                os_type="linux",
                tags={},
                raw={},
            ),
        ]

    with patch.object(p, "_list_member_accounts", side_effect=fake_members), \
         patch.object(p, "_assume_role", side_effect=fake_assume), \
         patch(
             "app.domains.cmdb.providers.aliyun.AliyunProvider.list_servers",
             new=fake_list_servers,
         ):
        out = await p.list_servers()

    # Master + AAA + CCC = 3. BBB skipped.
    assert len(out) == 3
    sub_ids = sorted(r.raw["__sub_account_id"] for r in out)
    assert sub_ids == ["AAA111", "CCC333", "__master__"]
    # BBB never appears.
    assert all("BBB222" not in r.raw["__sub_account_id"] for r in out)


def test_stamp_account_id_on_dataclass_resource() -> None:
    server = CloudServer(
        instance_id="i-x", name="x", status="running",
        region="us-east-1", zone=None, vcpu=1, memory_mb=1024,
        public_ips=[], private_ips=[], image_id=None, os_type="linux",
        tags={}, raw={},
    )
    _stamp_account_id(server, "5000001")
    assert server.raw["__sub_account_id"] == "5000001"
    assert server.tags["__sub_account_id"] == "5000001"


def test_stamp_account_id_on_dict_resource() -> None:
    """``list_object_storage`` returns dicts, not dataclasses — the
    stamp function must handle both shapes without losing data."""
    bucket = {"name": "my-bucket", "region": "us-east-1"}
    _stamp_account_id(bucket, "5000002")
    assert bucket["__sub_account_id"] == "5000002"
    assert bucket["raw"]["__sub_account_id"] == "5000002"
    # Existing keys untouched.
    assert bucket["name"] == "my-bucket"


def test_stamp_account_id_does_not_overwrite_existing() -> None:
    """If a downstream method already stamped the account_id, the
    second stamp must be a no-op — otherwise master-account-fanout
    would clobber sub-account stamps written upstream."""
    server = CloudServer(
        instance_id="i-x", name="x", status="running",
        region="us-east-1", zone=None, vcpu=1, memory_mb=1024,
        public_ips=[], private_ips=[], image_id=None, os_type="linux",
        tags={}, raw={"__sub_account_id": "ORIGINAL"},
    )
    _stamp_account_id(server, "OVERWRITER")
    assert server.raw["__sub_account_id"] == "ORIGINAL"


def _fake_ram_user(account_id: str) -> dict:
    return {
        "user_name": f"alice_{account_id}",
        "arn": f"acs:ram::{account_id}:user/alice",
        "user_id_external": "100001",
        "create_date": None,
        "last_login_at": None,
        "attached_policies": [],
        "groups": [],
        "mfa_enabled": False,
        "tags": {"__sub_account_id": account_id},
        "raw": {"__sub_account_id": account_id},
    }


@pytest.mark.asyncio
async def test_iam_user_fanout_includes_master_and_subs() -> None:
    """IAM mirrors the resource fanout: master RAM users + each member's RAM
    users, tagged with ``__sub_account_id``. When every account is readable
    the snapshot is complete."""
    p = AlicloudIntlProvider(_creds())

    async def fake_members():
        return [
            {"account_id": "AAA111"},
            {"account_id": "BBB222"},
        ]

    async def fake_assume(account_id):
        return (f"AK_{account_id}", f"SK_{account_id}", f"TOK_{account_id}")

    async def fake_list_users(account_id):
        return [_fake_ram_user(account_id)]

    with patch.object(p, "_list_member_accounts", side_effect=fake_members), \
         patch.object(p, "_assume_role", side_effect=fake_assume), \
         patch.object(
             p, "_list_ram_users_current_credentials",
             side_effect=fake_list_users,
         ):
        users = await p.list_iam_users()

    user_account_ids = sorted(u["raw"]["__sub_account_id"] for u in users)
    assert user_account_ids == ["AAA111", "BBB222", "__master__"]


@pytest.mark.asyncio
async def test_iam_user_fanout_raises_on_assume_role_failure() -> None:
    """Regression (#1645): a sub-account whose AssumeRole fails yields an
    INCOMPLETE snapshot, not an empty one. ``list_iam_users`` must RAISE so a
    downstream mark-and-sweep is disabled — silently returning master + AAA
    (dropping BBB's active RAM users) would let the sweep revoke in-use
    users."""
    p = AlicloudIntlProvider(_creds())

    async def fake_members():
        return [
            {"account_id": "AAA111"},
            {"account_id": "BBB222"},
        ]

    async def fake_assume(account_id):
        # BBB222 AssumeRole fails → _impersonate yields None.
        return None if account_id == "BBB222" else (
            f"AK_{account_id}", f"SK_{account_id}", f"TOK_{account_id}",
        )

    async def fake_list_users(account_id):
        return [_fake_ram_user(account_id)]

    with patch.object(p, "_list_member_accounts", side_effect=fake_members), \
         patch.object(p, "_assume_role", side_effect=fake_assume), \
         patch.object(
             p, "_list_ram_users_current_credentials",
             side_effect=fake_list_users,
         ), \
         pytest.raises(RuntimeError, match="partial RAM-user snapshot"):
        await p.list_iam_users()


@pytest.mark.asyncio
async def test_iam_user_fanout_raises_on_list_users_failure() -> None:
    """Regression (#1645): a transient ListUsers failure on one account (e.g.
    a 429 throttle) must also raise, not be swallowed with ``continue``."""
    p = AlicloudIntlProvider(_creds())

    async def fake_members():
        return [{"account_id": "AAA111"}]

    async def fake_assume(account_id):
        return (f"AK_{account_id}", f"SK_{account_id}", f"TOK_{account_id}")

    async def fake_list_users(account_id):
        if account_id == "AAA111":
            raise RuntimeError("Throttling.User: 429")
        return [_fake_ram_user(account_id)]

    with patch.object(p, "_list_member_accounts", side_effect=fake_members), \
         patch.object(p, "_assume_role", side_effect=fake_assume), \
         patch.object(
             p, "_list_ram_users_current_credentials",
             side_effect=fake_list_users,
         ), \
         pytest.raises(RuntimeError, match="partial RAM-user snapshot"):
        await p.list_iam_users()