"""Coverage for ``adopt_orphan_bailian_keys`` (存量百炼 Key 纳管).

The reconciler turns owned-but-orphan ``bailian_api_keys`` rows (created
directly in the Aliyun console, owner pinned via 编辑负责人, but never
linked to an :class:`ApiKeyAssignment`) into first-class assignment rows
so they surface in the central registry + each owner's self-service view.

Exercised against a real DB row + seeded provider / person / integration:

  1. An owned orphan is adopted → assignment minted (active, AK-ADOPT-,
     detail.source='adopted', no key_value) and the inventory row
     back-linked.
  2. Owner-less orphans are left alone and counted in skipped_no_owner.
  3. Re-running is idempotent — no duplicate assignment, no error on the
     request_number unique key; a lost back-link is re-attached.

Run isolated:
    PYTEST_XDIST_WORKER=bailianadopt uv run --python 3.12 --all-extras \\
        --group dev pytest tests/test_apikey/test_bailian_adopt.py
"""
from __future__ import annotations

import uuid

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

from app.domains.apikey.bailian_adopt import adopt_orphan_bailian_keys
from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider
from app.domains.bailian.models import BailianApiKey
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration


async def _seed_integration(db: AsyncSession) -> str:
    iid = uuid.uuid4()
    db.add(Integration(
        id=iid, kind="cloud", provider="alicloud_intl",
        name=f"bailian-{str(iid)[:6]}", slug=f"bl-{str(iid)[:6]}",
        enabled=True, is_primary=False,
    ))
    await db.commit()
    return str(iid)


async def _seed_provider(db: AsyncSession) -> ApiKeyProvider:
    provider = (await db.execute(
        select(ApiKeyProvider).where(
            ApiKeyProvider.name == "Alibaba Cloud Model Studio",
        )
    )).scalar_one_or_none()
    if provider is None:
        provider = ApiKeyProvider(name="Alibaba Cloud Model Studio", is_active=True)
        db.add(provider)
        await db.commit()
    return provider


async def _seed_person(db: AsyncSession) -> IdentityPerson:
    pid = uuid.uuid4()
    person = IdentityPerson(
        id=pid,
        hris_person_id=f"hp-{str(pid)[:8]}",
        work_email=f"{str(pid)[:8]}@pingcap.com",
        display_name="Adopt Tester",
        employment_status="active",
    )
    db.add(person)
    await db.commit()
    return person


@pytest.mark.asyncio
async def test_adopts_owned_orphan(db_session: AsyncSession):
    iid = await _seed_integration(db_session)
    await _seed_provider(db_session)
    person = await _seed_person(db_session)

    key = BailianApiKey(
        integration_id=iid,
        api_key_id="998001",
        display_name="console-key-for-tester",
        owner_person_id=str(person.id),
        status="active",
    )
    db_session.add(key)
    await db_session.commit()

    result = await adopt_orphan_bailian_keys(db_session, integration_id=iid)
    await db_session.commit()

    assert result["adopted"] == 1
    assert result["scanned"] == 1
    assert result["skipped_no_owner"] == 0

    await db_session.refresh(key)
    assert key.assignment_id is not None

    assignment = await db_session.get(ApiKeyAssignment, key.assignment_id)
    assert assignment is not None
    assert assignment.status == "active"
    assert assignment.assigned_to == person.id
    assert assignment.request_number == "AK-ADOPT-998001"
    assert assignment.key_value is None
    assert assignment.approval_request_id is None
    assert assignment.iam_integration_id == iid
    assert (assignment.detail or {}).get("source") == "adopted"
    assert (assignment.detail or {}).get("api_key_id") == "998001"
    # Revoke delete-handle: lets 回收 DeleteApiKey for-real, but must NOT
    # carry iam_user_name (the shared RAM sub-account is never torn down).
    blob = (assignment.detail or {}).get("bailian") or {}
    assert blob.get("adopted") is True
    assert blob.get("api_key_id") == "998001"
    assert blob.get("integration_id") == iid
    assert "iam_user_name" not in blob
    assert "granted" not in blob


@pytest.mark.asyncio
async def test_enriches_legacy_adopted_row_missing_blob(db_session: AsyncSession):
    """A row adopted before the revoke-handle existed (detail.bailian
    absent) is self-healed on the next run so 回收 can delete cloud-side."""
    iid = await _seed_integration(db_session)
    provider = await _seed_provider(db_session)
    person = await _seed_person(db_session)

    key = BailianApiKey(
        integration_id=iid, api_key_id="998004",
        display_name="legacy-adopted", owner_person_id=str(person.id),
        status="active",
    )
    db_session.add(key)
    await db_session.flush()
    # Simulate a pre-enrichment adopted assignment: linked, but detail
    # has no ``bailian`` blob.
    legacy = ApiKeyAssignment(
        provider_id=provider.id, assigned_to=person.id, status="active",
        request_number="AK-ADOPT-998004", project_name="legacy-adopted",
        reason="adopted", monthly_budget_usd=0,
        detail={"source": "adopted", "api_key_id": "998004"},
    )
    db_session.add(legacy)
    await db_session.flush()
    key.assignment_id = legacy.id
    await db_session.commit()

    result = await adopt_orphan_bailian_keys(db_session, integration_id=iid)
    await db_session.commit()
    assert result["adopted"] == 0
    assert result["enriched"] == 1

    await db_session.refresh(legacy)
    blob = (legacy.detail or {}).get("bailian") or {}
    assert blob.get("adopted") is True
    assert blob.get("api_key_id") == "998004"


@pytest.mark.asyncio
async def test_per_row_savepoint_isolates_failures(db_session, monkeypatch):
    """Regression: a flush/audit failure on ONE candidate must not poison
    the shared session (the nightly ``bailian.api_key_sync`` commits it
    once for every integration's writes) and must not block the OTHER
    candidates' adoption — per-row SAVEPOINT isolation. The summary
    counters must also reflect only the rows that actually committed.
    """
    iid = await _seed_integration(db_session)
    await _seed_provider(db_session)
    p_good = await _seed_person(db_session)
    p_bad = await _seed_person(db_session)
    good = BailianApiKey(
        integration_id=iid, api_key_id="998010", display_name="good",
        owner_person_id=str(p_good.id), status="active",
    )
    bad = BailianApiKey(
        integration_id=iid, api_key_id="998011", display_name="bad",
        owner_person_id=str(p_bad.id), status="active",
    )
    db_session.add_all([good, bad])
    await db_session.commit()

    # Inject a genuine DB error during ONLY the "bad" key's savepoint,
    # AFTER its assignment row has already been flushed — a real round-trip
    # error (the failure mode the per-row SAVEPOINT exists for), not a bare
    # Python raise. The savepoint must roll back this row's flushed insert
    # and leave the connection at a clean boundary for the rest.
    from sqlalchemy import text

    from app.domains.apikey import bailian_adopt as mod
    real_log = mod.AuditService.log

    async def _flaky_log(self, *a, **kw):
        if (kw.get("detail") or {}).get("api_key_id") == "998011":
            # Real DB error (table doesn't exist → 1146), completes the
            # protocol round-trip → recoverable via savepoint rollback.
            await db_session.execute(text("SELECT 1 FROM _no_such_table_xyz"))
        return await real_log(self, *a, **kw)

    monkeypatch.setattr(mod.AuditService, "log", _flaky_log)

    result = await adopt_orphan_bailian_keys(db_session, integration_id=iid)
    # The shared session MUST still be usable — a poisoned transaction
    # would raise PendingRollbackError here.
    await db_session.commit()

    # Exactly one row committed; the failed row rolled back cleanly and is
    # left orphan for the next idempotent tick. Counter reflects reality.
    assert result["adopted"] == 1
    await db_session.refresh(good)
    await db_session.refresh(bad)
    assert good.assignment_id is not None
    assert bad.assignment_id is None


@pytest.mark.asyncio
async def test_skips_orphan_without_owner(db_session: AsyncSession):
    iid = await _seed_integration(db_session)
    await _seed_provider(db_session)

    key = BailianApiKey(
        integration_id=iid,
        api_key_id="998002",
        display_name="ownerless-console-key",
        owner_person_id=None,
        status="active",
    )
    db_session.add(key)
    await db_session.commit()

    result = await adopt_orphan_bailian_keys(db_session, integration_id=iid)
    await db_session.commit()

    assert result["adopted"] == 0
    assert result["skipped_no_owner"] == 1

    await db_session.refresh(key)
    assert key.assignment_id is None


@pytest.mark.asyncio
async def test_idempotent_on_rerun(db_session: AsyncSession):
    iid = await _seed_integration(db_session)
    await _seed_provider(db_session)
    person = await _seed_person(db_session)

    key = BailianApiKey(
        integration_id=iid,
        api_key_id="998003",
        display_name="console-key-rerun",
        owner_person_id=str(person.id),
        status="active",
    )
    db_session.add(key)
    await db_session.commit()

    first = await adopt_orphan_bailian_keys(db_session, integration_id=iid)
    await db_session.commit()
    assert first["adopted"] == 1
    first_assignment_id = key.assignment_id

    # Second run: nothing new to adopt, no unique-key collision.
    second = await adopt_orphan_bailian_keys(db_session, integration_id=iid)
    await db_session.commit()
    assert second["adopted"] == 0

    count = (await db_session.execute(
        select(ApiKeyAssignment).where(
            ApiKeyAssignment.request_number == "AK-ADOPT-998003",
        )
    )).scalars().all()
    assert len(count) == 1

    # A lost back-link (assignment exists, inventory row unlinked) is
    # re-attached rather than colliding.
    key.assignment_id = None
    await db_session.commit()
    third = await adopt_orphan_bailian_keys(db_session, integration_id=iid)
    await db_session.commit()
    assert third["adopted"] == 0
    assert third["relinked"] == 1
    await db_session.refresh(key)
    assert key.assignment_id == first_assignment_id