"""Tests for ``ApprovalService.replay_callback_from_feishu``.

Bug it addresses (production incident 2026-05-28): Feishu webhook
delivery is at-most-once. When the callback for an APPROVED instance
is dropped, the original code paths leave the Helix row stuck on
``status='pending'`` forever:

* ``handle_feishu_callback`` never runs (no webhook).
* ``get_admin_approval_request_detail`` refreshed the cached timeline
  but did not update ``row.status``.
* ``feishu.approval_pending_reaper`` explicitly skipped COMPLETED state
  with the comment "callback should land" — which is exactly what
  doesn't happen on a dropped webhook.

The new ``replay_callback_from_feishu`` is the recovery path: it
fetches Feishu's live snapshot, confirms a terminal state, then runs
the same dispatcher path the original webhook would have. Idempotent
on re-invocation so the reaper + the admin refresh button can both
call it without double-applying side effects.
"""
from __future__ import annotations

import uuid
from unittest.mock import AsyncMock, patch

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

from app.approval.models import ApprovalRequest
from app.approval.service import ApprovalService
from app.integrations.models import Integration, IntegrationApprovalFlow

_FEISHU_CFG = {"app_id": "cli_test", "app_secret": "secret_test"}
_TEMPLATE_CODE = "AE3F30C0-0011-44A0-BC4D-EBC1A4F0CE10"


@pytest.fixture(autouse=True)
def _patch_feishu_cfg(monkeypatch):
    async def _fake_cfg(**_):
        return dict(_FEISHU_CFG)

    monkeypatch.setattr("app.integrations.compat.get_feishu_config", _fake_cfg)
    monkeypatch.setattr("app.approval.service.get_feishu_config", _fake_cfg)

    async def _fake_token(self):
        return "fake-tenant-token"
    monkeypatch.setattr(
        "app.approval.feishu_client.FeishuApprovalClient.get_tenant_access_token",
        _fake_token,
    )


async def _seed_flow(db_session, flow_key: str = "sso_grant_request"):
    integ = Integration(
        id=uuid.uuid4(),
        kind="chat_approval",
        provider="feishu",
        name="Feishu Test",
        slug=f"feishu-replay-{flow_key}",
        enabled=True,
        is_primary=True,
        config_public={"app_id": "cli_test"},
    )
    db_session.add(integ)
    await db_session.flush()
    flow = IntegrationApprovalFlow(
        id=uuid.uuid4(),
        integration_id=integ.id,
        flow_key=flow_key,
        label="Replay test flow",
        external_code=_TEMPLATE_CODE,
        active=True,
        accept_external_origin=False,
    )
    db_session.add(flow)
    await db_session.commit()
    return integ, flow


async def _seed_pending_request(
    db_session: AsyncSession,
    instance_code: str,
    approval_type: str = "sso_grant_request",
) -> ApprovalRequest:
    req = ApprovalRequest(
        id=uuid.uuid4(),
        external_id=instance_code,
        approval_type=approval_type,
        applicant_id=None,
        status="pending",
        source_payload={},
    )
    db_session.add(req)
    await db_session.commit()
    return req


@pytest.mark.asyncio
async def test_replay_no_instance_code(db_session):
    """Edge case: empty instance_code returns ``no_instance_code`` no-op."""
    svc = ApprovalService(db_session)
    result = await svc.replay_callback_from_feishu(
        instance_code="",
        actor_user_id=None,
    )
    assert result["replayed"] is False
    assert result["reason"] == "no_instance_code"


@pytest.mark.asyncio
async def test_replay_row_not_found(db_session):
    """Edge case: unknown instance_code returns ``row_not_found`` no-op.

    Important so an admin clicking refresh on a stale row in their
    browser doesn't 500 when the underlying ApprovalRequest was
    hard-deleted by an ops migration.
    """
    svc = ApprovalService(db_session)
    result = await svc.replay_callback_from_feishu(
        instance_code="feishu_does_not_exist_001",
        actor_user_id=None,
    )
    assert result["replayed"] is False
    assert result["reason"] == "row_not_found"


@pytest.mark.asyncio
async def test_replay_already_terminal_short_circuits(db_session):
    """A row already in ``approved`` short-circuits without a Feishu fetch.

    Critical for idempotency: the reaper + the admin refresh button
    can both race to replay the same row (e.g. the reaper runs at the
    moment an admin clicks refresh). Whichever loses the race must NOT
    re-run any side effects.
    """
    instance_code = "feishu_replay_terminal_001"
    req = await _seed_pending_request(db_session, instance_code)
    req.status = "approved"
    await db_session.commit()

    fetch_spy = AsyncMock()
    with patch(
        "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
        new=fetch_spy,
    ):
        svc = ApprovalService(db_session)
        result = await svc.replay_callback_from_feishu(
            instance_code=instance_code,
            actor_user_id=None,
        )
    # Did not even bother calling Feishu — the terminal-status guard
    # caught it before the fetch.
    fetch_spy.assert_not_called()
    assert result["replayed"] is False
    assert result["reason"] == "already_terminal"
    assert result["previous_status"] == "approved"
    assert result["new_status"] == "approved"


@pytest.mark.asyncio
async def test_replay_feishu_still_pending_noop(db_session):
    """If Feishu also reports PENDING, replay short-circuits.

    The reaper would otherwise loop forever on this row (Feishu says
    pending, Helix says pending, but the reaper would still call
    replay every day). Returning a clean no-op lets the caller log +
    move on without writing an audit row.
    """
    instance_code = "feishu_replay_still_pending_001"
    await _seed_flow(db_session, flow_key="sso_grant_request")
    await _seed_pending_request(db_session, instance_code)

    svc = ApprovalService(db_session)
    result = await svc.replay_callback_from_feishu(
        instance_code=instance_code,
        actor_user_id=None,
        prefetched_instance_detail={"status": "PENDING"},
    )
    assert result["replayed"] is False
    assert result["reason"] == "feishu_still_pending"
    assert result["previous_status"] == "pending"
    assert result["new_status"] == "pending"

    # Confirm DB row is still pending.
    db_session.expire_all()
    row = (await db_session.execute(
        select(ApprovalRequest).where(ApprovalRequest.external_id == instance_code),
    )).scalar_one()
    assert row.status == "pending"


@pytest.mark.asyncio
async def test_replay_feishu_fetch_failure_noop(db_session):
    """A Feishu transient error should NOT raise — return the failure
    reason so the caller (admin refresh / cron) can surface a yellow
    banner without 500'ing the endpoint.
    """
    instance_code = "feishu_replay_fetch_fail_001"
    await _seed_flow(db_session)
    await _seed_pending_request(db_session, instance_code)

    async def _boom(self, ic):
        raise RuntimeError("Feishu open-API 503")

    with patch(
        "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
        new=_boom,
    ):
        svc = ApprovalService(db_session)
        result = await svc.replay_callback_from_feishu(
            instance_code=instance_code,
            actor_user_id=None,
        )
    assert result["replayed"] is False
    assert result["reason"].startswith("feishu_fetch_failed:")
    assert result["previous_status"] == "pending"
    # Row left untouched.
    db_session.expire_all()
    row = (await db_session.execute(
        select(ApprovalRequest).where(ApprovalRequest.external_id == instance_code),
    )).scalar_one()
    assert row.status == "pending"


@pytest.mark.asyncio
async def test_replay_happy_path_flips_status_and_audits(db_session):
    """Terminal Feishu state + pending Helix row → replay runs the
    dispatcher and Helix catches up.

    Uses ``sso_grant_request`` so the per-domain handler can be
    mocked at the import boundary — we only care that
    ``replay_callback_from_feishu`` delegates to
    ``handle_feishu_callback`` with the prefetched detail and that
    ``ApprovalRequest.status`` ends up ``approved``.
    """
    instance_code = "feishu_replay_happy_001"
    await _seed_flow(db_session, flow_key="sso_grant_request")
    await _seed_pending_request(
        db_session, instance_code, approval_type="sso_grant_request",
    )

    detail = {
        "instance_code": instance_code,
        "approval_code": _TEMPLATE_CODE,
        "status": "APPROVED",
    }

    # Patch the per-domain handler so we don't need an SsoGrant row
    # — we're testing the replay glue, not the SSO grant business
    # logic (which has its own tests).
    async def _fake_sso_handle(self, *, approval_request, instance_detail):
        # The dispatcher already flipped approval_request.status above
        # this call (see approval/service.py:1484). Don't touch it
        # here — that would mask a regression where the dispatcher
        # stops doing so.
        return None

    with (
        patch(
            "app.domains.sso.service.SsoGrantService.handle_feishu_callback",
            new=_fake_sso_handle,
        ),
    ):
        svc = ApprovalService(db_session)
        result = await svc.replay_callback_from_feishu(
            instance_code=instance_code,
            actor_user_id=None,
            prefetched_instance_detail=detail,
        )
        await db_session.commit()

    assert result["replayed"] is True
    assert result["reason"] == "ok"
    assert result["previous_status"] == "pending"
    assert result["new_status"] == "approved"
    assert result["feishu_status"] == "APPROVED"

    db_session.expire_all()
    row = (await db_session.execute(
        select(ApprovalRequest).where(ApprovalRequest.external_id == instance_code),
    )).scalar_one()
    assert row.status == "approved"


def test_map_feishu_status_timeout_treated_as_rejected() -> None:
    """Feishu TIMEOUT is a terminal state (Feishu auto-closed because the
    per-template max-duration ran out, no approval decision). Pre-fix it
    mapped to ``pending``, which left rows stuck forever via both the
    callback AND the replay path. Post-fix it maps to ``rejected`` so
    the per-domain handler runs its rejection-side cleanup.
    """
    from app.approval.service import ApprovalService

    assert ApprovalService._map_feishu_status("TIMEOUT") == "rejected"
    assert ApprovalService._map_feishu_status("timeout") == "rejected"
    # Existing mappings unchanged
    assert ApprovalService._map_feishu_status("APPROVED") == "approved"
    assert ApprovalService._map_feishu_status("REJECTED") == "rejected"
    assert ApprovalService._map_feishu_status("CANCELED") == "cancelled"
    # COMPLETED is a positive terminal (instance finished its last node) and
    # MUST map to "approved" so every per-domain side-effect gate (which check
    # ``== "approved"``) fires. The old "completed" mapping matched no gate,
    # sealing the row terminal with NO side-effects → orphaned child request.
    assert ApprovalService._map_feishu_status("COMPLETED") == "approved"
    assert ApprovalService._map_feishu_status("completed") == "approved"
    assert ApprovalService._map_feishu_status("PENDING") == "pending"
    assert ApprovalService._map_feishu_status(None) == "pending"


@pytest.mark.asyncio
async def test_replay_uses_prefetched_detail_no_extra_feishu_call(db_session):
    """When the caller supplies ``prefetched_instance_detail`` the
    dispatcher must NOT make a second Feishu fetch — that round-trip
    is the whole reason we plumbed the kwarg through. A backlog of
    reaper-replays would otherwise multiply Feishu API load.
    """
    instance_code = "feishu_replay_prefetched_001"
    await _seed_flow(db_session, flow_key="sso_grant_request")
    await _seed_pending_request(
        db_session, instance_code, approval_type="sso_grant_request",
    )

    fetch_spy = AsyncMock(return_value={
        "instance_code": instance_code,
        "status": "APPROVED",
    })

    async def _fake_sso_handle(self, *, approval_request, instance_detail):
        return None

    with (
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
            new=fetch_spy,
        ),
        patch(
            "app.domains.sso.service.SsoGrantService.handle_feishu_callback",
            new=_fake_sso_handle,
        ),
    ):
        svc = ApprovalService(db_session)
        await svc.replay_callback_from_feishu(
            instance_code=instance_code,
            actor_user_id=None,
            prefetched_instance_detail={
                "instance_code": instance_code,
                "status": "APPROVED",
            },
        )
        await db_session.commit()

    fetch_spy.assert_not_called()