"""Bedrock auto-grant: on approved-status callback, the helper creates
an IAM user via the AWS SG cloud integration and writes the AK/SK
onto the assignment as a structured ``key_value`` blob.

We don't hit AWS — ``AWSProvider.create_iam_user_with_policy`` is
patched to return a canned ``AccessKey`` dict.
"""
from __future__ import annotations

import uuid
from decimal import Decimal
from unittest.mock import AsyncMock

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.apikey.bedrock_autogrant import (
    _is_bedrock_provider,
    _sanitize_user_name,
    maybe_auto_grant_bedrock,
)
from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration, PlatformSetting


async def _make_cloud_integ(db_session: AsyncSession, slug: str = "aws-sg") -> Integration:
    # We skip real encryption; decrypt path checks only when
    # secrets_encrypted is truthy. The autogrant helper tolerates an
    # integration with no secrets because the monkey-patched AWS driver
    # call never actually reads config.
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="aws",
        name="AWS SG (Singapore)",
        slug=slug,
        enabled=True,
        is_primary=False,
        capabilities=["servers", "iam"],
        config_public={"default_region": "ap-southeast-1"},
        # Non-empty cipher triggers the decrypt_fields code path which
        # our test patches to return fake AK/SK — the bytes themselves
        # are never read because the patch intercepts.
        secrets_encrypted="ciphertext-placeholder",
        secrets_nonce="nonce-placeholder",
    )
    db_session.add(integ)
    await db_session.commit()
    return integ


async def _make_person(db_session: AsyncSession) -> IdentityPerson:
    # hris_person_id + source_system are NOT NULL per the identity_persons
    # schema; fabricate a stable pair per test so the FK lookups in the
    # bedrock grant path resolve deterministically.
    person = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name="Shuai Feng",
        work_email="shuai.feng@pingcap.com",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.commit()
    return person


async def _make_bedrock_assignment(
    db_session: AsyncSession, person_id: uuid.UUID,
) -> ApiKeyAssignment:
    provider = ApiKeyProvider(
        id=uuid.uuid4(),
        name="AWS Bedrock",
        description="Bedrock access",
        is_active=True,
        sort_order=0,
    )
    db_session.add(provider)
    await db_session.commit()
    assignment = ApiKeyAssignment(
        id=uuid.uuid4(),
        provider_id=provider.id,
        assigned_to=person_id,
        status="approved",
        request_number=f"AK-20260422-{uuid.uuid4().hex[:4].upper()}",
        project_name="LLM eval",
        reason="Bedrock model benchmarking",
        monthly_budget_usd=Decimal("200.00"),
    )
    db_session.add(assignment)
    await db_session.commit()
    return assignment


@pytest.mark.asyncio
async def test_autogrant_succeeds_for_bedrock(
    db_session: AsyncSession, monkeypatch,
):
    integ = await _make_cloud_integ(db_session)
    person = await _make_person(db_session)
    assignment = await _make_bedrock_assignment(db_session, person.id)

    captured: dict = {}

    async def fake_create(
        self, *, user_name, policy_name, policy_document, tags, group_names,
    ):
        captured["user_name"] = user_name
        captured["tags"] = tags
        captured["group_names"] = list(group_names or [])
        return {
            "access_key_id": "AKIAFAKE",
            "secret_access_key": "sk-secret-fake",
            "user_arn": f"arn:aws:iam::000:user/{user_name}",
            "groups_attached": list(group_names or []),
        }

    from app.domains.cmdb.providers.aws import AWSProvider
    monkeypatch.setattr(
        AWSProvider, "create_iam_user_with_policy", fake_create,
    )
    monkeypatch.setattr(AWSProvider, "close", AsyncMock())
    # Skip real credential decryption for this test.
    # ``decrypt_fields`` post-2026-05-06 takes ``integration_id`` as a
    # keyword arg (AEAD AAD binding); accept + ignore here.
    monkeypatch.setattr(
        "app.domains.apikey.bedrock_autogrant.decrypt_fields",
        lambda cipher, nonce, *, integration_id=None: {
            "access_key_id": "x", "secret_access_key": "y",
        },
    )

    notify_spy = AsyncMock()
    from app.notifications.service import NotificationService
    monkeypatch.setattr(NotificationService, "dispatch", notify_spy)

    result = await maybe_auto_grant_bedrock(db_session, assignment)
    assert result["status"] == "granted"
    assert assignment.status == "active"
    assert assignment.key_value is not None
    # ``key_value`` is at-rest encrypted (AES-256-GCM, JSON envelope —
    # see ``_encrypt_key_value`` / ``_decrypt_key_value``). Decrypt
    # before asserting on the cleartext payload.
    from app.domains.apikey.service import _decrypt_key_value
    # Round-20 audit (A2F6): Bedrock auto-grant now binds AES-GCM AAD
    # to the assignment's id; the corresponding decrypt MUST pass the
    # same assignment_id or the AEAD check fails and the function
    # returns the raw stored ciphertext envelope.
    decrypted_kv = _decrypt_key_value(
        assignment.key_value, assignment_id=assignment.id,
    )
    assert "ACCESS_KEY_ID=AKIAFAKE" in decrypted_kv
    assert "SECRET_ACCESS_KEY=sk-secret-fake" in decrypted_kv
    detail = assignment.detail or {}
    assert detail.get("bedrock", {}).get("granted") is True
    assert detail["bedrock"]["integration_slug"] == integ.slug

    # ── IAM upsert: the freshly-minted IAM user is visible in
    # cmdb_iam_users immediately, without a separate IAM sync run.
    from sqlalchemy import select as _select

    from app.domains.cmdb.models import CmdbIamUser
    iam_row = (await db_session.execute(
        _select(CmdbIamUser).where(
            CmdbIamUser.integration_id == str(integ.id),
            CmdbIamUser.user_name == captured["user_name"],
        )
    )).scalar_one()
    assert iam_row.tags["__source"] == "helix_autogrant.bedrock"
    assert iam_row.tags["helix-project-name"] == "LLM eval"
    assert iam_row.owner_person_id == str(person.id)
    assert assignment.iam_user_id == iam_row.id

    # ── Notification dispatch happened exactly once with the right
    # template_key + recipient — autogrant is now expected to actively
    # tell the assignee that the credential is ready (previously they
    # had to remember they applied and check /me/api-keys).
    assert notify_spy.await_count == 1
    env = notify_spy.await_args.args[0]
    assert env.recipient.id == person.id
    assert env.template_key == "apikey.ready"
    assert env.related_resource_type == "api_key_assignment"
    assert env.related_resource_id == assignment.id
    # IAM username derives from project_name ("LLM eval"), not the
    # employee's email — confirms the rename to project-based naming.
    assert captured["user_name"].startswith("bedrock-LLM_eval-")
    assert "shuai" not in captured["user_name"].lower()
    assert captured["tags"]["helix-project-name"] == "LLM eval"
    # AWS IAM-principal-tag cost allocation: ``Owner`` carries the
    # **applicant's work_email** so Cost Explorer's
    # ``iamPrincipal/Owner`` dimension splits Bedrock spend per
    # employee. Earlier revisions tagged ``Owner=<project>`` which
    # collapsed multiple applicants into one bucket and made the
    # matching ``CmdbCostAllocationRule`` ambiguous.
    assert captured["tags"]["Owner"] == person.work_email
    assert captured["tags"]["helix-owner-email"] == person.work_email
    # Default IAM group is attached.
    assert captured["group_names"] == ["DBaaS-DevAdmin-Role"]
    assert detail["bedrock"]["iam_group"] == "DBaaS-DevAdmin-Role"

    # ── Cost-allocation rule: a priority-10 rule keyed on
    # (Owner=<email>) was committed in the same transaction as
    # ``granted=True`` so the next ``cmdb.cost_facts_sync`` run
    # immediately stamps Bedrock facts to this applicant's
    # ``cost_owner_person_id``. Without this rule the principal-tag
    # mode produces facts with ``tags.Owner=<email>`` that nobody
    # owns — i.e. the per-employee chargeback we're after never
    # materialises. Mirrors Azure OpenAI auto-grant.
    from app.domains.cmdb.models import CmdbCostAllocationRule
    rules = (await db_session.execute(
        _select(CmdbCostAllocationRule).where(
            CmdbCostAllocationRule.match_tag_key == "Owner",
            CmdbCostAllocationRule.match_tag_value == person.work_email,
            CmdbCostAllocationRule.assign_person_id == str(person.id),
        )
    )).scalars().all()
    assert len(rules) == 1, (
        f"expected exactly one allocation rule for the applicant; "
        f"got {len(rules)}: {[(r.name, r.priority) for r in rules]}"
    )
    rule = rules[0]
    assert rule.priority == 10
    assert rule.match_integration_ids == [str(integ.id)]
    assert rule.match_provider == "aws"
    assert rule.enabled is True
    # The autogrant returns the rule id in its detail JSON so
    # admins / audits can trace fact rows back to the spawning
    # request without a separate JOIN.
    assert detail["bedrock"]["allocation_rule_id"] == str(rule.id)
    assert detail["bedrock"]["owner_tag_value"] == person.work_email


@pytest.mark.asyncio
async def test_autogrant_iam_group_overridable_via_platform_setting(
    db_session: AsyncSession, monkeypatch,
):
    """The standing IAM group can be overridden per-environment via
    ``bedrock.provisioning_iam_group``; an empty value disables group
    attachment entirely (some tenants prefer per-user policies only)."""
    await _make_cloud_integ(db_session)
    person = await _make_person(db_session)
    assignment = await _make_bedrock_assignment(db_session, person.id)
    db_session.add(PlatformSetting(
        key="bedrock.provisioning_iam_group", value="",
    ))
    await db_session.commit()

    captured: dict = {}

    async def fake_create(
        self, *, user_name, policy_name, policy_document, tags, group_names,
    ):
        captured["group_names"] = list(group_names or [])
        return {
            "access_key_id": "AK",
            "secret_access_key": "SK",
            "user_arn": f"arn:aws:iam::000:user/{user_name}",
            "groups_attached": [],
        }

    from app.domains.cmdb.providers.aws import AWSProvider
    monkeypatch.setattr(
        AWSProvider, "create_iam_user_with_policy", fake_create,
    )
    monkeypatch.setattr(AWSProvider, "close", AsyncMock())
    monkeypatch.setattr(
        "app.domains.apikey.bedrock_autogrant.decrypt_fields",
        lambda cipher, nonce, *, integration_id=None: {
            "access_key_id": "x", "secret_access_key": "y",
        },
    )

    result = await maybe_auto_grant_bedrock(db_session, assignment)
    assert result["status"] == "granted"
    # Empty platform_setting → no group attached.
    assert captured["group_names"] == []
    assert assignment.detail["bedrock"]["iam_group"] is None


def test_sanitize_user_name_project_based() -> None:
    # Spaces become underscores; alphanumerics preserved.
    assert _sanitize_user_name("AI Support", "AK-20260429-F446") == (
        "bedrock-AI_Support-20260429-F446"
    )
    # Multiple spaces collapse to one underscore.
    assert _sanitize_user_name("  AI   Support  ", "AK-20260429-F446") == (
        "bedrock-AI_Support-20260429-F446"
    )
    # Slashes / non-IAM punctuation are dropped.
    assert _sanitize_user_name("AI/Support+v2", "AK-20260429-F446") == (
        "bedrock-AISupportv2-20260429-F446"
    )
    # CJK-only project names fall through to the suffix-only form.
    assert _sanitize_user_name("项目名称", "AK-20260429-F446") == (
        "bedrock-20260429-F446"
    )
    # Leading/trailing punctuation gets stripped.
    assert _sanitize_user_name("--hello--", "AK-20260429-F446") == (
        "bedrock-hello-20260429-F446"
    )
    # Long names truncate to fit within 64 chars.
    long_name = "x" * 200
    out = _sanitize_user_name(long_name, "AK-20260429-F446")
    assert len(out) <= 64
    assert out.endswith("-20260429-F446")
    assert out.startswith("bedrock-x")


@pytest.mark.asyncio
async def test_autogrant_noop_for_non_bedrock(
    db_session: AsyncSession,
):
    person = await _make_person(db_session)
    other_provider = ApiKeyProvider(
        id=uuid.uuid4(),
        name="OpenAI",
        description="", is_active=True, sort_order=0,
    )
    db_session.add(other_provider)
    await db_session.commit()
    assignment = ApiKeyAssignment(
        id=uuid.uuid4(),
        provider_id=other_provider.id,
        assigned_to=person.id,
        status="approved",
        request_number="AK-X",
        project_name="x",
        reason="x",
        monthly_budget_usd=Decimal("10.00"),
    )
    db_session.add(assignment)
    await db_session.commit()

    result = await maybe_auto_grant_bedrock(db_session, assignment)
    assert result == {"status": "not-bedrock"}
    assert assignment.key_value is None


@pytest.mark.asyncio
async def test_autogrant_skipped_when_no_integration(
    db_session: AsyncSession,
):
    person = await _make_person(db_session)
    assignment = await _make_bedrock_assignment(db_session, person.id)
    # No cloud integration seeded — helper logs the miss in detail.
    result = await maybe_auto_grant_bedrock(db_session, assignment)
    assert result["status"] == "no-integration"
    detail = assignment.detail or {}
    assert detail.get("bedrock", {}).get("granted") is False


@pytest.mark.asyncio
async def test_autogrant_idempotent_when_already_granted(
    db_session: AsyncSession,
):
    person = await _make_person(db_session)
    assignment = await _make_bedrock_assignment(db_session, person.id)
    assignment.detail = {"bedrock": {"granted": True}}
    await db_session.commit()

    result = await maybe_auto_grant_bedrock(db_session, assignment)
    assert result == {"status": "already-granted"}


@pytest.mark.asyncio
async def test_autogrant_refuses_retry_when_in_flight(
    db_session: AsyncSession,
):
    """Leak-safety: a previous callback crashed after the AWS call but
    before the Helix commit; the in-flight marker stops us from
    minting a second IAM user on retry."""
    person = await _make_person(db_session)
    assignment = await _make_bedrock_assignment(db_session, person.id)
    assignment.detail = {"bedrock": {"in_flight": True, "user_name_attempted": "bedrock-foo-bar"}}
    await db_session.commit()

    result = await maybe_auto_grant_bedrock(db_session, assignment)
    assert result == {"status": "in-flight-needs-manual-recovery"}


# ─── Pure-function tests ────────────────────────────────────────────


def test_is_bedrock_provider_normalises_whitespace() -> None:
    """Same matcher class as the Azure OpenAI prod incident
    (AK-20260509-EF8A): admin-typed double space / NBSP / full-width /
    tab in the provider name must not bypass the autogrant."""

    class _P:
        def __init__(self, name: str) -> None:
            self.name = name

    assert _is_bedrock_provider(_P("AWS Bedrock"))
    assert _is_bedrock_provider(_P("AWS  Bedrock"))  # 2 spaces
    assert _is_bedrock_provider(_P("AWS Bedrock"))  # NBSP
    assert _is_bedrock_provider(_P("AWS　Bedrock"))  # full-width
    assert _is_bedrock_provider(_P("AWS\tBedrock"))  # tab
    assert _is_bedrock_provider(_P("  Bedrock  "))  # padding
    assert _is_bedrock_provider(_P("Amazon Bedrock (SG)"))
    assert not _is_bedrock_provider(_P("Azure OpenAI"))
    assert not _is_bedrock_provider(_P("Aliyun Bailian"))
    assert not _is_bedrock_provider(None)