"""AWS driver: ``list_principal_assignments`` — the live-principal
enumeration that powers the offboarding leaver-sweep.

Resolves a person's email → Identity Store user id, then walks
``list_account_assignments_for_principal`` to return every account /
permission-set they hold. A principal already deleted from the Identity
Store (SCIM ran first) yields ``[]`` — there's nothing left to sweep.
"""
from __future__ import annotations

from unittest.mock import MagicMock, patch

import pytest

from app.integrations.driver_base import CloudPrincipalNotFoundError
from app.integrations.drivers.aws_cloud import AwsCloudDriver

_EMAIL = "leaver@pingcap.com"
_USER_ID = "95670a3c83-c108d80c-cbbe-4c76-a5c7-d95928e4b566"
_INSTANCE = "arn:aws:sso:::instance/ssoins-x"
_STORE = "d-1234567890"
_PS = "arn:aws:sso:::permissionSet/ssoins-x/ps-dbaasleader"

_CFG = {
    "access_key_id": "AKIA-test",
    "secret_access_key": "test-sk",
    "default_region": "ap-southeast-1",
    "account_mode": "sso",
    "sso_instance_arn": _INSTANCE,
    "sso_identity_store_id": _STORE,
}


def _driver() -> AwsCloudDriver:
    return AwsCloudDriver(integration_id="aws-test", config=dict(_CFG))


def _fake_identitystore(user_id: str | None) -> MagicMock:
    ids = MagicMock()
    ids.list_users.return_value = (
        {"Users": [{"UserId": user_id}]} if user_id else {"Users": []}
    )
    return ids


def _fake_sso_with_assignments(pages: list[dict]) -> MagicMock:
    sso = MagicMock()
    paginator = MagicMock()
    paginator.paginate.return_value = pages
    sso.get_paginator.return_value = paginator
    sso.describe_permission_set.return_value = {
        "PermissionSet": {"Name": "DBaaS-DevLeader-Role"},
    }
    return sso


@pytest.mark.asyncio
async def test_list_principal_assignments_returns_all_held() -> None:
    d = _driver()
    ids = _fake_identitystore(_USER_ID)
    sso = _fake_sso_with_assignments([
        {"AccountAssignments": [
            {"AccountId": "448141240262", "PermissionSetArn": _PS},
            {"AccountId": "123456789012", "PermissionSetArn": _PS},
        ]},
    ])

    def _client(service, region=None):
        return ids if service == "identitystore" else sso

    with patch.object(d, "_boto_client", side_effect=_client):
        out = await d.list_principal_assignments(_EMAIL)

    assert len(out) == 2
    accounts = {a.account_id for a in out}
    assert accounts == {"448141240262", "123456789012"}
    for a in out:
        assert a.principal_id == _USER_ID
        assert a.role_key == _PS
        assert a.role_display_name == "DBaaS-DevLeader-Role"
        assert a.external_ref == f"{a.account_id}::{_PS}::{_USER_ID}"


@pytest.mark.asyncio
async def test_list_principal_assignments_empty_when_principal_gone() -> None:
    """Identity Store has no user for the email (SCIM already deleted the
    principal) → nothing to sweep, return []."""
    d = _driver()
    ids = _fake_identitystore(None)
    # ``_resolve_identity_store_user_id`` paginates a client-side scan
    # when the UserName filter misses — make that empty too.
    ids.list_users.side_effect = [
        {"Users": []},  # UserName filter
        {"Users": []},  # paginated scan page 1 (no NextToken)
    ]
    sso = _fake_sso_with_assignments([])

    def _client(service, region=None):
        return ids if service == "identitystore" else sso

    with patch.object(d, "_boto_client", side_effect=_client):
        out = await d.list_principal_assignments(_EMAIL)

    assert out == []
    # Never even asked sso-admin for assignments — bailed at resolution.
    sso.get_paginator.assert_not_called()


@pytest.mark.asyncio
async def test_list_principal_assignments_empty_in_iam_mode() -> None:
    """IAM mode has no Identity Center principal index — returns []."""
    cfg = dict(_CFG)
    cfg["account_mode"] = "iam"
    d = AwsCloudDriver(integration_id="aws-test", config=cfg)
    out = await d.list_principal_assignments(_EMAIL)
    assert out == []


@pytest.mark.asyncio
async def test_resolve_raises_are_surfaced_not_swallowed() -> None:
    """A non-'not found' resolution failure must propagate — the sweep
    must not report a clean no-op while the leaver keeps access."""
    d = _driver()

    def _boom(email):
        raise RuntimeError("throttled")

    with patch.object(
        d, "_resolve_identity_store_user_id", side_effect=_boom,
    ):
        with pytest.raises(RuntimeError, match="throttled"):
            await d.list_principal_assignments(_EMAIL)


@pytest.mark.asyncio
async def test_resolve_not_found_is_swallowed_to_empty() -> None:
    d = _driver()
    with patch.object(
        d, "_resolve_identity_store_user_id",
        side_effect=CloudPrincipalNotFoundError("gone"),
    ):
        out = await d.list_principal_assignments(_EMAIL)
    assert out == []