"""Regression tests for the Feishu cancel-desync visibility primitive.

Prod 2026-06-10: a user withdrew a DNS request in Helix; the best-effort
``/approval/v4/instances/cancel`` failed with Feishu 60009 (template has
「允许撤销审批中的申请」 disabled). Helix showed 撤回成功, the Feishu card
stayed 审批中, approvers kept getting todos — the divergence was
completely invisible. The fix (this PR):

  * every withdraw-style cancel path records
    ``last_feishu_cancel_synced`` on its service and the router/MCP
    response carries ``feishu_cancel_synced`` so the frontend can warn;
  * :func:`app.approval.cancel_desync.notify_feishu_cancel_desync`
    lands an in-app (outbox-only) notification on every active
    global_admin, carrying the numeric Feishu error code, deduped per
    flow over 24h, and never raises into the withdraw path.

Covers the helper unit behaviors plus two end-to-end HTTP paths with
opposite local-flip orderings: VPN (Feishu cancel before local flip)
and DNS (local commit before Feishu cancel).
"""
from __future__ import annotations

import json
import uuid
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest
from sqlalchemy import select

from app.approval.cancel_desync import (
    DESYNC_TEMPLATE_KEY,
    notify_feishu_cancel_desync,
)
from app.core.exceptions import AppException
from app.notifications.models import NotificationOutbox


def _feishu_60009() -> AppException:
    # Same shape FeishuApprovalClient._request raises — the numeric code
    # must survive into the notification context via
    # parse_feishu_error_code.
    return AppException(
        code="FEISHU_API_ERROR",
        message="Feishu API error 60009: 审批定义不支持撤回",
        status_code=502,
    )


async def _desync_rows(db_session) -> list[NotificationOutbox]:
    return list(
        (
            await db_session.execute(
                select(NotificationOutbox).where(
                    NotificationOutbox.template_key == DESYNC_TEMPLATE_KEY,
                )
            )
        ).scalars().all()
    )


# ── Helper unit behaviors ────────────────────────────────────────


@pytest.mark.asyncio
async def test_notify_dispatches_to_global_admin_with_error_code(
    db_session, admin_user,
):
    admin, _ = admin_user
    sent = await notify_feishu_cancel_desync(
        db_session,
        flow_label="单元测试流",
        resource_type="unit_test_flow",
        resource_id=str(uuid.uuid4()),
        instance_code="INST-UNIT-1",
        error=_feishu_60009(),
        trace_id="t-unit",
    )
    assert sent is True
    rows = await _desync_rows(db_session)
    assert rows, "expected an outbox row for the global admin"
    recipients = {r.recipient_person_id for r in rows}
    assert admin.id in recipients
    row = next(r for r in rows if r.recipient_person_id == admin.id)
    assert row.context["feishu_error_code"] == 60009
    assert row.context["instance_code"] == "INST-UNIT-1"
    # In-app only: no external channel is attempted (the Feishu API is
    # exactly what just failed), so the row must not sit pending retry.
    assert row.channels == []
    # The code reaches the reader through the template, not through a
    # hardcoded body: the row stores the ``(see template)`` sentinel and
    # ``render_inbox_body`` interpolates the locale-fragment triple, so the
    # admin's own language decides the wording around it.
    assert [
        row.context[f"error_code_label_{lang}"] for lang in ("zh", "en", "ja")
    ] == ["60009", "60009", "60009"]


@pytest.mark.asyncio
async def test_notify_dedupes_same_flow_within_window(db_session, admin_user):
    first = await notify_feishu_cancel_desync(
        db_session,
        flow_label="单元测试流",
        resource_type="unit_test_dedup_flow",
        resource_id=str(uuid.uuid4()),
        instance_code="INST-DEDUP-1",
        error=_feishu_60009(),
    )
    second = await notify_feishu_cancel_desync(
        db_session,
        flow_label="单元测试流",
        resource_type="unit_test_dedup_flow",
        resource_id=str(uuid.uuid4()),  # different request, same flow
        instance_code="INST-DEDUP-2",
        error=_feishu_60009(),
    )
    assert first is True
    assert second is False
    rows = [
        r for r in await _desync_rows(db_session)
        if r.related_resource_type == "unit_test_dedup_flow"
    ]
    # one admin recipient → exactly one row despite two failures
    assert len(rows) == 1


@pytest.mark.asyncio
async def test_notify_never_raises(db_session, admin_user, monkeypatch):
    # Poison the recipient lookup — the helper must swallow and return
    # False rather than break the caller's withdraw path.
    import app.scheduled_jobs.failure_notifier as fn

    async def _boom(session):
        raise RuntimeError("recipient lookup exploded")

    monkeypatch.setattr(fn, "list_admin_recipients", _boom)
    sent = await notify_feishu_cancel_desync(
        db_session,
        flow_label="单元测试流",
        resource_type="unit_test_raise_flow",
        resource_id=str(uuid.uuid4()),
        instance_code="INST-RAISE-1",
        error=_feishu_60009(),
    )
    assert sent is False


@pytest.mark.asyncio
async def test_notify_non_feishu_error_has_null_code(db_session, admin_user):
    sent = await notify_feishu_cancel_desync(
        db_session,
        flow_label="单元测试流",
        resource_type="unit_test_nullcode_flow",
        resource_id=str(uuid.uuid4()),
        instance_code="INST-NULL-1",
        error=TimeoutError("connect timeout"),
    )
    assert sent is True
    rows = [
        r for r in await _desync_rows(db_session)
        if r.related_resource_type == "unit_test_nullcode_flow"
    ]
    assert rows[0].context["feishu_error_code"] is None


@pytest.mark.asyncio
async def test_notify_redacts_provider_credentials_from_persisted_payload(
    db_session, admin_user,
):
    secrets = ("bearer-secret", "provider-secret-key", "query-secret")
    error = RuntimeError(
        "Authorization: Bearer bearer-secret "
        "SecretKey=provider-secret-key "
        "https://provider.invalid/cancel?token=query-secret"
    )

    sent = await notify_feishu_cancel_desync(
        db_session,
        flow_label="单元测试流",
        resource_type="unit_test_redacted_flow",
        resource_id=str(uuid.uuid4()),
        instance_code="INST-REDACTED-1",
        error=error,
    )

    assert sent is True
    row = next(
        row for row in await _desync_rows(db_session)
        if row.related_resource_type == "unit_test_redacted_flow"
    )
    persisted = f"{row.body_text}\n{json.dumps(row.context)}"
    for secret in secrets:
        assert secret not in persisted
    # The scrubbed error travels on the context (``{error_detail}`` in the
    # ``approval_feishu_cancel_desync`` template) rather than in a hardcoded
    # Chinese body — the redaction must survive that move, not just the
    # rendering of it.
    detail = row.context["error_detail"]
    assert "RuntimeError" in detail
    assert "<redacted>" in detail
    assert "<query-redacted>" in detail


# ── End-to-end: VPN cancel (Feishu cancel BEFORE local flip) ─────


async def _make_vpn_app(db_session, person_id) -> str:
    from app.domains.vpn_application.models import VpnApplication

    row = VpnApplication(
        person_id=str(person_id),
        vpn_node="tokyo-gw",
        usage_description="pytest desync",
        resource_checklist="git, ci",
        status="pending",
        feishu_instance_code=f"INST-VPN-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(row)
    await db_session.commit()
    return str(row.id)


@pytest.mark.asyncio
async def test_vpn_cancel_feishu_failure_flags_response_and_notifies(
    client, db_session, employee_user, admin_user, monkeypatch,
):
    me, token = employee_user
    admin, _ = admin_user
    app_id = await _make_vpn_app(db_session, me.id)

    from app.domains.vpn_application.service import VpnApplicationService

    async def _broken_client(self):
        raise _feishu_60009()

    monkeypatch.setattr(VpnApplicationService, "_client", _broken_client)

    resp = await client.post(
        f"/api/v1/vpn-applications/{app_id}/cancel",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["success"] is True
    # Local cancel still wins; the desync is surfaced, not blocking.
    assert body["data"]["status"] == "cancelled"
    assert body["data"]["feishu_cancel_synced"] is False

    rows = [
        r for r in await _desync_rows(db_session)
        if r.related_resource_type == "vpn_application"
    ]
    assert rows, "global admin must get an in-app desync notification"
    assert rows[0].recipient_person_id == admin.id
    assert rows[0].context["feishu_error_code"] == 60009


@pytest.mark.asyncio
async def test_vpn_cancel_feishu_success_reports_synced(
    client, db_session, employee_user, admin_user, monkeypatch,
):
    me, token = employee_user
    app_id = await _make_vpn_app(db_session, me.id)

    import app.domains.vpn_application.service as vpn_service_module
    from app.domains.vpn_application.service import VpnApplicationService

    class _FakeClient:
        async def resolve_user_id_by_email(self, email):
            return "feishu-user-1"

        async def cancel_instance(self, **kwargs):
            return {}

    async def _ok_client(self):
        return _FakeClient()

    monkeypatch.setattr(VpnApplicationService, "_client", _ok_client)
    monkeypatch.setattr(
        vpn_service_module,
        "get_approval_flow",
        AsyncMock(return_value=SimpleNamespace(external_code="VPN_CODE")),
    )

    resp = await client.post(
        f"/api/v1/vpn-applications/{app_id}/cancel",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200
    assert resp.json()["data"]["feishu_cancel_synced"] is True

    rows = [
        r for r in await _desync_rows(db_session)
        if r.related_resource_type == "vpn_application"
    ]
    assert rows == [], "no desync notification on a clean Feishu cancel"


# ── End-to-end: DNS cancel (local commit BEFORE Feishu cancel) ───


@pytest.mark.asyncio
async def test_dns_cancel_feishu_failure_flags_response_and_notifies(
    client, db_session, employee_user, admin_user, monkeypatch,
):
    me, token = employee_user
    admin, _ = admin_user

    from app.domains.cmdb.models import (
        DNS_REQ_ACTION_CREATE,
        DNS_REQ_PENDING,
        CmdbDnsRecordRequest,
    )
    from app.integrations.models import Integration

    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="tencent",
        name=f"pytest-desync-{uuid.uuid4().hex[:8]}",
        slug=f"pytest-desync-{uuid.uuid4().hex[:8]}",
        enabled=True,
        is_primary=False,
        capabilities=["dns_records"],
    )
    db_session.add(integ)
    await db_session.flush()
    row = CmdbDnsRecordRequest(
        requester_person_id=str(me.id),
        action=DNS_REQ_ACTION_CREATE,
        integration_id=str(integ.id),
        zone_name="example.com",
        name="api",
        record_type="A",
        value="10.0.0.1",
        ttl=600,
        line=None,
        purpose="pytest: feishu cancel desync regression",
        status=DNS_REQ_PENDING,
        feishu_instance_code=f"INST-DNS-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(row)
    await db_session.commit()
    req_id = str(row.id)

    from app.domains.cmdb.dns_request_service import DnsRequestService

    async def _broken_client(self):
        raise _feishu_60009()

    monkeypatch.setattr(DnsRequestService, "_client", _broken_client)

    resp = await client.post(
        f"/api/v1/cmdb/dns-record-requests/{req_id}/cancel",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["data"]["status"] == "cancelled"
    assert body["data"]["feishu_cancel_synced"] is False

    rows = [
        r for r in await _desync_rows(db_session)
        if r.related_resource_type == "cmdb_dns_record_request"
    ]
    assert rows, "global admin must get an in-app desync notification"
    assert rows[0].recipient_person_id == admin.id
    assert rows[0].context["feishu_error_code"] == 60009