"""Tests for the AWS sub-account create-scope path — owner-alias
derivation, list_create_account_status idempotency, and the
ClientRequestToken bug regression guard.

Background (2026-05-25): AWS Organizations CreateAccount does NOT
accept a ``ClientRequestToken`` parameter — botocore rejects it
client-side with ``Parameter validation failed``. The driver
previously passed one, so every retry blew up before reaching AWS.
Idempotency is now achieved via ``list_create_account_status``
sweeping for an in-flight / succeeded entry matching AccountName.
"""

from __future__ import annotations

from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import pytest

# ─────────────────────────────────────────────────────────────────────
# Alias derivation
# ─────────────────────────────────────────────────────────────────────


def test_derive_alias_basic_human_name() -> None:
    """A typical scope name → slugged-as-typed alias (no forced
    ``aws-`` prefix; the name itself carries the provider intent)."""
    from app.domains.cmdb.scope_request_service import derive_aws_owner_alias

    assert derive_aws_owner_alias("AWS PingCAP Datalake") == \
        "aws-pingcap-datalake@pingcap.com"


def test_derive_alias_no_forced_aws_prefix() -> None:
    """**Key 2026-05-25 owner call**: names that don't start with
    ``aws-`` MUST NOT be force-prefixed. The alias = the sanitised
    request name verbatim.
    """
    from app.domains.cmdb.scope_request_service import derive_aws_owner_alias

    # Names with no aws- prefix stay un-prefixed.
    assert derive_aws_owner_alias("pingcap-foo") == \
        "pingcap-foo@pingcap.com"
    assert derive_aws_owner_alias("datalake-prod") == \
        "datalake-prod@pingcap.com"
    # Names that already carry aws- pass through 1:1 (slug rules apply
    # but no second aws- is added).
    assert derive_aws_owner_alias("aws-pingcap-datalake") == \
        "aws-pingcap-datalake@pingcap.com"


def test_derive_alias_empty_slug_raises() -> None:
    """Pathological all-CJK / all-punctuation names raise instead of
    falling back to a synthetic placeholder — a fixed fallback like
    ``unnamed@pingcap.com`` would collide between two pathological
    requests, far harder to diagnose than a clear up-front error."""
    import pytest as _pytest

    from app.domains.cmdb.scope_request_service import derive_aws_owner_alias

    for bad in ("数据湖", "   ", "___", "***", ""):
        with _pytest.raises(ValueError, match="empty slug"):
            derive_aws_owner_alias(bad)


def test_derive_alias_overlong_raises() -> None:
    """Google caps the local part at 64 chars. Slugged length over
    the cap raises so the admin renames the scope — silent truncation
    could collide with another truncated-to-the-same-prefix row."""
    import pytest as _pytest

    from app.domains.cmdb.scope_request_service import (
        _AWS_OWNER_ALIAS_LOCAL_PART_MAX,
        derive_aws_owner_alias,
    )

    with _pytest.raises(ValueError, match="exceeds Google"):
        derive_aws_owner_alias("a" * (_AWS_OWNER_ALIAS_LOCAL_PART_MAX + 5))


def test_derive_alias_at_cap_succeeds() -> None:
    """Exactly at the cap is allowed (boundary inclusive)."""
    from app.domains.cmdb.scope_request_service import (
        _AWS_OWNER_ALIAS_LOCAL_PART_MAX,
        derive_aws_owner_alias,
    )

    name = "a" * _AWS_OWNER_ALIAS_LOCAL_PART_MAX
    out = derive_aws_owner_alias(name)
    local = out.split("@")[0]
    assert len(local) == _AWS_OWNER_ALIAS_LOCAL_PART_MAX


def test_derive_alias_special_chars_collapse() -> None:
    """Multiple non-slug chars collapse to a single dash; leading /
    trailing dashes get stripped; CJK fragments next to ASCII drop."""
    from app.domains.cmdb.scope_request_service import derive_aws_owner_alias

    assert derive_aws_owner_alias("AWS_DBaaS-Prod-2.0") == \
        "aws-dbaas-prod-2-0@pingcap.com"
    assert derive_aws_owner_alias("--leading---trailing--") == \
        "leading-trailing@pingcap.com"
    assert derive_aws_owner_alias("数据湖 PROD") == \
        "prod@pingcap.com"
    assert derive_aws_owner_alias("My Project NN!") == \
        "my-project-nn@pingcap.com"


# ─────────────────────────────────────────────────────────────────────
# _find_existing_create_account_request — list-based idempotency
# ─────────────────────────────────────────────────────────────────────


def _make_paginator(pages: list[dict]) -> MagicMock:
    paginator = MagicMock()
    paginator.paginate.return_value = iter(pages)
    return paginator


def test_find_existing_create_account_returns_matching_request_id() -> None:
    """An IN_PROGRESS row matching AccountName → return its Id so the
    caller resumes polling that RequestId instead of creating a
    second sub-account."""
    from app.integrations.drivers.aws_cloud import AwsCloudDriver

    org = MagicMock()
    paginator = _make_paginator([
        {
            "CreateAccountStatuses": [
                {
                    "Id": "car-other-001",
                    "AccountName": "Other Account",
                    "State": "IN_PROGRESS",
                },
                {
                    "Id": "car-match-002",
                    "AccountName": "aws-pingcap-datalake",
                    "State": "IN_PROGRESS",
                },
            ],
        },
    ])
    org.get_paginator.return_value = paginator

    rid = AwsCloudDriver._find_existing_create_account_request(
        org, account_name="aws-pingcap-datalake",
    )
    assert rid == "car-match-002"
    # Sanity: we asked for the right paginator + the right States filter.
    org.get_paginator.assert_called_once_with("list_create_account_status")
    paginator.paginate.assert_called_once_with(States=["IN_PROGRESS", "SUCCEEDED"])


def test_find_existing_create_account_no_match() -> None:
    """No row matches — returns None so the caller proceeds with a
    fresh ``create_account``."""
    from app.integrations.drivers.aws_cloud import AwsCloudDriver

    org = MagicMock()
    org.get_paginator.return_value = _make_paginator([
        {"CreateAccountStatuses": [
            {"Id": "x", "AccountName": "different", "State": "SUCCEEDED"},
        ]},
    ])
    assert AwsCloudDriver._find_existing_create_account_request(
        org, account_name="aws-pingcap-datalake",
    ) is None


def test_find_existing_create_account_swallows_client_error() -> None:
    """ClientError on list_create_account_status falls through to
    None so the caller can still issue create_account — better than
    crashing the whole approve path on a list-call hiccup."""
    from botocore.exceptions import ClientError

    from app.integrations.drivers.aws_cloud import AwsCloudDriver

    org = MagicMock()
    paginator = MagicMock()
    paginator.paginate.side_effect = ClientError(
        {"Error": {"Code": "AccessDeniedException", "Message": "nope"}},
        "ListCreateAccountStatus",
    )
    org.get_paginator.return_value = paginator

    assert AwsCloudDriver._find_existing_create_account_request(
        org, account_name="anything",
    ) is None


def test_find_existing_create_account_trims_whitespace() -> None:
    """AccountName matching is whitespace-tolerant — historical
    rows from manual SQL fix-ups sometimes carry trailing space."""
    from app.integrations.drivers.aws_cloud import AwsCloudDriver

    org = MagicMock()
    org.get_paginator.return_value = _make_paginator([
        {"CreateAccountStatuses": [
            {"Id": "rid", "AccountName": "  aws-foo  ", "State": "SUCCEEDED"},
        ]},
    ])
    assert AwsCloudDriver._find_existing_create_account_request(
        org, account_name="aws-foo",
    ) == "rid"


# ─────────────────────────────────────────────────────────────────────
# _create_account_sync — ClientRequestToken absence + paths
# ─────────────────────────────────────────────────────────────────────


def _make_aws_driver_sso_mode():
    """Build an AwsCloudDriver in SSO mode (account_mode='sso') with
    enough config to satisfy the create-scope precondition checks."""
    from app.integrations.drivers.aws_cloud import AwsCloudDriver

    return AwsCloudDriver(
        integration_id="11111111-1111-1111-1111-111111111111",
        config={
            "account_mode": "sso",
            "sso_instance_arn": "arn:aws:sso:::instance/ssoins-abc",
            "sso_region": "us-east-1",
            "aws_access_key_id": "AKIA-FAKE",
            "aws_secret_access_key": "SECRET-FAKE",
            "default_region": "us-east-1",
        },
    )


def test_create_account_sync_does_not_pass_client_request_token() -> None:
    """**Regression guard for the bug**: ``ClientRequestToken`` is NOT
    a valid parameter on AWS Organizations CreateAccount. The driver
    MUST NOT pass it even when an ``idempotency_key`` is supplied.
    """
    driver = _make_aws_driver_sso_mode()

    org = MagicMock()
    # No existing in-flight — fall through to fresh create_account.
    org.get_paginator.return_value = _make_paginator([{"CreateAccountStatuses": []}])
    org.create_account.return_value = {
        "CreateAccountStatus": {
            "Id": "car-fresh-001",
            "State": "SUCCEEDED",
            "AccountId": "123456789012",
        },
    }
    org.describe_create_account_status.return_value = {
        "CreateAccountStatus": {
            "Id": "car-fresh-001",
            "State": "SUCCEEDED",
            "AccountId": "123456789012",
        },
    }

    with patch.object(driver.__class__, "_boto_client", return_value=org):
        info = driver._create_account_sync(
            name="aws-pingcap-datalake",
            description="data lake env",
            owner_email="aws-pingcap-datalake@pingcap.com",
            budget_usd_monthly=1000.0,
            metadata={"scope_request_id": "req-1"},
            idempotency_key="req-1",
        )

    assert info.external_id == "123456789012"
    # The actual call body — assert ClientRequestToken absent.
    org.create_account.assert_called_once()
    kwargs = org.create_account.call_args.kwargs
    assert "ClientRequestToken" not in kwargs, \
        "AWS Organizations CreateAccount does not accept " \
        "ClientRequestToken — passing it makes botocore reject the request"
    # And the real params ARE passed.
    assert kwargs["Email"] == "aws-pingcap-datalake@pingcap.com"
    assert kwargs["AccountName"] == "aws-pingcap-datalake"
    assert kwargs["IamUserAccessToBilling"] == "ALLOW"


def test_create_account_sync_resumes_existing_in_flight_request() -> None:
    """If list_create_account_status surfaces an IN_PROGRESS row for
    this AccountName, the driver MUST NOT issue a second
    create_account — it polls the existing RequestId instead.
    Crash-redrive safety; without it, a worker restart mid-poll
    would mint a duplicate sub-account."""
    driver = _make_aws_driver_sso_mode()

    org = MagicMock()
    org.get_paginator.return_value = _make_paginator([{
        "CreateAccountStatuses": [
            {
                "Id": "car-existing-resume",
                "AccountName": "aws-pingcap-datalake",
                "State": "IN_PROGRESS",
            },
        ],
    }])
    # Describe transitions: first call still IN_PROGRESS, second SUCCEEDED
    org.describe_create_account_status.side_effect = [
        {"CreateAccountStatus": {
            "Id": "car-existing-resume",
            "State": "IN_PROGRESS",
        }},
        {"CreateAccountStatus": {
            "Id": "car-existing-resume",
            "State": "SUCCEEDED",
            "AccountId": "999999999999",
        }},
    ]

    with patch.object(driver.__class__, "_boto_client", return_value=org), \
         patch("time.sleep"):  # noqa: SIM117 — kill the polling sleep
        info = driver._create_account_sync(
            name="aws-pingcap-datalake",
            description=None,
            owner_email="aws-pingcap-datalake@pingcap.com",
            budget_usd_monthly=1000.0,
            metadata={},
            idempotency_key="req-1",
        )

    assert info.external_id == "999999999999"
    # KEY assertion: create_account was NEVER called — we resumed.
    org.create_account.assert_not_called()
    # Describe was called against the resumed RequestId.
    first_describe = org.describe_create_account_status.call_args_list[0]
    assert first_describe.kwargs["CreateAccountRequestId"] == "car-existing-resume"


def test_create_account_sync_iam_mode_raises_not_implemented() -> None:
    """IAM-mode (single-account) tenants can't auto-create sub-
    accounts — the API path is meaningless. Caller's
    ``failure_reason`` should show this."""
    from app.integrations.drivers.aws_cloud import AwsCloudDriver

    driver = AwsCloudDriver(
        integration_id="22222222-2222-2222-2222-222222222222",
        config={"account_mode": "iam"},
    )
    with pytest.raises(NotImplementedError) as exc_info:
        # create_scope is async — drive it via .__wrapped__ equivalent;
        # easier: hit the sync precondition via the async wrapper.
        import asyncio
        asyncio.get_event_loop().run_until_complete(
            driver.create_scope(
                name="x", description=None,
                owner_email="x@example.com",
                budget_usd_monthly=0.0,
            ),
        )
    assert "IAM" in str(exc_info.value) or "single-account" in str(exc_info.value)


# ─────────────────────────────────────────────────────────────────────
# _resolve_default_owner_role — provider dispatch
# ─────────────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_resolve_default_owner_role_gcp_returns_static_tuple(
    monkeypatch,
) -> None:
    """GCP entry is a static (role_key, display_name) tuple — no
    Identity-Center-style lookup needed."""
    from app.domains.cmdb.lifecycle_router import _resolve_default_owner_role

    out = await _resolve_default_owner_role(
        "gcp",
        integration_id="33333333-3333-3333-3333-333333333333",
        session=MagicMock(),  # never touched for static providers
    )
    assert out == ("roles/owner", "Owner")


@pytest.mark.asyncio
async def test_resolve_default_owner_role_unknown_provider_returns_none(
    monkeypatch,
) -> None:
    """A provider with no map entry → None; caller falls through to
    "no auto-grant on this provider".

    The stand-in used to be ``"azure"``, which stopped being unmapped on
    2026-08-17 (Azure sub-accounts arrive via manual-provision and were
    landing zero-IAM owners — see
    ``test_auto_owner_role_map_coverage_2026_08_17``). Assert the chosen
    provider really is absent from the map first, so this test fails as
    "pick a different provider" rather than silently asserting nothing
    the next time the map grows.
    """
    from app.domains.cmdb.lifecycle_router import (
        _DEFAULT_OWNER_ROLE_BY_PROVIDER,
        _resolve_default_owner_role,
    )

    unmapped = "ksyun"
    assert unmapped not in _DEFAULT_OWNER_ROLE_BY_PROVIDER, (
        f"{unmapped!r} now has an owner-role entry — this test needs a "
        "provider that genuinely has none"
    )
    out = await _resolve_default_owner_role(
        unmapped,
        integration_id="33333333-3333-3333-3333-333333333333",
        session=MagicMock(),
    )
    assert out is None


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("provider", "expected"),
    [
        # Domestic aliyun has NO grant_access override → base RAM impl,
        # which rejects any role_key without a "Type:Name" colon. Owner
        # must be the RAM system policy, not the bare CloudSSO "admin".
        ("aliyun", ("System:AdministratorAccess", "AdministratorAccess")),
        # AliCloud-海外 overrides grant_access and maps the bare CloudSSO
        # AccessConfiguration name "admin" → CreateAccessAssignment.
        ("alicloud_intl", ("admin", "admin")),
    ],
)
async def test_resolve_default_owner_role_aliyun_static_admin(
    provider,
    expected,
) -> None:
    """Regression (2026-06-03 / corrected 2026-06-04): Aliyun / AliCloud-海外
    were MISSING from _DEFAULT_OWNER_ROLE_BY_PROVIDER, so post-provision
    finalize was a no-op and the self-service sub-account requester landed
    with zero cloud-side IAM (prod scope 32e44d04 needed the owner grant
    applied by hand). The first fix shipped ('admin','admin') for BOTH
    providers, which silently broke the DOMESTIC path: 'admin' has no colon,
    so the RAM grant_access returned ok=False and the grant stranded PENDING.
    Each provider now resolves to the role_key shape its OWN grant_access
    consumes — domestic = System:AdministratorAccess (RAM), intl = admin
    (CloudSSO)."""
    from app.domains.cmdb.lifecycle_router import _resolve_default_owner_role

    out = await _resolve_default_owner_role(
        provider,
        integration_id="33333333-3333-3333-3333-333333333333",
        session=MagicMock(),  # static entry — session untouched
    )
    assert out == expected


@pytest.mark.asyncio
async def test_resolve_default_owner_role_aws_by_name_match(
    monkeypatch,
) -> None:
    """AWS resolves the permission-set Name → ARN at runtime by
    listing Identity Center permission sets and matching by
    display_name."""
    from app.integrations.driver_base import CloudRoleInfo

    fake_session = MagicMock()
    # session.get(Integration, ...) → return a fake integration row
    fake_integration = SimpleNamespace(
        id="44444444-4444-4444-4444-444444444444",
    )
    fake_session.get = MagicMock()

    async def _async_get(*_args, **_kwargs):
        return fake_integration

    fake_session.get = _async_get

    # Patch the integration resolver to return a config the driver
    # can consume, and the AWS driver's list_scope_roles to return
    # a permission set list containing the role we want.
    fake_resolved = SimpleNamespace(
        row=fake_integration,
        config={
            "account_mode": "sso",
            "sso_instance_arn": "arn:aws:sso:::instance/ssoins-xyz",
            "sso_region": "us-east-1",
        },
    )

    async def _fake_resolve(_session, _integration):
        return fake_resolved

    # Mock signature MUST match the real ``get_platform_setting``:
    # ``(key, default=None, *, session=None)``. The pre-2026-06-02
    # mock was ``(_session, _key, default=None)`` which silently
    # validated the WRONG production call shape (positional session
    # aliased to ``key``, key aliased to ``default``, then kwarg
    # ``default=`` re-bound ``default`` → TypeError). The test passed
    # because mock + bug agreed; prod crashed on first inline call.
    # Keyword-only ``session`` here pins the contract so a future
    # signature drift can't slip past CI again.
    async def _fake_get_platform_setting(key, default=None, *, session=None):
        return default  # fall back to "DBaaS-DevLeader-Role"

    async def _fake_list_scope_roles(_self, _scope):
        return [
            CloudRoleInfo(
                role_key="arn:aws:sso:::permissionSet/ssoins-xyz/ps-other",
                display_name="ReadOnly",
                source="permission_set",
            ),
            CloudRoleInfo(
                role_key="arn:aws:sso:::permissionSet/ssoins-xyz/ps-devleader",
                display_name="DBaaS-DevLeader-Role",
                source="permission_set",
            ),
        ]

    async def _fake_close(_self):
        return None

    monkeypatch.setattr(
        "app.integrations.service._resolve", _fake_resolve,
    )
    monkeypatch.setattr(
        "app.integrations.service.get_platform_setting",
        _fake_get_platform_setting,
    )
    monkeypatch.setattr(
        "app.integrations.drivers.aws_cloud.AwsCloudDriver.list_scope_roles",
        _fake_list_scope_roles,
    )
    monkeypatch.setattr(
        "app.integrations.drivers.aws_cloud.AwsCloudDriver.close",
        _fake_close,
    )

    from app.domains.cmdb.lifecycle_router import _resolve_default_owner_role

    out = await _resolve_default_owner_role(
        "aws",
        integration_id=str(fake_integration.id),
        session=fake_session,
    )
    assert out == (
        "arn:aws:sso:::permissionSet/ssoins-xyz/ps-devleader",
        "DBaaS-DevLeader-Role",
    )


@pytest.mark.asyncio
async def test_resolve_default_owner_role_aws_name_not_found_raises(
    monkeypatch,
) -> None:
    """If Identity Center has no permission set matching the configured
    name, raise ``_OwnerRoleUnresolvable`` — a DETERMINISTIC config miss
    (distinct from a transient driver error). The caller stamps the
    backfill backoff marker so the cron stops re-listing every tick,
    rather than picking some arbitrary other role."""
    from app.integrations.driver_base import CloudRoleInfo

    fake_session = MagicMock()
    fake_integration = SimpleNamespace(
        id="55555555-5555-5555-5555-555555555555",
    )

    async def _async_get(*_args, **_kwargs):
        return fake_integration

    fake_session.get = _async_get

    fake_resolved = SimpleNamespace(
        row=fake_integration,
        config={
            "account_mode": "sso",
            "sso_instance_arn": "arn:aws:sso:::instance/ssoins-zzz",
        },
    )

    async def _fake_resolve(_session, _integration):
        return fake_resolved

    # Mock signature MUST match the real ``get_platform_setting``:
    # ``(key, default=None, *, session=None)``. See sibling test
    # ``test_resolve_default_owner_role_aws_by_name_match`` for the
    # full rationale + prior-bug history.
    async def _fake_get_platform_setting(key, default=None, *, session=None):
        return "Brand-New-Role-That-Does-Not-Exist"

    async def _fake_list_scope_roles(_self, _scope):
        return [
            CloudRoleInfo(
                role_key="arn:other",
                display_name="ReadOnly",
                source="permission_set",
            ),
        ]

    async def _fake_close(_self):
        return None

    monkeypatch.setattr(
        "app.integrations.service._resolve", _fake_resolve,
    )
    monkeypatch.setattr(
        "app.integrations.service.get_platform_setting",
        _fake_get_platform_setting,
    )
    monkeypatch.setattr(
        "app.integrations.drivers.aws_cloud.AwsCloudDriver.list_scope_roles",
        _fake_list_scope_roles,
    )
    monkeypatch.setattr(
        "app.integrations.drivers.aws_cloud.AwsCloudDriver.close",
        _fake_close,
    )

    from app.domains.cmdb.lifecycle_router import (
        _OwnerRoleUnresolvable,
        _resolve_default_owner_role,
    )

    with pytest.raises(_OwnerRoleUnresolvable) as excinfo:
        await _resolve_default_owner_role(
            "aws",
            integration_id=str(fake_integration.id),
            session=fake_session,
        )
    assert excinfo.value.provider == "aws"
    assert "Brand-New-Role-That-Does-Not-Exist" in excinfo.value.reason