"""Regression: revoking an API-key assignment that deletes its backing
Bailian / DashScope key on Aliyun must also retire the local
``bailian_api_keys`` mirror row.

Production drift (2026-06-18): an admin 回收'd an adopted Bailian key in
"API Key 管理". ``_revoke_cloud_resources`` called
``BailianClient.delete_api_key`` and the real Aliyun key was gone — but
the ``bailian_api_keys`` registry row stayed ``status='active'`` /
``deleted_at IS NULL``. The nightly ``bailian.api_key_sync`` deliberately
never soft-deletes assignment-linked rows (autogrant rows carry RAM
AccessKey IDs absent from the ModelStudio listing), so the row would
linger forever, leaving the cost dashboard / admin list showing a key
that no longer exists upstream.

Fix: the revoke path now mirrors the cloud-side deletion into the
registry via ``BailianService.mark_api_key_deleted_on_revoke`` inside the
same transaction as the status flip.

Production-grade per CLAUDE.md: real assignment + real bailian_api_keys
rows + real audit/registry SQL session. Only the Aliyun SDK call
(``BailianClient.delete_api_key``) is stubbed — no network in unit tests.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime
from decimal import Decimal
from types import SimpleNamespace

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.models import AuditLog
from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider
from app.domains.apikey.service import ApiKeyService
from app.domains.bailian.models import BailianApiKey, BailianWorkspace
from app.domains.bailian.service import BailianService
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration


def _admin_auth(person: IdentityPerson) -> SimpleNamespace:
    return SimpleNamespace(id=person.id, role_names=["it_admin"])


async def _seed(
    db: AsyncSession, *, api_key_id: str, pre_deleted: bool = False,
    with_registry_row: bool = True,
) -> tuple[ApiKeyAssignment, BailianApiKey | None]:
    person = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name="adopter",
        work_email=f"adopter-{uuid.uuid4().hex[:6]}@pingcap.com",
        employment_status="active",
    )
    db.add(person)
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="alicloud_intl",
        name="Aliyun intl test",
        slug=f"aliyun-revoke-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=[],
        config_public={},
    )
    db.add(integ)
    provider = ApiKeyProvider(
        id=uuid.uuid4(), name="Aliyun Bailian", description="",
        is_active=True, sort_order=0,
    )
    db.add(provider)
    await db.commit()

    assignment = ApiKeyAssignment(
        id=uuid.uuid4(),
        provider_id=provider.id,
        assigned_to=person.id,
        status="active",
        request_number=f"AK-{uuid.uuid4().hex[:8].upper()}",
        project_name="LLM eval",
        reason="adopted bailian key",
        monthly_budget_usd=Decimal("0"),
        iam_integration_id=str(integ.id),
        detail={
            "source": "adopted",
            "bailian": {
                "adopted": True,
                "api_key_id": api_key_id,
                "integration_id": str(integ.id),
                "workspace_id": "ws-1",
            },
        },
    )
    db.add(assignment)

    registry_row = None
    if with_registry_row:
        registry_row = BailianApiKey(
            id=uuid.uuid4(),
            integration_id=str(integ.id),
            api_key_id=api_key_id,
            display_name=f"apikey-{api_key_id}",
            status="deleted" if pre_deleted else "active",
            assignment_id=assignment.id,
            last_synced_at=datetime.now(UTC),
            deleted_at=datetime.now(UTC) if pre_deleted else None,
        )
        db.add(registry_row)

    await db.commit()
    return assignment, registry_row


def _stub_aliyun(monkeypatch):
    """Stub the Aliyun SDK call so revoke's cloud teardown succeeds
    without a network hop. delete_api_key reports the key as deleted."""
    class _FakeClient:
        def __init__(self, *a, **k):
            pass

        async def delete_api_key(
            self, *, api_key_id, workspace_id=None, region=None,
        ):
            return {"already_absent": False, "api_key_id": api_key_id}

    monkeypatch.setattr(
        "app.domains.bailian.client.BailianClient", _FakeClient,
    )
    monkeypatch.setattr(
        "app.domains.bailian.client.resolve_credentials",
        lambda integration: SimpleNamespace(),
    )


@pytest.mark.asyncio
async def test_revoke_retires_bailian_registry_row(
    db_session: AsyncSession, monkeypatch,
):
    assignment, registry_row = await _seed(db_session, api_key_id="121285")
    _stub_aliyun(monkeypatch)

    response = await ApiKeyService(db_session).revoke_key(
        assignment.id,
        reason="admin reclaim",
        current_user=_admin_auth(
            await db_session.get(IdentityPerson, assignment.assigned_to),
        ),
        trace_id="t-bailian-revoke",
        ip_address="127.0.0.1",
        user_agent="pytest",
    )
    assert response.status == "revoked"

    # Mirror row retired in-place, in the same txn as the revoke.
    await db_session.refresh(registry_row)
    assert registry_row.status == "deleted"
    assert registry_row.deleted_at is not None

    # Audit summary records the registry retire for forensics.
    audit = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "apikey.key.revoke",
            AuditLog.resource_id == str(assignment.id),
        )
    )).scalars().one()
    attempted = (audit.detail or {}).get("cloud_revoke", {}).get("attempted", [])
    registry_entries = [
        a for a in attempted if a.get("kind") == "bailian.api_key.registry"
    ]
    assert len(registry_entries) == 1
    assert registry_entries[0]["result"]["matched"] is True
    assert registry_entries[0]["result"]["already_deleted"] is False


@pytest.mark.asyncio
async def test_revoke_idempotent_when_registry_already_deleted(
    db_session: AsyncSession, monkeypatch,
):
    """An already soft-deleted mirror row is a no-op — matched but
    already_deleted, no crash, revoke still succeeds."""
    assignment, registry_row = await _seed(
        db_session, api_key_id="999001", pre_deleted=True,
    )
    _stub_aliyun(monkeypatch)

    await ApiKeyService(db_session).revoke_key(
        assignment.id,
        reason="admin reclaim",
        current_user=_admin_auth(
            await db_session.get(IdentityPerson, assignment.assigned_to),
        ),
        trace_id="t-bailian-revoke-2",
        ip_address="127.0.0.1",
        user_agent="pytest",
    )
    await db_session.refresh(registry_row)
    assert registry_row.status == "deleted"


@pytest.mark.asyncio
async def test_revoke_no_registry_row_does_not_fail(
    db_session: AsyncSession, monkeypatch,
):
    """Granted (autogrant) keys may have no ModelStudio-keyed registry
    row, or the row was never synced. Revoke must not fail — the mirror
    cleanup reports matched=False and the lifecycle endpoint completes."""
    assignment, _ = await _seed(
        db_session, api_key_id="888002", with_registry_row=False,
    )
    _stub_aliyun(monkeypatch)

    response = await ApiKeyService(db_session).revoke_key(
        assignment.id,
        reason="admin reclaim",
        current_user=_admin_auth(
            await db_session.get(IdentityPerson, assignment.assigned_to),
        ),
        trace_id="t-bailian-revoke-3",
        ip_address="127.0.0.1",
        user_agent="pytest",
    )
    assert response.status == "revoked"
    audit = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "apikey.key.revoke",
            AuditLog.resource_id == str(assignment.id),
        )
    )).scalars().one()
    attempted = (audit.detail or {}).get("cloud_revoke", {}).get("attempted", [])
    registry_entries = [
        a for a in attempted if a.get("kind") == "bailian.api_key.registry"
    ]
    assert len(registry_entries) == 1
    assert registry_entries[0]["result"]["matched"] is False


@pytest.mark.asyncio
async def test_sync_api_keys_does_not_resurrect_revoke_retired_row(
    db_session: AsyncSession, monkeypatch,
):
    """Reverse-drift regression: Aliyun DeleteApiKey is eventually consistent,
    so a just-revoked adopted key can still surface in ListApiKeys for a
    window. ``sync_api_keys`` must NOT un-soft-delete a revoke-retired
    (deleted_at + assignment_id) mirror row — doing so reopens the exact
    two-table drift the revoke retire closed (row active again while its
    assignment is revoked)."""
    api_key_id = "770099"
    assignment, registry_row = await _seed(
        db_session, api_key_id=api_key_id, pre_deleted=True,
    )
    integration_id = registry_row.integration_id
    # The workspace the upstream listing fans out over.
    db_session.add(BailianWorkspace(
        id=uuid.uuid4(), integration_id=integration_id,
        workspace_id="ws-1", name="WS 1",
    ))
    await db_session.commit()

    class _StubClient:
        async def list_api_keys(self, *, workspace_id, region=None):
            # Aliyun still lists the just-deleted key (propagation lag).
            return [{
                "api_key_id": api_key_id,
                "display_name": "still-listed",
                "created_by": "ram-user",
                "disabled": False,
                "workspace_id": workspace_id,
                "raw": {},
            }]

    from unittest.mock import AsyncMock
    svc = BailianService(db_session)
    monkeypatch.setattr(svc, "_get_client", AsyncMock(return_value=_StubClient()))

    await svc.sync_api_keys(integration_id=integration_id)
    await db_session.commit()
    await db_session.refresh(registry_row)

    # Stays retired despite reappearing upstream.
    assert registry_row.deleted_at is not None
    assert registry_row.status == "deleted"