"""Tests for the self-service API-Key release flow.

Rewritten 2026-05-20: the release flow no longer routes through a
Feishu approval template. It is now a Helix-internal approval — submit
inserts an :class:`ApiKeyReleaseRequestModel` row in ``pending``, parks
the assignment in ``pending_release``, and pushes a Feishu / email
notification (NOT a Feishu approval card) to the requester's direct
manager. The manager (or any IT admin) clicks Approve / Reject inside
Helix — the service writes the decision and (on approve) calls
``revoke_key`` with ``actor_role=self_with_approval``.

Properties under test:

1. ``submit_release_request`` parks the assignment in ``pending_release``
   without invoking cloud teardown, inserts an
   ``api_key_release_requests`` row in ``pending``, and stashes the
   prior status in both the table column ``previous_status`` AND the
   legacy ``assignment.detail.release.previous_status`` (for the
   reject / cancel rollback).

2. ``approve_release_request``:
   * Only the requester's direct manager OR an IT Admin may call;
     other employees get 403.
   * On approve, invokes ``revoke_key`` with
     ``actor_role=self_with_approval`` which writes all five
     revocation-metadata columns and stamps the row in
     ``status='revoked'``.
   * Idempotent: a second approve on an already-approved request
     returns the existing state without re-firing the cloud
     teardown.
   * If the requester has departed between submit and approve, the
     happy path is short-circuited to a "refused_leaver" auto-reject
     so the offboarding pipeline owns the cleanup.

3. ``reject_release_request`` flips the request to ``rejected``,
   reverts the assignment to its pre-release status, and notifies
   the requester. Same authz as approve.

4. ``cancel_release_request`` (requester-side withdrawal): marks the
   request ``cancelled`` and reverts the assignment.

5. The cross-state guards still hold: pending_release blocks reveal /
   extend, a pending AssetTransferRequest blocks a new release submit.

6. ``revoke_key`` still works for the admin path (default
   ``actor_role='admin'``) and the offboarding system path
   (``actor_role='system_offboarding'``) — these are unchanged by the
   release-flow rewrite but pinned here as a regression guard.

Cloud teardown is monkeypatched (same posture as
``test_revoke_disables_allocation_rule.py``). NotificationService
dispatches are no-op'd via monkeypatch so we don't need a full
notifications-outbox seed for these tests.
"""
from __future__ import annotations

import uuid
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.models import (
    ApiKeyReleaseRequest as ApiKeyReleaseRequestModel,
)
from app.domains.apikey.schemas import ApiKeyReleaseRequest
from app.domains.apikey.service import (
    ACTOR_ROLE_ADMIN,
    ACTOR_ROLE_SELF_WITH_APPROVAL,
    STATUS_ACTIVE,
    STATUS_PENDING_RELEASE,
    STATUS_REVOKED,
    ApiKeyService,
)
from app.identity.person_models import IdentityPerson

# ── Auth fixtures ──────────────────────────────────────────────────


def _user_auth(person: IdentityPerson) -> SimpleNamespace:
    return SimpleNamespace(
        id=person.id,
        role_names=["employee"],
        work_email=person.work_email,
        display_name=person.display_name,
    )


def _admin_auth(person: IdentityPerson) -> SimpleNamespace:
    # ``global_admin`` is the role name the GLOBAL_ADMIN_ROLE constant
    # resolves to in ``core.rbac`` — has_role match is case-sensitive
    # exact, so the test seed must use the canonical literal.
    return SimpleNamespace(
        id=person.id,
        role_names=["global_admin"],
        work_email=person.work_email,
        display_name=person.display_name,
    )


# ── DB helpers ─────────────────────────────────────────────────────


async def _make_person(
    db: AsyncSession,
    *,
    email: str,
    display_name: str | None = None,
    manager: IdentityPerson | None = None,
) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name=display_name or email.split("@")[0],
        work_email=email,
        employment_status="active",
        manager_person_id=manager.id if manager else None,
    )
    db.add(p)
    await db.commit()
    return p


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


async def _make_active_assignment(
    db: AsyncSession,
    *,
    provider: ApiKeyProvider,
    applicant: IdentityPerson,
) -> ApiKeyAssignment:
    a = ApiKeyAssignment(
        id=uuid.uuid4(),
        provider_id=provider.id,
        assigned_to=applicant.id,
        status=STATUS_ACTIVE,
        request_number=f"AK-{uuid.uuid4().hex[:8].upper()}",
        project_name="release-test",
        reason="release flow tests",
        monthly_budget_usd=Decimal("100.00"),
        key_value=None,
        detail={},
    )
    db.add(a)
    await db.commit()
    return a


def _bypass_notify(monkeypatch) -> None:
    """Drop notification dispatch — the outbox path exercises the full
    NotificationOutbox table and would require seeding more
    integrations than these tests need. Each notify helper is async-
    safe to no-op.
    """
    async def _noop(*args, **kwargs):  # noqa: ANN001,D401
        return None

    monkeypatch.setattr(ApiKeyService, "_notify_admin_revoke", _noop)
    monkeypatch.setattr(ApiKeyService, "_notify_release_pending", _noop)
    monkeypatch.setattr(ApiKeyService, "_notify_release_decided", _noop)


# ── Submit ─────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_submit_release_request_parks_in_pending_release(
    db_session: AsyncSession, monkeypatch,
):
    """Submit creates a release_request row, flips the assignment to
    pending_release, and does NOT call the cloud."""
    _bypass_notify(monkeypatch)
    manager = await _make_person(db_session, email="mgr-1@pingcap.com")
    employee = await _make_person(
        db_session, email="alice-rel@pingcap.com", manager=manager,
    )
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=employee,
    )

    # Trip-wire: cloud teardown must NOT run on submit.
    async def cloud_teardown_must_not_run(self, assignment, *, trace_id):
        raise AssertionError(
            "Cloud revoke must not fire on submit_release_request",
        )

    monkeypatch.setattr(
        ApiKeyService,
        "_revoke_cloud_resources",
        cloud_teardown_must_not_run,
    )

    service = ApiKeyService(db_session)
    response = await service.submit_release_request(
        assignment.id,
        ApiKeyReleaseRequest(
            reason_code="no_longer_needed",
            reason_note=None,
        ),
        current_user=_user_auth(employee),
        trace_id="t-rel-submit",
        ip_address="127.0.0.1",
        user_agent="pytest",
    )
    await db_session.commit()
    assert response.status == STATUS_PENDING_RELEASE

    await db_session.refresh(assignment)
    assert assignment.status == STATUS_PENDING_RELEASE
    release_meta = (assignment.detail or {}).get("release") or {}
    # Legacy detail stash retained for backward compatibility with the
    # admin UI that already reads ``detail.release``.
    assert release_meta.get("previous_status") == STATUS_ACTIVE
    assert release_meta.get("reason_code") == "no_longer_needed"

    rows = (
        await db_session.execute(
            select(ApiKeyReleaseRequestModel).where(
                ApiKeyReleaseRequestModel.assignment_id == assignment.id,
            ),
        )
    ).scalars().all()
    assert len(rows) == 1
    req = rows[0]
    assert req.status == "pending"
    assert req.requester_id == employee.id
    assert req.approver_id is None
    assert req.reason_code == "no_longer_needed"
    assert req.previous_status == STATUS_ACTIVE
    assert req.responded_at is None


@pytest.mark.asyncio
async def test_submit_release_request_owner_only(
    db_session: AsyncSession, monkeypatch,
):
    """Only the assignee may submit a release for their own AK;
    another employee submitting the same id gets ForbiddenError."""
    from app.core.exceptions import ForbiddenError

    _bypass_notify(monkeypatch)
    owner = await _make_person(db_session, email="owner@pingcap.com")
    intruder = await _make_person(db_session, email="intruder@pingcap.com")
    provider = await _make_provider(db_session, name="Test Provider 2")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=owner,
    )

    service = ApiKeyService(db_session)
    with pytest.raises(ForbiddenError):
        await service.submit_release_request(
            assignment.id,
            ApiKeyReleaseRequest(reason_code="project_ended"),
            current_user=_user_auth(intruder),
            trace_id="t-rel-intruder",
            ip_address=None,
            user_agent=None,
        )


@pytest.mark.asyncio
async def test_submit_release_request_idempotent_on_pending(
    db_session: AsyncSession, monkeypatch,
):
    """Submitting twice in pending_release returns the existing row
    rather than creating a duplicate request."""
    _bypass_notify(monkeypatch)
    employee = await _make_person(db_session, email="dup@pingcap.com")
    provider = await _make_provider(db_session, name="Dup Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=employee,
    )

    service = ApiKeyService(db_session)
    await service.submit_release_request(
        assignment.id,
        ApiKeyReleaseRequest(reason_code="no_longer_needed"),
        current_user=_user_auth(employee),
        trace_id="t-dup-1",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    # Second submit on the still-pending_release row.
    await service.submit_release_request(
        assignment.id,
        ApiKeyReleaseRequest(reason_code="no_longer_needed"),
        current_user=_user_auth(employee),
        trace_id="t-dup-2",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    rows = (
        await db_session.execute(
            select(ApiKeyReleaseRequestModel).where(
                ApiKeyReleaseRequestModel.assignment_id == assignment.id,
            ),
        )
    ).scalars().all()
    assert len(rows) == 1


# ── Approve ────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_approve_by_manager_revokes_with_self_actor_role(
    db_session: AsyncSession, monkeypatch,
):
    """Manager approves → revoke_key fires with actor_role=
    self_with_approval; all five revocation-metadata columns
    populate; status → revoked."""
    _bypass_notify(monkeypatch)

    manager = await _make_person(db_session, email="bob-mgr@pingcap.com")
    employee = await _make_person(
        db_session, email="bob-rel@pingcap.com",
        display_name="Bob", manager=manager,
    )
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=employee,
    )

    cloud_calls: list[uuid.UUID] = []

    async def fake_revoke_cloud_resources(self, assignment, *, trace_id):
        cloud_calls.append(assignment.id)
        return {"attempted": [{"kind": "stub", "ok": True}], "errors": []}

    monkeypatch.setattr(
        ApiKeyService, "_revoke_cloud_resources",
        fake_revoke_cloud_resources,
    )

    service = ApiKeyService(db_session)
    await service.submit_release_request(
        assignment.id,
        ApiKeyReleaseRequest(
            reason_code="project_ended",
            reason_note="moved off this provider",
        ),
        current_user=_user_auth(employee),
        trace_id="t-rel-bob",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    req = (
        await db_session.execute(
            select(ApiKeyReleaseRequestModel).where(
                ApiKeyReleaseRequestModel.assignment_id == assignment.id,
            ),
        )
    ).scalar_one()

    # Manager approves in Helix Web UI (no Feishu callback).
    await service.approve_release_request(
        req.id,
        current_user=_user_auth(manager),
        decision_note="OK, project really did wrap up",
        trace_id="t-rel-bob-approve",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    # Cloud teardown ran exactly once.
    assert cloud_calls == [assignment.id]

    await db_session.refresh(assignment)
    assert assignment.status == STATUS_REVOKED
    assert assignment.revoke_actor_role == ACTOR_ROLE_SELF_WITH_APPROVAL
    assert assignment.revoked_by == employee.id  # assignee, not the manager
    assert assignment.revoked_at is not None
    assert assignment.revoke_reason_code == "project_ended"
    assert assignment.revoke_reason_note == "moved off this provider"

    await db_session.refresh(req)
    assert req.status == "approved"
    assert req.approver_id == manager.id
    assert req.decision_note == "OK, project really did wrap up"
    assert req.responded_at is not None


@pytest.mark.asyncio
async def test_approve_by_non_manager_non_admin_forbidden(
    db_session: AsyncSession, monkeypatch,
):
    """A peer who is neither the requester's manager NOR an IT admin
    cannot approve — ForbiddenError."""
    from app.core.exceptions import ForbiddenError

    _bypass_notify(monkeypatch)
    manager = await _make_person(db_session, email="real-mgr@pingcap.com")
    employee = await _make_person(
        db_session, email="peer-rel@pingcap.com", manager=manager,
    )
    bystander = await _make_person(db_session, email="bystander@pingcap.com")
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=employee,
    )

    async def fake_cloud(self, assignment, *, trace_id):
        return {"attempted": [], "errors": []}

    monkeypatch.setattr(
        ApiKeyService, "_revoke_cloud_resources", fake_cloud,
    )

    service = ApiKeyService(db_session)
    await service.submit_release_request(
        assignment.id,
        ApiKeyReleaseRequest(reason_code="no_longer_needed"),
        current_user=_user_auth(employee),
        trace_id="t-peer-submit",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    req = (
        await db_session.execute(
            select(ApiKeyReleaseRequestModel).where(
                ApiKeyReleaseRequestModel.assignment_id == assignment.id,
            ),
        )
    ).scalar_one()

    with pytest.raises(ForbiddenError):
        await service.approve_release_request(
            req.id,
            current_user=_user_auth(bystander),
            decision_note=None,
            trace_id="t-peer-approve",
            ip_address=None,
            user_agent=None,
        )
    # Assignment must still be pending_release; no cloud call.
    await db_session.refresh(assignment)
    assert assignment.status == STATUS_PENDING_RELEASE
    await db_session.refresh(req)
    assert req.status == "pending"


@pytest.mark.asyncio
async def test_admin_can_approve_when_manager_link_broken(
    db_session: AsyncSession, monkeypatch,
):
    """IT Admin escape valve: if the requester's manager_person_id is
    NULL (C-level / contractor) an admin can still approve."""
    _bypass_notify(monkeypatch)

    admin = await _make_person(db_session, email="admin-rel@pingcap.com")
    orphan = await _make_person(db_session, email="orphan-rel@pingcap.com")
    # ``orphan`` has no manager_person_id by default.
    assert orphan.manager_person_id is None
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=orphan,
    )

    async def fake_cloud(self, assignment, *, trace_id):
        return {"attempted": [], "errors": []}

    monkeypatch.setattr(
        ApiKeyService, "_revoke_cloud_resources", fake_cloud,
    )

    service = ApiKeyService(db_session)
    await service.submit_release_request(
        assignment.id,
        ApiKeyReleaseRequest(reason_code="wrong_provider"),
        current_user=_user_auth(orphan),
        trace_id="t-orphan",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    req = (
        await db_session.execute(
            select(ApiKeyReleaseRequestModel).where(
                ApiKeyReleaseRequestModel.assignment_id == assignment.id,
            ),
        )
    ).scalar_one()

    await service.approve_release_request(
        req.id,
        current_user=_admin_auth(admin),
        decision_note="manager link broken; admin clearing",
        trace_id="t-orphan-approve",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    await db_session.refresh(assignment)
    assert assignment.status == STATUS_REVOKED
    assert assignment.revoke_actor_role == ACTOR_ROLE_SELF_WITH_APPROVAL
    assert assignment.revoked_by == orphan.id


@pytest.mark.asyncio
async def test_approve_idempotent_on_already_decided(
    db_session: AsyncSession, monkeypatch,
):
    """Approving an already-approved request does NOT fire the cloud
    teardown a second time. Returns current state."""
    _bypass_notify(monkeypatch)

    manager = await _make_person(db_session, email="idemp-mgr@pingcap.com")
    employee = await _make_person(
        db_session, email="idemp-rel@pingcap.com", manager=manager,
    )
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=employee,
    )

    cloud_calls: list[uuid.UUID] = []

    async def fake_cloud(self, assignment, *, trace_id):
        cloud_calls.append(assignment.id)
        return {"attempted": [], "errors": []}

    monkeypatch.setattr(
        ApiKeyService, "_revoke_cloud_resources", fake_cloud,
    )

    service = ApiKeyService(db_session)
    await service.submit_release_request(
        assignment.id,
        ApiKeyReleaseRequest(reason_code="no_longer_needed"),
        current_user=_user_auth(employee),
        trace_id="t-idemp-submit",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    req = (
        await db_session.execute(
            select(ApiKeyReleaseRequestModel).where(
                ApiKeyReleaseRequestModel.assignment_id == assignment.id,
            ),
        )
    ).scalar_one()

    await service.approve_release_request(
        req.id,
        current_user=_user_auth(manager),
        decision_note=None,
        trace_id="t-idemp-1",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()
    assert cloud_calls == [assignment.id]

    # Second approve on the already-approved row.
    await service.approve_release_request(
        req.id,
        current_user=_user_auth(manager),
        decision_note=None,
        trace_id="t-idemp-2",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()
    # No second cloud call.
    assert cloud_calls == [assignment.id]


@pytest.mark.asyncio
async def test_approve_refuses_leaver_assignee(
    db_session: AsyncSession, monkeypatch,
):
    """If the requester departed between submit and the manager
    clicking Approve, the happy path is short-circuited: the request
    auto-rejects (reason='assignee_offboarded_during_pending'), the
    assignment reverts to ``active`` so the offboarding pipeline can
    own the cleanup, and NO cloud teardown fires."""
    _bypass_notify(monkeypatch)

    manager = await _make_person(db_session, email="leaver-mgr@pingcap.com")
    employee = await _make_person(
        db_session, email="leaver-rel@pingcap.com", manager=manager,
    )
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=employee,
    )

    cloud_calls: list = []

    async def fake_cloud(self, assignment, *, trace_id):
        cloud_calls.append(assignment.id)
        return {"attempted": [], "errors": []}

    monkeypatch.setattr(
        ApiKeyService, "_revoke_cloud_resources", fake_cloud,
    )

    service = ApiKeyService(db_session)
    await service.submit_release_request(
        assignment.id,
        ApiKeyReleaseRequest(reason_code="project_ended"),
        current_user=_user_auth(employee),
        trace_id="t-leaver-submit",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    # Employee departs while manager's notification sits unread.
    employee.employment_status = "leaver_pending"
    await db_session.commit()

    req = (
        await db_session.execute(
            select(ApiKeyReleaseRequestModel).where(
                ApiKeyReleaseRequestModel.assignment_id == assignment.id,
            ),
        )
    ).scalar_one()

    await service.approve_release_request(
        req.id,
        current_user=_user_auth(manager),
        decision_note=None,
        trace_id="t-leaver-approve",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    # No cloud teardown fired under the leaver's identity.
    assert cloud_calls == []

    await db_session.refresh(assignment)
    assert assignment.status == STATUS_ACTIVE
    assert assignment.revoke_actor_role is None

    await db_session.refresh(req)
    assert req.status == "rejected"

    audits = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "apikey.release.refused_leaver",
                AuditLog.resource_id == str(assignment.id),
            ),
        )
    ).scalars().all()
    assert len(audits) == 1


# ── Reject ─────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_reject_by_manager_reverts_to_active(
    db_session: AsyncSession, monkeypatch,
):
    """Manager rejects → no cloud call, assignment reverts to the
    pre-release status, request flips to rejected with the manager's
    decision_note captured."""
    _bypass_notify(monkeypatch)

    manager = await _make_person(db_session, email="rej-mgr@pingcap.com")
    employee = await _make_person(
        db_session, email="carol-rel@pingcap.com", manager=manager,
    )
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=employee,
    )

    async def cloud_must_not_run(self, assignment, *, trace_id):
        raise AssertionError("cloud teardown on REJECT is wrong")

    monkeypatch.setattr(
        ApiKeyService, "_revoke_cloud_resources", cloud_must_not_run,
    )

    service = ApiKeyService(db_session)
    await service.submit_release_request(
        assignment.id,
        ApiKeyReleaseRequest(reason_code="wrong_provider"),
        current_user=_user_auth(employee),
        trace_id="t-rej-submit",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    req = (
        await db_session.execute(
            select(ApiKeyReleaseRequestModel).where(
                ApiKeyReleaseRequestModel.assignment_id == assignment.id,
            ),
        )
    ).scalar_one()

    await service.reject_release_request(
        req.id,
        current_user=_user_auth(manager),
        decision_note="please migrate your code first",
        trace_id="t-rej-do",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    await db_session.refresh(assignment)
    assert assignment.status == STATUS_ACTIVE
    assert assignment.revoked_at is None
    assert assignment.revoked_by is None
    assert assignment.revoke_actor_role is None

    await db_session.refresh(req)
    assert req.status == "rejected"
    assert req.approver_id == manager.id
    assert req.decision_note == "please migrate your code first"
    assert req.responded_at is not None


@pytest.mark.asyncio
async def test_reject_by_non_manager_non_admin_forbidden(
    db_session: AsyncSession, monkeypatch,
):
    """Same authz as approve — a random peer cannot reject."""
    from app.core.exceptions import ForbiddenError

    _bypass_notify(monkeypatch)
    manager = await _make_person(db_session, email="rej-real-mgr@pingcap.com")
    employee = await _make_person(
        db_session, email="rej-peer@pingcap.com", manager=manager,
    )
    bystander = await _make_person(db_session, email="rej-bystander@pingcap.com")
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=employee,
    )

    service = ApiKeyService(db_session)
    await service.submit_release_request(
        assignment.id,
        ApiKeyReleaseRequest(reason_code="no_longer_needed"),
        current_user=_user_auth(employee),
        trace_id="t-rej-peer-sub",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    req = (
        await db_session.execute(
            select(ApiKeyReleaseRequestModel).where(
                ApiKeyReleaseRequestModel.assignment_id == assignment.id,
            ),
        )
    ).scalar_one()

    with pytest.raises(ForbiddenError):
        await service.reject_release_request(
            req.id,
            current_user=_user_auth(bystander),
            decision_note=None,
            trace_id="t-rej-peer",
            ip_address=None,
            user_agent=None,
        )


# ── Cancel ─────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_cancel_release_reverts_and_marks_cancelled(
    db_session: AsyncSession, monkeypatch,
):
    """Requester withdraws their own pending release → assignment
    reverts to active, request row flips to cancelled."""
    _bypass_notify(monkeypatch)

    employee = await _make_person(db_session, email="dan-rel@pingcap.com")
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=employee,
    )

    service = ApiKeyService(db_session)
    await service.submit_release_request(
        assignment.id,
        ApiKeyReleaseRequest(reason_code="no_longer_needed"),
        current_user=_user_auth(employee),
        trace_id="t-rel-dan",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    await service.cancel_release_request(
        assignment.id,
        current_user=_user_auth(employee),
        trace_id="t-rel-dan-cancel",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    await db_session.refresh(assignment)
    assert assignment.status == STATUS_ACTIVE

    req = (
        await db_session.execute(
            select(ApiKeyReleaseRequestModel).where(
                ApiKeyReleaseRequestModel.assignment_id == assignment.id,
            ),
        )
    ).scalar_one()
    assert req.status == "cancelled"
    assert req.responded_at is not None


@pytest.mark.asyncio
async def test_cancel_release_idempotent_when_not_pending(
    db_session: AsyncSession, monkeypatch,
):
    """Cancelling an already-decided release (assignment back to
    active) is a no-op — returns current state."""
    _bypass_notify(monkeypatch)
    employee = await _make_person(db_session, email="dan2-rel@pingcap.com")
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=employee,
    )

    service = ApiKeyService(db_session)
    # No release in flight — cancel must not raise.
    response = await service.cancel_release_request(
        assignment.id,
        current_user=_user_auth(employee),
        trace_id="t-noop-cancel",
        ip_address=None,
        user_agent=None,
    )
    assert response.status == STATUS_ACTIVE


# ── Listing ────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_list_for_approver_returns_only_my_subordinates(
    db_session: AsyncSession, monkeypatch,
):
    """As a manager, listing pending approvals returns only requests
    whose requester reports to me — never another team's queue."""
    _bypass_notify(monkeypatch)

    my_team_mgr = await _make_person(db_session, email="myteam-mgr@pingcap.com")
    other_mgr = await _make_person(db_session, email="otherteam-mgr@pingcap.com")

    my_emp = await _make_person(
        db_session, email="my-emp@pingcap.com", manager=my_team_mgr,
    )
    other_emp = await _make_person(
        db_session, email="other-emp@pingcap.com", manager=other_mgr,
    )

    provider = await _make_provider(db_session, name="Test Provider")
    a1 = await _make_active_assignment(
        db_session, provider=provider, applicant=my_emp,
    )
    a2 = await _make_active_assignment(
        db_session, provider=provider, applicant=other_emp,
    )

    service = ApiKeyService(db_session)
    await service.submit_release_request(
        a1.id, ApiKeyReleaseRequest(reason_code="no_longer_needed"),
        current_user=_user_auth(my_emp),
        trace_id="t-list-1", ip_address=None, user_agent=None,
    )
    await service.submit_release_request(
        a2.id, ApiKeyReleaseRequest(reason_code="project_ended"),
        current_user=_user_auth(other_emp),
        trace_id="t-list-2", ip_address=None, user_agent=None,
    )
    await db_session.commit()

    rows = await service.list_release_requests_for_approver(
        current_user=_user_auth(my_team_mgr), status="pending",
    )
    assert len(rows) == 1
    assert rows[0].requester_id == my_emp.id


@pytest.mark.asyncio
async def test_list_for_admin_returns_all_pending(
    db_session: AsyncSession, monkeypatch,
):
    """IT Admin sees every pending request across teams."""
    _bypass_notify(monkeypatch)

    admin = await _make_person(db_session, email="all-admin@pingcap.com")
    mgr_a = await _make_person(db_session, email="mgr-a@pingcap.com")
    mgr_b = await _make_person(db_session, email="mgr-b@pingcap.com")
    emp_a = await _make_person(
        db_session, email="emp-a@pingcap.com", manager=mgr_a,
    )
    emp_b = await _make_person(
        db_session, email="emp-b@pingcap.com", manager=mgr_b,
    )

    provider = await _make_provider(db_session, name="Test Provider")
    a1 = await _make_active_assignment(
        db_session, provider=provider, applicant=emp_a,
    )
    a2 = await _make_active_assignment(
        db_session, provider=provider, applicant=emp_b,
    )

    service = ApiKeyService(db_session)
    await service.submit_release_request(
        a1.id, ApiKeyReleaseRequest(reason_code="no_longer_needed"),
        current_user=_user_auth(emp_a),
        trace_id="t-all-1", ip_address=None, user_agent=None,
    )
    await service.submit_release_request(
        a2.id, ApiKeyReleaseRequest(reason_code="project_ended"),
        current_user=_user_auth(emp_b),
        trace_id="t-all-2", ip_address=None, user_agent=None,
    )
    await db_session.commit()

    rows = await service.list_release_requests_for_approver(
        current_user=_admin_auth(admin), status="pending",
    )
    requester_ids = {r.requester_id for r in rows}
    assert emp_a.id in requester_ids
    assert emp_b.id in requester_ids


@pytest.mark.asyncio
async def test_list_for_requester_returns_own_only(
    db_session: AsyncSession, monkeypatch,
):
    """The requester-side list returns own requests, never anyone
    else's — regardless of role."""
    _bypass_notify(monkeypatch)

    person_a = await _make_person(db_session, email="self-a@pingcap.com")
    person_b = await _make_person(db_session, email="self-b@pingcap.com")
    provider = await _make_provider(db_session, name="Test Provider")
    aa = await _make_active_assignment(
        db_session, provider=provider, applicant=person_a,
    )
    bb = await _make_active_assignment(
        db_session, provider=provider, applicant=person_b,
    )

    service = ApiKeyService(db_session)
    await service.submit_release_request(
        aa.id, ApiKeyReleaseRequest(reason_code="no_longer_needed"),
        current_user=_user_auth(person_a),
        trace_id="t-self-a", ip_address=None, user_agent=None,
    )
    await service.submit_release_request(
        bb.id, ApiKeyReleaseRequest(reason_code="no_longer_needed"),
        current_user=_user_auth(person_b),
        trace_id="t-self-b", ip_address=None, user_agent=None,
    )
    await db_session.commit()

    rows = await service.list_release_requests_for_requester(
        current_user=_user_auth(person_a),
    )
    assert len(rows) == 1
    assert rows[0].requester_id == person_a.id


# ── Cross-state guards ─────────────────────────────────────────────


@pytest.mark.asyncio
async def test_release_reason_other_requires_note(
    db_session: AsyncSession,
):
    """Schema-layer validator: reason_code='other' without a note →
    Pydantic ValidationError before the service is reached."""
    from pydantic import ValidationError

    with pytest.raises(ValidationError):
        ApiKeyReleaseRequest(reason_code="other", reason_note=None)
    with pytest.raises(ValidationError):
        ApiKeyReleaseRequest(reason_code="other", reason_note="   ")
    body = ApiKeyReleaseRequest(reason_code="other", reason_note="bespoke")
    assert body.reason_note == "bespoke"


@pytest.mark.asyncio
async def test_release_refuses_when_transfer_pending(
    db_session: AsyncSession, monkeypatch,
):
    """Cross-flow mutual exclusion: a pending api_key transfer blocks
    a new release submission (Round-45 hardening)."""
    from app.core.exceptions import AppException
    from app.domains.apikey.schemas import ApiKeyTransferRequestSubmit

    _bypass_notify(monkeypatch)
    employee = await _make_person(db_session, email="frank-rel@pingcap.com")
    recipient = await _make_person(db_session, email="grace-rel@pingcap.com")
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=employee,
    )

    service = ApiKeyService(db_session)
    await service.submit_transfer_request(
        assignment.id,
        ApiKeyTransferRequestSubmit(recipient_person_id=recipient.id),
        current_user=_user_auth(employee),
        trace_id="t-rel-xfer-conflict",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    with pytest.raises(AppException) as excinfo:
        await service.submit_release_request(
            assignment.id,
            ApiKeyReleaseRequest(reason_code="no_longer_needed"),
            current_user=_user_auth(employee),
            trace_id="t-rel-xfer-conflict-2",
            ip_address=None,
            user_agent=None,
        )
    assert excinfo.value.code == "TRANSFER_PENDING"
    assert excinfo.value.status_code == 409


@pytest.mark.asyncio
async def test_extend_validity_blocks_with_clean_error_on_pending_release(
    db_session: AsyncSession, monkeypatch,
):
    """extend_validity surfaces PENDING_RELEASE_BLOCKS_EXTEND 409
    rather than the generic INVALID_STATUS_FOR_EXTEND when the row
    is parked in pending_release, so the UI can render an actionable
    'cancel the release first' message."""
    from app.core.exceptions import AppException
    from app.domains.apikey.schemas import ApiKeyExtendRequest

    _bypass_notify(monkeypatch)
    employee = await _make_person(db_session, email="ext-rel@pingcap.com")
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=employee,
    )

    service = ApiKeyService(db_session)
    await service.submit_release_request(
        assignment.id,
        ApiKeyReleaseRequest(reason_code="no_longer_needed"),
        current_user=_user_auth(employee),
        trace_id="t-ext-rel",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    from datetime import date, timedelta
    with pytest.raises(AppException) as excinfo:
        await service.extend_validity(
            assignment.id,
            ApiKeyExtendRequest(
                new_valid_until=date.today() + timedelta(days=180),
            ),
            current_user=_user_auth(employee),
            trace_id="t-ext-blocked",
            ip_address=None,
            user_agent=None,
        )
    assert excinfo.value.code == "PENDING_RELEASE_BLOCKS_EXTEND"
    assert excinfo.value.status_code == 409


@pytest.mark.asyncio
async def test_release_rejects_invalid_status(
    db_session: AsyncSession, monkeypatch,
):
    """Submitting a release on a revoked / pending_approval row
    raises INVALID_STATUS_FOR_RELEASE."""
    from app.core.exceptions import AppException

    _bypass_notify(monkeypatch)
    employee = await _make_person(db_session, email="evan-rel@pingcap.com")
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=employee,
    )
    assignment.status = STATUS_REVOKED
    await db_session.commit()

    service = ApiKeyService(db_session)
    with pytest.raises(AppException) as excinfo:
        await service.submit_release_request(
            assignment.id,
            ApiKeyReleaseRequest(reason_code="no_longer_needed"),
            current_user=_user_auth(employee),
            trace_id="t-rel-evan",
            ip_address=None,
            user_agent=None,
        )
    assert excinfo.value.code == "INVALID_STATUS_FOR_RELEASE"


# ── Admin revoke / offboarding revoke (regression guards) ──────────


@pytest.mark.asyncio
async def test_admin_revoke_populates_metadata_columns(
    db_session: AsyncSession, monkeypatch,
):
    """Direct admin revoke writes revoke_actor_role='admin',
    revoke_reason_code defaults to 'admin_revoke' when omitted, and
    the revoked_by FK points at the admin's person id."""
    _bypass_notify(monkeypatch)

    admin = await _make_person(db_session, email="admin@pingcap.com")
    other = await _make_person(db_session, email="someone@pingcap.com")
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=other,
    )

    async def fake_cloud(self, assignment, *, trace_id):
        return {"attempted": [], "errors": []}

    monkeypatch.setattr(
        ApiKeyService, "_revoke_cloud_resources", fake_cloud,
    )

    await ApiKeyService(db_session).revoke_key(
        assignment.id,
        reason="leaked AK seen on github",
        current_user=_admin_auth(admin),
        trace_id="t-admin-revoke",
        ip_address=None,
        user_agent=None,
    )
    await db_session.commit()

    await db_session.refresh(assignment)
    assert assignment.status == STATUS_REVOKED
    assert assignment.revoke_actor_role == ACTOR_ROLE_ADMIN
    assert assignment.revoke_reason_code == "admin_revoke"
    assert assignment.revoke_reason_note == "leaked AK seen on github"
    assert assignment.revoked_by == admin.id
    assert assignment.revoked_at is not None


@pytest.mark.asyncio
async def test_admin_revoke_with_explicit_reason_code(
    db_session: AsyncSession, monkeypatch,
):
    """Admin passes ``reason_code='leaked'``; the column captures it."""
    _bypass_notify(monkeypatch)

    admin = await _make_person(db_session, email="adminb@pingcap.com")
    other = await _make_person(db_session, email="someoneb@pingcap.com")
    provider = await _make_provider(db_session, name="Test Provider B")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=other,
    )

    async def fake_cloud(self, assignment, *, trace_id):
        return {"attempted": [], "errors": []}

    monkeypatch.setattr(
        ApiKeyService, "_revoke_cloud_resources", fake_cloud,
    )

    await ApiKeyService(db_session).revoke_key(
        assignment.id,
        reason="found in a Slack screenshot",
        current_user=_admin_auth(admin),
        trace_id="t-admin-revoke-leaked",
        ip_address=None,
        user_agent=None,
        reason_code="leaked",
    )
    await db_session.commit()

    await db_session.refresh(assignment)
    assert assignment.revoke_reason_code == "leaked"


@pytest.mark.asyncio
async def test_offboarding_auto_revoke_invokes_cloud_teardown(
    db_session: AsyncSession, monkeypatch,
):
    """``ApiKeyService.revoke_key`` with ``actor_role=
    ACTOR_ROLE_SYSTEM_OFFBOARDING`` (the path the provisioning
    handler uses on offboarding auto-revoke) MUST fire cloud teardown
    just like the admin / self_with_approval paths.
    """
    from app.domains.apikey.service import ACTOR_ROLE_SYSTEM_OFFBOARDING

    _bypass_notify(monkeypatch)

    admin = await _make_person(db_session, email="admin-offboard@pingcap.com")
    leaver = await _make_person(db_session, email="leaver@pingcap.com")
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = ApiKeyAssignment(
        id=uuid.uuid4(),
        provider_id=provider.id,
        assigned_to=leaver.id,
        status=STATUS_ACTIVE,
        request_number=f"AK-{uuid.uuid4().hex[:8].upper()}",
        project_name="offboarding-test",
        reason="offboarding cascade test",
        monthly_budget_usd=Decimal("100.00"),
        key_value=None,
        detail={},
    )
    db_session.add(assignment)
    await db_session.commit()

    cloud_calls: list = []

    async def fake_cloud(self, assignment, *, trace_id):
        cloud_calls.append(assignment.id)
        return {"attempted": [{"kind": "stub", "ok": True}], "errors": []}

    monkeypatch.setattr(
        ApiKeyService, "_revoke_cloud_resources", fake_cloud,
    )

    await ApiKeyService(db_session).revoke_key(
        assignment.id,
        reason="leaver offboarding",
        current_user=_admin_auth(admin),
        trace_id="t-offboard-revoke",
        ip_address=None,
        user_agent=None,
        reason_code="offboarding",
        actor_role=ACTOR_ROLE_SYSTEM_OFFBOARDING,
    )
    await db_session.commit()

    assert cloud_calls == [assignment.id]

    await db_session.refresh(assignment)
    assert assignment.status == STATUS_REVOKED
    assert assignment.revoke_actor_role == ACTOR_ROLE_SYSTEM_OFFBOARDING
    assert assignment.revoked_by == admin.id
    assert assignment.revoke_reason_code == "offboarding"
    assert assignment.revoked_at is not None


@pytest.mark.asyncio
async def test_system_offboarding_revoke_with_no_actor_writes_null_revoked_by(
    db_session: AsyncSession, monkeypatch,
):
    """Pure cron sweep: current_user=None, actor_role=system →
    revoked_by=NULL but everything else captured."""
    from app.domains.apikey.service import ACTOR_ROLE_SYSTEM_OFFBOARDING

    _bypass_notify(monkeypatch)

    leaver = await _make_person(db_session, email="leaver-cron@pingcap.com")
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = ApiKeyAssignment(
        id=uuid.uuid4(),
        provider_id=provider.id,
        assigned_to=leaver.id,
        status=STATUS_ACTIVE,
        request_number=f"AK-{uuid.uuid4().hex[:8].upper()}",
        project_name="offboarding-cron-test",
        reason="cron path",
        monthly_budget_usd=Decimal("100.00"),
        key_value=None,
        detail={},
    )
    db_session.add(assignment)
    await db_session.commit()

    async def fake_cloud(self, assignment, *, trace_id):
        return {"attempted": [], "errors": []}

    monkeypatch.setattr(
        ApiKeyService, "_revoke_cloud_resources", fake_cloud,
    )

    await ApiKeyService(db_session).revoke_key(
        assignment.id,
        reason="t+0 cron sweep",
        current_user=None,
        trace_id="t-cron-revoke",
        ip_address=None,
        user_agent=None,
        reason_code="offboarding",
        actor_role=ACTOR_ROLE_SYSTEM_OFFBOARDING,
    )
    await db_session.commit()

    await db_session.refresh(assignment)
    assert assignment.status == STATUS_REVOKED
    assert assignment.revoke_actor_role == ACTOR_ROLE_SYSTEM_OFFBOARDING
    assert assignment.revoked_by is None
    assert assignment.revoke_reason_code == "offboarding"


@pytest.mark.asyncio
async def test_admin_revoke_rejects_none_current_user(
    db_session: AsyncSession, monkeypatch,
):
    """Defence-in-depth: actor_role=admin with current_user=None
    must not silently no-op the role gate."""
    from app.core.exceptions import AppException

    _bypass_notify(monkeypatch)

    leaver = await _make_person(db_session, email="x@pingcap.com")
    provider = await _make_provider(db_session, name="Test Provider")
    assignment = ApiKeyAssignment(
        id=uuid.uuid4(),
        provider_id=provider.id,
        assigned_to=leaver.id,
        status=STATUS_ACTIVE,
        request_number=f"AK-{uuid.uuid4().hex[:8].upper()}",
        project_name="defense",
        reason="defense",
        monthly_budget_usd=Decimal("100.00"),
        key_value=None,
        detail={},
    )
    db_session.add(assignment)
    await db_session.commit()

    with pytest.raises(AppException) as excinfo:
        await ApiKeyService(db_session).revoke_key(
            assignment.id,
            reason="x",
            current_user=None,
            trace_id="t-no-actor",
            ip_address=None,
            user_agent=None,
            actor_role=ACTOR_ROLE_ADMIN,
        )
    assert excinfo.value.code == "FORBIDDEN"


@pytest.mark.asyncio
async def test_provisioning_invoke_connector_short_circuits_api_key(
    db_session: AsyncSession,
):
    """PR3-1 regression: ``_invoke_connector`` short-circuits
    ``target_type='api_key'`` tasks with ok=True (the real teardown
    happens in ``_post_success_side_effects``)."""
    from app.domains.provisioning.service import ProvisioningService

    svc = ProvisioningService(db_session)
    fake_task = SimpleNamespace(
        id=uuid.uuid4(),
        software_id=None,
        target_type="api_key",
        target_id=uuid.uuid4(),
        user_id=uuid.uuid4(),
        action="revoke",
        context_data={},
    )
    result = await svc._invoke_connector(fake_task)
    assert result.ok is True
    assert result.external_ref == str(fake_task.target_id)


# ── Regression: release-decision notification deep-link URL ────────


@pytest.mark.asyncio
@pytest.mark.parametrize("approved", [True, False])
async def test_release_decided_notification_uses_canonical_apikey_url(
    db_session: AsyncSession, monkeypatch, approved: bool,
):
    """``_notify_release_decided`` must point the requester at the
    canonical ``/me/api-keys`` route (which the mobile deep-link mapper
    rewrites to ``/m/api-keys``), NOT the stale ``/my/apikeys`` outlier
    that 404'd on desktop and fell through to the mobile home screen.

    Regression guard for the mobile-experience overhaul PR2 URL fix.
    """
    requester = await _make_person(db_session, email="rel-url@pingcap.com")
    provider = await _make_provider(db_session, name="URL Provider")
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=requester,
    )
    request = ApiKeyReleaseRequestModel(
        id=uuid.uuid4(),
        assignment_id=assignment.id,
        requester_id=requester.id,
        reason_code="no_longer_needed",
        status="approved" if approved else "rejected",
        previous_status=STATUS_ACTIVE,
        decision_note="ok" if not approved else None,
    )
    db_session.add(request)
    await db_session.commit()

    # Capture the envelope the notifier would dispatch.
    captured: dict[str, str | None] = {}

    from app.notifications.service import NotificationService

    async def _capture_dispatch(self, envelope, *args, **kwargs):  # noqa: ANN001
        captured["url"] = envelope.url
        captured["context_url"] = (envelope.context or {}).get("my_url")
        return None

    monkeypatch.setattr(NotificationService, "dispatch", _capture_dispatch)

    service = ApiKeyService(db_session)
    await service._notify_release_decided(
        request, assignment, approved=approved, trace_id="t-rel-url",
    )

    assert captured.get("url") is not None
    assert captured["url"].endswith("/me/api-keys"), captured["url"]
    assert "/my/apikeys" not in captured["url"]
    # The template context mirror must agree with the button URL.
    assert captured["context_url"] == captured["url"]