"""Tests for the API-Key Reports tab analytics
(``app.domains.apikey.analytics.compute_apikey_analytics``, surfaced at
``GET /api/v1/apikeys/analytics``).

Production-grade per CLAUDE.md: real ``cmdb_cost_facts`` rows, real
``ApiKeyAssignment`` / ``IdentityPerson`` / ``CmdbCloudScope``, no DB
mocks. The spend axis routes through the SAME ``compute_assignment_spend``
dispatch the per-key panel + budget cron use, so the report's totals
reconcile to the cent with those surfaces by construction.

Contract under test:

* Distribution counts the FULL population (by provider / status /
  cost-attribution binding).
* The committed monthly-budget pool sums only active/approved keys.
* Actual MTD spend is computed only for the resolvable subset; Bailian
  (Aliyun BSS account-level only) and unbound keys land in
  ``spend.unresolvable_*`` rather than under-counting the total.
* ``budget_utilization_percent`` = total_mtd / total_budget * 100.
"""
from __future__ import annotations

import uuid
from datetime import date, datetime, timedelta
from decimal import Decimal
from time import perf_counter
from zoneinfo import ZoneInfo

import pytest
from httpx import AsyncClient
from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.apikey.analytics import compute_apikey_analytics
from app.domains.apikey.analytics_spend import compute_assignments_spend_bulk
from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider
from app.domains.apikey.spend import compute_assignment_spend
from app.domains.cmdb.models import (
    CmdbBedrockProfileFact,
    CmdbCloudScope,
    CmdbCostFact,
    CmdbIamUser,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration


async def _provider(db: AsyncSession, name: str) -> ApiKeyProvider:
    p = ApiKeyProvider(id=uuid.uuid4(), name=name, is_active=True, sort_order=0)
    db.add(p)
    await db.commit()
    return p


async def _person(db: AsyncSession, email: str) -> IdentityPerson:
    person = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name=email.split("@")[0],
        work_email=email,
        employment_status="active",
    )
    db.add(person)
    await db.commit()
    return person


async def _integration(db: AsyncSession) -> Integration:
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="azure",
        name="Azure Test",
        slug=f"azure-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=[],
        config_public={},
    )
    db.add(integ)
    await db.commit()
    return integ


async def _rg_scope(
    db: AsyncSession, *, integration: Integration, rg_name: str,
) -> CmdbCloudScope:
    scope = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integration.id),
        scope_type="azure_resource_group",
        external_id=rg_name,
        display_name=f"Foundry · {rg_name}",
        status="active",
    )
    db.add(scope)
    await db.commit()
    return scope


async def _assignment(
    db: AsyncSession,
    *,
    provider: ApiKeyProvider,
    applicant: IdentityPerson,
    budget: Decimal,
    status: str = "active",
    cloud_scope: CmdbCloudScope | None = None,
) -> ApiKeyAssignment:
    a = ApiKeyAssignment(
        id=uuid.uuid4(),
        provider_id=provider.id,
        assigned_to=applicant.id,
        status=status,
        request_number=f"AK-{uuid.uuid4().hex[:8].upper()}",
        project_name="LLM eval",
        reason="test",
        monthly_budget_usd=budget,
        cloud_scope_id=str(cloud_scope.id) if cloud_scope else None,
    )
    db.add(a)
    await db.commit()
    return a


async def _cost_fact(
    db: AsyncSession,
    *,
    integration: Integration,
    fact_date: date,
    amount_usd: Decimal,
    service: str,
    resource_group: str,
    resource_id: str | None = None,
) -> None:
    db.add(CmdbCostFact(
        fact_key=uuid.uuid4().hex,
        date=fact_date,
        integration_id=str(integration.id),
        service=service,
        amount_usd=amount_usd,
        amount_native=amount_usd,
        currency="USD",
        resource_group=resource_group,
        resource_id=resource_id,
        auto_owner_source="scope_owner",
    ))
    await db.commit()


@pytest.mark.asyncio
async def test_analytics_distribution_budget_and_resolvable_spend(
    db_session: AsyncSession,
):
    """One Foundry key (resolvable, $12 spend) + one Bailian key
    (unresolvable). The report counts both for distribution + budget,
    but actual spend covers only the Foundry key and the Bailian key
    lands in ``unresolvable``."""
    foundry = await _provider(db_session, "Azure OpenAI")
    bailian = await _provider(db_session, "阿里云百炼 Model Studio")
    integ = await _integration(db_session)
    alice = await _person(db_session, "alice@pingcap.com")
    bob = await _person(db_session, "bob@pingcap.com")
    rg = await _rg_scope(db_session, integration=integ, rg_name="NexusCRM")

    a_foundry = await _assignment(
        db_session, provider=foundry, applicant=alice,
        budget=Decimal("400.00"), cloud_scope=rg,
    )
    await _assignment(
        db_session, provider=bailian, applicant=bob,
        budget=Decimal("100.00"),
    )

    today = datetime.now(ZoneInfo("Asia/Shanghai")).date()
    await _cost_fact(
        db_session, integration=integ, fact_date=today,
        amount_usd=Decimal("12.00"), service="Foundry Models",
        resource_group="NexusCRM",
    )
    # A fact in a different RG must NOT count toward the Foundry key.
    await _cost_fact(
        db_session, integration=integ, fact_date=today,
        amount_usd=Decimal("99.99"), service="Foundry Models",
        resource_group="OtherRG",
    )

    result = await compute_apikey_analytics(db_session, month=None)

    # ── Totals ──
    assert result["totals"]["total_keys"] == 2
    assert result["totals"]["active_keys"] == 2
    assert result["totals"]["total_monthly_budget_usd"] == 500.0
    assert result["totals"]["total_mtd_usd"] == 12.0
    # 12 / 500 * 100 = 2.4
    assert result["totals"]["budget_utilization_percent"] == 2.4

    # ── Distribution ──
    provider_names = {
        r["provider_name"] for r in result["distribution"]["by_provider"]
    }
    assert provider_names == {"Azure OpenAI", "阿里云百炼 Model Studio"}
    attribution = {
        r["mode"]: r["count"]
        for r in result["distribution"]["by_cost_attribution"]
    }
    assert attribution == {"iam": 0, "scope": 1, "unattributed": 1}

    # ── Spend ──
    spend = result["spend"]
    assert spend["resolvable_keys"] == 1
    assert spend["unresolvable_keys"] == 1
    assert spend["data_complete"] is False
    assert spend["total_mtd_usd"] == 12.0
    bucket_map = {r["bucket"]: r["mtd_usd"] for r in spend["by_bucket"]}
    assert bucket_map.get("foundry") == 12.0
    unresolvable_buckets = {
        r["bucket"]: r["count"] for r in spend["unresolvable_by_bucket"]
    }
    assert unresolvable_buckets.get("bailian") == 1
    # Top spender is the Foundry key.
    assert spend["top_spenders"][0]["request_number"] == a_foundry.request_number
    assert spend["top_spenders"][0]["mtd_usd"] == 12.0
    # Budget health: the Foundry key is well under 80% → ok; the
    # Bailian key is unresolvable → unknown.
    assert spend["budget_health"]["ok"] == 1
    assert spend["budget_health"]["unknown"] == 1


@pytest.mark.asyncio
async def test_analytics_excludes_pending_and_rejected_from_budget(
    db_session: AsyncSession,
):
    """pending_approval / rejected keys count in the population but carry
    stale budgets that must NOT inflate the active budget pool, and are
    never spend-computed."""
    foundry = await _provider(db_session, "Azure OpenAI")
    carol = await _person(db_session, "carol@pingcap.com")

    await _assignment(
        db_session, provider=foundry, applicant=carol,
        budget=Decimal("50.00"), status="active",
    )
    await _assignment(
        db_session, provider=foundry, applicant=carol,
        budget=Decimal("9999.00"), status="pending_approval",
    )
    await _assignment(
        db_session, provider=foundry, applicant=carol,
        budget=Decimal("1234.00"), status="rejected",
    )

    result = await compute_apikey_analytics(db_session, month=None)

    assert result["totals"]["total_keys"] == 3
    # active_keys counts active + approved only.
    assert result["totals"]["active_keys"] == 1
    # Only the active key's $50 budget is in the pool.
    assert result["totals"]["total_monthly_budget_usd"] == 50.0
    status_map = {
        r["status"]: r["count"] for r in result["distribution"]["by_status"]
    }
    assert status_map == {"active": 1, "pending_approval": 1, "rejected": 1}


class _SelectCounter:
    def __init__(self, session: AsyncSession) -> None:
        self._engine = session.bind.sync_engine  # type: ignore[union-attr]
        self.count = 0

    def __enter__(self):
        event.listen(self._engine, "before_cursor_execute", self._on_query)
        return self

    def __exit__(self, *_exc) -> None:
        event.remove(self._engine, "before_cursor_execute", self._on_query)

    def _on_query(self, _conn, _cursor, statement, _params, _ctx, _many) -> None:
        if statement.lstrip().upper().startswith("SELECT"):
            self.count += 1


@pytest.mark.asyncio
async def test_analytics_query_count_is_constant_for_10_and_500_keys(
    db_session: AsyncSession,
):
    """Fleet size must affect rows processed, not SQL round trips."""
    foundry = await _provider(db_session, "Azure OpenAI")
    integ = await _integration(db_session)
    holder = await _person(db_session, "fleet@pingcap.com")
    scope = await _rg_scope(db_session, integration=integ, rg_name="FleetRG")

    def assignments(start: int, end: int) -> list[ApiKeyAssignment]:
        return [
            ApiKeyAssignment(
                id=uuid.uuid4(), provider_id=foundry.id,
                assigned_to=holder.id, status="active",
                request_number=f"AK-FLEET-{i:04d}", project_name="fleet",
                reason="query-count regression", monthly_budget_usd=Decimal("10"),
                cloud_scope_id=str(scope.id),
            )
            for i in range(start, end)
        ]

    db_session.add_all(assignments(0, 10))
    await db_session.commit()
    with _SelectCounter(db_session) as ten:
        await compute_apikey_analytics(db_session)

    db_session.add_all(assignments(10, 500))
    await db_session.commit()
    started = perf_counter()
    with _SelectCounter(db_session) as five_hundred:
        result = await compute_apikey_analytics(db_session)
    elapsed = perf_counter() - started

    assert result["totals"]["total_keys"] == 500
    assert five_hundred.count <= ten.count + 1
    assert five_hundred.count <= 12
    # A generous CI budget catches accidental reintroduction of 500
    # serial awaits without turning ordinary runner variance into flakes.
    assert elapsed < 5.0


@pytest.mark.asyncio
async def test_bulk_spend_empty_input_issues_no_queries(db_session: AsyncSession):
    with _SelectCounter(db_session) as queries:
        result = await compute_assignments_spend_bulk(
            db_session, assignments=[], provider_names={},
            month_start=date(2026, 8, 1), month_end=date(2026, 8, 2),
        )
    assert result == {}
    assert queries.count == 0


@pytest.mark.asyncio
async def test_bulk_spend_matches_per_key_for_revoked_and_no_data(
    db_session: AsyncSession,
):
    foundry = await _provider(db_session, "Azure OpenAI")
    bailian = await _provider(db_session, "阿里云百炼 Model Studio")
    integ = await _integration(db_session)
    holder = await _person(db_session, "equivalence@pingcap.com")
    scope = await _rg_scope(db_session, integration=integ, rg_name="EqRG")
    today = datetime.now(ZoneInfo("Asia/Shanghai")).date()
    month_start = today.replace(day=1)
    revoked = await _assignment(
        db_session, provider=foundry, applicant=holder,
        budget=Decimal("100"), status="revoked", cloud_scope=scope,
    )
    revoked.revoked_at = datetime.now(ZoneInfo("UTC"))
    missing = await _assignment(
        db_session, provider=bailian, applicant=holder,
        budget=Decimal("100"),
    )
    await _cost_fact(
        db_session, integration=integ, fact_date=today,
        amount_usd=Decimal("12.345678"), service="Foundry Models",
        resource_group="EqRG",
    )
    await db_session.commit()

    rows = [revoked, missing]
    bulk = await compute_assignments_spend_bulk(
        db_session, assignments=rows,
        provider_names={foundry.id: foundry.name, bailian.id: bailian.name},
        month_start=month_start, month_end=today,
    )
    for assignment in rows:
        old = await compute_assignment_spend(
            db_session, assignment=assignment,
            month_start=month_start, month_end=today,
        )
        assert bulk[assignment.id]["bucket"] == old["bucket"]
        assert bulk[assignment.id]["data_available"] == old["data_available"]
        assert bulk[assignment.id]["mtd_usd"] == old["mtd_usd"]
        assert bulk[assignment.id]["last_fact_date"] == old["last_fact_date"]


async def _bedrock_fixture(
    db: AsyncSession, *, email: str,
) -> tuple[ApiKeyProvider, Integration, IdentityPerson]:
    provider = await _provider(db, "AWS BedRock")
    integration = Integration(
        id=uuid.uuid4(), kind="cloud", provider="aws", name="AWS Test",
        slug=f"aws-{uuid.uuid4().hex[:6]}", enabled=True, is_primary=False,
        capabilities=[], config_public={},
    )
    db.add(integration)
    await db.commit()
    return provider, integration, await _person(db, email)


async def _bedrock_key(
    db: AsyncSession, *, provider: ApiKeyProvider, integration: Integration,
    holder: IdentityPerson, request_number: str, created_at: datetime,
) -> ApiKeyAssignment:
    iam = CmdbIamUser(
        id=uuid.uuid4(), integration_id=str(integration.id),
        user_name=request_number.lower(), sub_account_id="",
        arn=f"arn:aws:iam::123456789012:user/{request_number.lower()}",
        tags={"Owner": holder.work_email}, owner_person_id=str(holder.id),
    )
    db.add(iam)
    await db.flush()
    assignment = ApiKeyAssignment(
        id=uuid.uuid4(), provider_id=provider.id, assigned_to=holder.id,
        status="active", request_number=request_number,
        project_name="bedrock", reason="analytics equivalence",
        monthly_budget_usd=Decimal("100"), iam_user_id=iam.id,
        iam_user_arn=iam.arn, iam_integration_id=str(integration.id),
        created_at=created_at,
    )
    db.add(assignment)
    await db.commit()
    return assignment


async def _assert_bulk_matches_old(
    db: AsyncSession, *, assignments: list[ApiKeyAssignment],
    provider: ApiKeyProvider, month_start: date, month_end: date,
) -> dict[uuid.UUID, dict]:
    bulk = await compute_assignments_spend_bulk(
        db, assignments=assignments, provider_names={provider.id: provider.name},
        month_start=month_start, month_end=month_end,
    )
    for assignment in assignments:
        old = await compute_assignment_spend(
            db, assignment=assignment,
            month_start=month_start, month_end=month_end,
        )
        for field in ("bucket", "data_available", "mtd_usd", "last_fact_date"):
            assert bulk[assignment.id][field] == old[field]
    return bulk


@pytest.mark.asyncio
async def test_bulk_bedrock_matches_shared_owner_and_per_key_deduction(
    db_session: AsyncSession,
):
    provider, integration, holder = await _bedrock_fixture(
        db_session, email="shared-owner@pingcap.com",
    )
    first = await _bedrock_key(
        db_session, provider=provider, integration=integration, holder=holder,
        request_number="AK-SHARED-1", created_at=datetime(2026, 1, 1),
    )
    second = await _bedrock_key(
        db_session, provider=provider, integration=integration, holder=holder,
        request_number="AK-SHARED-2", created_at=datetime(2026, 1, 2),
    )
    today = datetime.now(ZoneInfo("Asia/Shanghai")).date()
    month_start = today.replace(day=1)
    await _cost_fact(
        db_session, integration=integration, fact_date=today,
        amount_usd=Decimal("100"), service="Amazon Bedrock",
        resource_group="", resource_id=(
            "aws-principal:Owner=shared-owner@pingcap.com|service=Amazon Bedrock"
        ),
    )

    shared = await _assert_bulk_matches_old(
        db_session, assignments=[first, second], provider=provider,
        month_start=month_start, month_end=today,
    )
    assert shared[first.id]["mtd_usd"] == Decimal("100")
    assert shared[first.id]["data_available"] is True
    assert shared[second.id]["data_available"] is False

    db_session.add(CmdbBedrockProfileFact(
        fact_key=uuid.uuid4().hex, date=today,
        integration_id=str(integration.id), scope_external_id="123456789012",
        owner_email=holder.work_email,
        profile_tag_key="iamPrincipal/helix-request-number",
        profile=second.request_number, service="Amazon Bedrock",
        amount_usd=Decimal("25"), amount_native=Decimal("25"), currency="USD",
    ))
    await db_session.commit()
    isolated = await _assert_bulk_matches_old(
        db_session, assignments=[first, second], provider=provider,
        month_start=month_start, month_end=today,
    )
    assert isolated[first.id]["mtd_usd"] == Decimal("75")
    assert isolated[second.id]["mtd_usd"] == Decimal("25")
    assert isolated[second.id]["data_available"] is True


@pytest.mark.asyncio
async def test_bulk_bedrock_matches_transfer_epoch(db_session: AsyncSession):
    from app.domains.offboarding.models import XFER_COMPLETED, AssetTransferRequest

    provider, integration, current = await _bedrock_fixture(
        db_session, email="current@pingcap.com",
    )
    previous = await _person(db_session, "previous@pingcap.com")
    assignment = await _bedrock_key(
        db_session, provider=provider, integration=integration, holder=current,
        request_number="AK-TRANSFER", created_at=datetime(2026, 1, 1),
    )
    today = datetime.now(ZoneInfo("Asia/Shanghai")).date()
    month_start = today.replace(day=1)
    db_session.add(AssetTransferRequest(
        id=uuid.uuid4(), initiator_id=previous.id, recipient_id=current.id,
        asset_type="api_key", asset_ref_id=assignment.id,
        asset_label=assignment.request_number, status=XFER_COMPLETED,
        responded_at=datetime.now(ZoneInfo("UTC")),
        completed_at=datetime.now(ZoneInfo("UTC")),
    ))
    await db_session.commit()
    await _cost_fact(
        db_session, integration=integration, fact_date=month_start,
        amount_usd=Decimal("20"), service="Amazon Bedrock", resource_group="",
        resource_id="aws-principal:Owner=previous@pingcap.com|service=Amazon Bedrock",
    )
    await _cost_fact(
        db_session, integration=integration, fact_date=today,
        amount_usd=Decimal("10"), service="Amazon Bedrock", resource_group="",
        resource_id="aws-principal:Owner=current@pingcap.com|service=Amazon Bedrock",
    )
    bulk = await _assert_bulk_matches_old(
        db_session, assignments=[assignment], provider=provider,
        month_start=month_start, month_end=today,
    )
    assert bulk[assignment.id]["mtd_usd"] == Decimal("30")


@pytest.mark.asyncio
async def test_bulk_bedrock_normalizes_owner_and_bounds_fact_rows(
    db_session: AsyncSession,
):
    """Case variants are one owner; unrelated account owners stay out."""
    provider, integration, holder = await _bedrock_fixture(
        db_session, email="case-owner@pingcap.com",
    )
    first = await _bedrock_key(
        db_session, provider=provider, integration=integration, holder=holder,
        request_number="AK-CASE-1", created_at=datetime(2026, 1, 1),
    )
    second = await _bedrock_key(
        db_session, provider=provider, integration=integration, holder=holder,
        request_number="AK-CASE-2", created_at=datetime(2026, 1, 2),
    )
    second_iam = await db_session.get(CmdbIamUser, second.iam_user_id)
    assert second_iam is not None
    second_iam.tags = {"Owner": holder.work_email.upper()}

    today = datetime.now(ZoneInfo("Asia/Shanghai")).date()
    db_session.add(CmdbCostFact(
        fact_key=uuid.uuid4().hex, date=today,
        integration_id=str(integration.id), service="Amazon Bedrock",
        amount_usd=Decimal("100"), amount_native=Decimal("100"),
        currency="USD", resource_group="",
        resource_id=(
            "aws-principal:Owner=case-owner@pingcap.com|service=Amazon Bedrock"
        ),
        auto_owner_source="scope_owner",
    ))
    for idx in range(40):
        db_session.add(CmdbCostFact(
            fact_key=uuid.uuid4().hex, date=today,
            integration_id=str(integration.id), service="Amazon Bedrock",
            amount_usd=Decimal("999"), amount_native=Decimal("999"),
            currency="EUR", resource_group="",
            resource_id=(
                f"aws-principal:Owner=unrelated-{idx}@example.com|"
                "service=Amazon Bedrock"
            ),
            auto_owner_source="scope_owner",
        ))
    await db_session.commit()

    owner_query_rows: list[int] = []
    engine = db_session.bind.sync_engine  # type: ignore[union-attr]

    def capture_rows(_conn, cursor, statement, _params, _ctx, _many) -> None:
        sql = statement.lower()
        if "cmdb_cost_facts" in sql and "substring_index" in sql:
            owner_query_rows.append(cursor.rowcount)

    event.listen(engine, "after_cursor_execute", capture_rows)
    try:
        bulk = await compute_assignments_spend_bulk(
            db_session,
            assignments=[first, second],
            provider_names={provider.id: provider.name},
            month_start=today.replace(day=1),
            month_end=today,
        )
    finally:
        event.remove(engine, "after_cursor_execute", capture_rows)

    available = [row for row in bulk.values() if row["data_available"]]
    assert len(available) == 1
    assert sum((row["mtd_usd"] for row in available), Decimal(0)) == Decimal("100")
    assert owner_query_rows == [1]


@pytest.mark.asyncio
async def test_bulk_bedrock_profile_query_uses_exact_integration_pairs(
    db_session: AsyncSession,
):
    """A profile ref from another account is not fetched or attributed."""
    provider, first_integration, holder = await _bedrock_fixture(
        db_session, email="pair-scope@pingcap.com",
    )
    second_integration = Integration(
        id=uuid.uuid4(), kind="cloud", provider="aws", name="AWS Pair 2",
        slug=f"aws-{uuid.uuid4().hex[:6]}", enabled=True, is_primary=False,
        capabilities=[], config_public={},
    )
    db_session.add(second_integration)
    await db_session.commit()
    first = await _bedrock_key(
        db_session, provider=provider, integration=first_integration,
        holder=holder, request_number="AK-PAIR-A",
        created_at=datetime(2026, 1, 1),
    )
    second = await _bedrock_key(
        db_session, provider=provider, integration=second_integration,
        holder=holder, request_number="AK-PAIR-B",
        created_at=datetime(2026, 1, 2),
    )
    today = datetime.now(ZoneInfo("Asia/Shanghai")).date()
    for integration, cross_ref in (
        (first_integration, second.request_number),
        (second_integration, first.request_number),
    ):
        db_session.add(CmdbBedrockProfileFact(
            fact_key=uuid.uuid4().hex, date=today,
            integration_id=str(integration.id), scope_external_id="123456789012",
            owner_email=holder.work_email,
            profile_tag_key="iamPrincipal/helix-request-number",
            profile=cross_ref, service="Amazon Bedrock",
            amount_usd=Decimal("999"), amount_native=Decimal("999"),
            currency="USD",
        ))
    await db_session.commit()

    profile_query_rows: list[int] = []
    engine = db_session.bind.sync_engine  # type: ignore[union-attr]

    def capture_rows(_conn, cursor, statement, _params, _ctx, _many) -> None:
        if "cmdb_bedrock_profile_facts" in statement.lower():
            profile_query_rows.append(cursor.rowcount)

    event.listen(engine, "after_cursor_execute", capture_rows)
    try:
        result = await compute_assignments_spend_bulk(
            db_session,
            assignments=[first, second],
            provider_names={provider.id: provider.name},
            month_start=today.replace(day=1),
            month_end=today,
        )
    finally:
        event.remove(engine, "after_cursor_execute", capture_rows)

    assert profile_query_rows == [0]
    assert all(row["mtd_usd"] == 0 for row in result.values())


@pytest.mark.asyncio
async def test_analytics_http_is_read_only_and_projects_expiry(
    client: AsyncClient,
    db_session: AsyncSession,
    admin_headers: dict,
    employee_headers: dict,
    employee_user,
):
    person, _token = employee_user
    provider = await _provider(db_session, "Read-only Analytics")
    assignment = await _assignment(
        db_session, provider=provider, applicant=person,
        budget=Decimal("500"),
    )
    assignment.valid_until = (
        datetime.now(ZoneInfo("Asia/Shanghai")).date()
        - timedelta(days=1)
    )
    await db_session.commit()

    denied = await client.get(
        "/api/v1/apikeys/analytics",
        headers=employee_headers,
    )
    assert denied.status_code == 403

    statements: list[str] = []
    engine = db_session.bind.sync_engine  # type: ignore[union-attr]

    def capture(_conn, _cursor, statement, _params, _ctx, _many) -> None:
        statements.append(statement.lower())

    event.listen(engine, "before_cursor_execute", capture)
    try:
        response = await client.get(
            "/api/v1/apikeys/analytics",
            headers=admin_headers,
        )
    finally:
        event.remove(engine, "before_cursor_execute", capture)

    assert response.status_code == 200, response.text
    payload = response.json()["data"]
    statuses = {
        row["status"]: row["count"]
        for row in payload["distribution"]["by_status"]
    }
    assert statuses["expired"] == 1
    assert payload["totals"]["active_keys"] == 0
    assert payload["totals"]["total_monthly_budget_usd"] == 0
    relevant = "\n".join(statements)
    assert "for update" not in relevant
    assert "update " not in relevant
    assert "insert " not in relevant
    assert "delete " not in relevant
    await db_session.refresh(assignment)
    assert assignment.status == "active"