"""2026-07-28 — ``Idempotency-Key`` replay protection on grant approve/reject.

The mobile write primitive (``useMobileMutation``) has always minted an
``Idempotency-Key`` per logical operation and documented the guarantee that
"a network blip after the server already processed the write does not
double-execute". For cloud-grant approve that guarantee was **false**: the
endpoint did not read the header, so a retried approve fired a second
provider dispatch (a second IAM Identity Center / SCIM write). The
client-side re-entry guard only covers a double TAP inside one page
instance — not a retry, a second tab, or a second device.

Fixed by ``@idempotent`` (``app/core/idempotent_route.py``). This test
exercises the real HTTP endpoint end-to-end, because the interesting part is
the interaction between the reservation row, the service's status check, and
the response envelope.

Note on what "double-execute" would look like without the guard: the second
call re-reads the grant, finds it ``active`` rather than ``pending``, and
raises ``ConflictError`` → the CLIENT sees a 409 for an operation that
actually succeeded. That is the benign half. The malignant half is a replay
that arrives while the first dispatch is still in flight, which the
reservation turns into a clean "still in progress" 409 instead of two
concurrent provider writes.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime

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

from app.core.idempotency import IdempotencyRecord
from app.domains.cmdb.models import (
    GRANT_STATUS_PENDING,
    SCOPE_STATUS_ACTIVE,
    CmdbCloudAccessGrant,
    CmdbCloudScope,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration


async def _integration(db: AsyncSession) -> Integration:
    row = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="aws",
        name=f"aws-idem-{uuid.uuid4().hex[:6]}",
        slug=f"aws-idem-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers"],
    )
    db.add(row)
    await db.flush()
    return row


async def _scope(
    db: AsyncSession, integration: Integration, *, owner_person_id: str,
) -> CmdbCloudScope:
    s = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integration.id),
        scope_type="aws_account",
        external_id=uuid.uuid4().hex[:12],
        display_name=f"scope-{uuid.uuid4().hex[:6]}",
        owner_person_id=owner_person_id,
        status=SCOPE_STATUS_ACTIVE,
        first_seen_at=datetime.now(UTC),
        last_synced_at=datetime.now(UTC),
        environment="dev",
    )
    db.add(s)
    await db.flush()
    return s


async def _requester(db: AsyncSession) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"idem-req-{uuid.uuid4().hex[:6]}@test.com",
        display_name="Idem Requester",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-idem-{uuid.uuid4().hex[:6]}",
    )
    db.add(p)
    await db.flush()
    return p


async def _pending_grant(
    db: AsyncSession, scope: CmdbCloudScope, *, person_id: str,
) -> CmdbCloudAccessGrant:
    g = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=person_id,
        role_key=f"arn:aws:sso:::permissionSet/idem/{uuid.uuid4().hex[:8]}",
        role_display_name="ReadOnlyAccess",
        status=GRANT_STATUS_PENDING,
        granted_at=datetime.now(UTC),
        detail={"reason": "idempotency test", "requested_ttl_days": 7},
    )
    db.add(g)
    await db.flush()
    return g


@pytest.mark.asyncio
async def test_reject_replay_returns_cached_body_not_a_conflict(
    client, db_session: AsyncSession, admin_user,
):
    """Same key twice → one state change, identical body, no 409.

    Reject (rather than approve) is the cleanest probe: it is a pure state
    transition with no provider driver to stub, so any difference between
    the two responses is attributable to the idempotency layer alone.
    """
    admin, token = admin_user
    integ = await _integration(db_session)
    scope = await _scope(db_session, integ, owner_person_id=str(admin.id))
    requester = await _requester(db_session)
    grant = await _pending_grant(db_session, scope, person_id=str(requester.id))
    await db_session.commit()

    key = f"cloud_grant.reject:{uuid.uuid4()}"
    headers = {"Authorization": f"Bearer {token}", "Idempotency-Key": key}

    first = await client.post(
        f"/api/v1/cmdb/grants/{grant.id}/reject",
        json={"reason": "not needed"},
        headers=headers,
    )
    assert first.status_code == 200, first.text

    second = await client.post(
        f"/api/v1/cmdb/grants/{grant.id}/reject",
        json={"reason": "not needed"},
        headers=headers,
    )
    # The replay must NOT be a 409 "grant is 'revoked', not pending" — that
    # is what the client saw before the guard, for an operation that had in
    # fact succeeded.
    assert second.status_code == 200, second.text
    assert second.json()["data"] == first.json()["data"]

    row = await db_session.scalar(
        select(IdempotencyRecord).where(IdempotencyRecord.key == key)
    )
    assert row is not None
    # The scope carries the CALLER as well as the resource. `lookup()`
    # returns the cached response before the handler body runs, so any
    # authorization written in that body is skipped on a replay — without
    # the caller in the scope, a second person holding the same key would
    # be served the first person's serialised result with no check and no
    # audit row. (Added 2026-07-28 after the security scan.)
    assert row.scope == f"cloud_grant.reject:{grant.id}:{admin.id}"
    assert row.state == "completed"


@pytest.mark.asyncio
async def test_no_key_means_no_reservation(
    client, db_session: AsyncSession, admin_user,
):
    """A client that does not send the header behaves exactly as before."""
    admin, token = admin_user
    integ = await _integration(db_session)
    scope = await _scope(db_session, integ, owner_person_id=str(admin.id))
    requester = await _requester(db_session)
    grant = await _pending_grant(db_session, scope, person_id=str(requester.id))
    await db_session.commit()

    before = len(
        (await db_session.execute(select(IdempotencyRecord))).scalars().all()
    )
    resp = await client.post(
        f"/api/v1/cmdb/grants/{grant.id}/reject",
        json={"reason": "no key"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text
    after = len(
        (await db_session.execute(select(IdempotencyRecord))).scalars().all()
    )
    assert after == before


@pytest.mark.asyncio
async def test_same_key_on_a_different_grant_is_a_separate_operation(
    client, db_session: AsyncSession, admin_user,
):
    """``key_params=("grant_id",)`` is what prevents cross-grant dedupe.

    Without the per-resource scope, a client reusing one key would get
    grant A's cached response when rejecting grant B — i.e. B would look
    decided while staying pending. Worst possible failure mode for an
    approval surface, so it gets its own test.
    """
    admin, token = admin_user
    integ = await _integration(db_session)
    scope = await _scope(db_session, integ, owner_person_id=str(admin.id))
    requester = await _requester(db_session)
    grant_a = await _pending_grant(db_session, scope, person_id=str(requester.id))
    grant_b = await _pending_grant(db_session, scope, person_id=str(requester.id))
    await db_session.commit()

    key = f"shared:{uuid.uuid4()}"
    headers = {"Authorization": f"Bearer {token}", "Idempotency-Key": key}

    ra = await client.post(
        f"/api/v1/cmdb/grants/{grant_a.id}/reject",
        json={"reason": "a"}, headers=headers,
    )
    rb = await client.post(
        f"/api/v1/cmdb/grants/{grant_b.id}/reject",
        json={"reason": "b"}, headers=headers,
    )
    assert ra.status_code == 200, ra.text
    assert rb.status_code == 200, rb.text
    assert ra.json()["data"]["id"] == str(grant_a.id)
    assert rb.json()["data"]["id"] == str(grant_b.id)

    scopes = {
        r.scope
        for r in (
            await db_session.execute(
                select(IdempotencyRecord).where(IdempotencyRecord.key == key)
            )
        ).scalars().all()
    }
    assert scopes == {
        f"cloud_grant.reject:{grant_a.id}:{admin.id}",
        f"cloud_grant.reject:{grant_b.id}:{admin.id}",
    }