"""Tests for Feishu approval-backed asset self-service APIs."""

import time
from decimal import Decimal
from unittest.mock import AsyncMock, patch

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

from app.approval.models import ApprovalRequest
from app.core.exceptions import AppException
from app.domains.asset.models import (
    AssetEmployeeRequest,
    AssetItem,
    AssetOperation,
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)


@pytest.fixture(autouse=True)
def _approval_idempotency_headers(request):
    for fixture_name in ("employee_headers", "admin_headers"):
        if fixture_name in request.fixturenames:
            request.getfixturevalue(fixture_name)["Idempotency-Key"] = (
                f"test-{request.node.name}"
            )


def _fresh_feishu_headers() -> dict:
    """Mimic the headers Feishu sends on every state-changing push.

    The webhook handler enforces a 5-minute freshness window on the
    token-only verification path (replay protection — pre-launch
    hardening 2026-05-03). Every test POST that exercises the body-
    token branch must supply ``X-Lark-Request-Timestamp`` so the test
    matches what the real Feishu push looks like.
    """
    return {"X-Lark-Request-Timestamp": str(int(time.time()))}


# Matches the 4 asset templates the original tests covered.
#
# 2026-05-02: ``event_verification_token`` is now mandatory at the
# webhook entry (security hardening — empty token returns 503). Tests
# set a fixed value here and supply the matching ``"token"`` in each
# callback payload below; tests covering the missing / mismatched-token
# branches override the cfg locally.
_TEST_FEISHU_VERIFICATION_TOKEN = "test-feishu-token"

_FAKE_FEISHU_CFG = {
    "app_id": "cli_test",
    "app_secret": "secret_test",
    "event_verification_token": _TEST_FEISHU_VERIFICATION_TOKEN,
}

# Approval flows now live in integration_approval_flows. Tests patch
# ``get_approval_flows_map`` to return this mapping so the service
# doesn't depend on a seeded DB row.
_FAKE_APPROVAL_FLOWS = {
    "laptop_replacement":  "7694D6BE-7FEE-4E49-96B5-A106FC4623DD",
    "computer_borrow":     "8D3A735C-B91C-4EF0-9476-6517EC31B110",
    "asset_return":        "E4ED0075-D2D2-479F-85F7-11374B0CC626",
    "offboarding_buyback": "3B6AAF34-4762-4A24-AE3A-8E035B7BEDA8",
    # software_seat / api_key_request intentionally omitted so the
    # 4-template asset-request picker stays to 4 rows these tests cover.
}


@pytest.fixture(autouse=True)
async def _reset_login_rate_limit():
    """Wipe the IP-keyed login rate-limit bucket before each callback
    test. The Feishu webhook gates on ``check_login_rate_limit`` which
    counts every call (not just failures). Running 7+ callback tests in
    one session against the shared Redis would otherwise trip the 20-
    attempts-per-15-min cap and 429 the last test in the file. This
    fixture is independent of Feishu config — kept on its own so the
    cause of failure is obvious from the fixture name.
    """
    from app.core.redis_client import get_redis
    try:
        redis = get_redis()
        async for key in redis.scan_iter(match="login_ratelimit:*"):
            await redis.delete(key)
    except Exception:  # noqa: BLE001 — defensive; tests run without Redis too
        pass
    yield


@pytest.fixture(autouse=True)
def _patch_feishu_config(monkeypatch):
    """Stub out every DB read the Feishu-backed approval flow performs:

    - ``get_feishu_config`` — tenant credentials (app_id/secret)
    - ``get_approval_flows_map`` — ``flow_key → external_code`` map
      (lives in ``integration_approval_flows`` in production)
    - ``FeishuApprovalClient.get_tenant_access_token`` — short-circuit
      the HTTP round-trip
    """
    async def _fake_cfg(**_kwargs):
        return dict(_FAKE_FEISHU_CFG)

    async def _fake_flows(**_kwargs):
        return dict(_FAKE_APPROVAL_FLOWS)

    monkeypatch.setattr("app.integrations.compat.get_feishu_config", _fake_cfg)
    monkeypatch.setattr("app.approval.service.get_feishu_config", _fake_cfg)
    monkeypatch.setattr("app.approval.router.get_feishu_config", _fake_cfg, raising=False)
    monkeypatch.setattr(
        "app.integrations.service.get_approval_flows_map", _fake_flows,
    )

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


class TestAssetFeishuApprovalAPI:
    @pytest.mark.asyncio
    async def test_template_api_only_exposes_applicant_fields(
        self,
        client: AsyncClient,
        employee_headers: dict,
    ):
        with patch(
            "app.approval.service.FeishuApprovalClient.get_approval_definition",
            new=AsyncMock(
                side_effect=[
                    {
                        "approval_name": "Laptop Replacement Request",
                        "form": [
                            # 2026-05-12: mirror the live Feishu form
                            # — Chinese-renamed widget names, a FY-year
                            # spec block at the top, a model-config
                            # input, and a delivery-time advisory at
                            # the bottom. All applicant-facing widgets
                            # must flow through; only ``Payment
                            # Information`` (approver-fills) is hidden.
                            {
                                "id": "spec_block",
                                "name": "说明 1",
                                "type": "text",
                                "value": (
                                    "FY27 财年公司电脑机型说明：\n"
                                    "• MacBook Air：13 英寸，M5 芯片，16GB 内存，512GB 存储。\n"
                                    "• MacBook Pro：14 英寸，M5 芯片，32GB 内存，1TB 存储。\n"
                                    "• ThinkPad X1：14 英寸，Ultra 7 芯片，32GB 内存，1TB SSD 存储。"
                                ),
                            },
                            {"id": "request_type", "name": "申请类型", "type": "radioV2", "required": True},
                            {"id": "device_sn", "name": "设备序列号", "type": "input", "required": True},
                            {"id": "base_location", "name": "Base Location", "type": "radioV2", "required": True},
                            {
                                "id": "model",
                                "name": "申请电脑机型",
                                "type": "radioV2",
                                "required": True,
                                "option": [
                                    {"value": "MacBook Air", "text": "MacBook Air"},
                                    {"value": "MacBook Pro", "text": "MacBook Pro"},
                                    {"value": "ThinkPad X1", "text": "ThinkPad X1"},
                                    {"value": "非标机型", "text": "非标机型"},
                                ],
                            },
                            {"id": "model_config", "name": "机型配置", "type": "input", "required": True},
                            {"id": "reason", "name": "使用电脑说明", "type": "textarea", "required": True},
                            {"id": "buyback", "name": "是否回购旧电脑", "type": "radioV2", "required": True},
                            {
                                "id": "delivery_notice",
                                "name": "说明 2",
                                "type": "text",
                                "value": "资产设备正在加急采购备货中，预计到货时间为 2 周左右，请耐心等待。",
                            },
                            {"id": "payment", "name": "Payment Information", "type": "textarea", "required": True},
                        ],
                    },
                    {
                        "approval_name": "Computer Borrowing",
                        "form": [
                            {"id": "date_interval", "name": "DateInterval", "type": "dateInterval", "required": True},
                            {"id": "scenario", "name": "Requirement Scenario", "type": "textarea", "required": True},
                        ],
                    },
                    {
                        "approval_name": "资产归还",
                        "form": [
                            {"id": "leave_date", "name": "离职日期", "type": "date", "required": True},
                            {"id": "return_type", "name": "归还方式", "type": "radioV2", "required": True},
                            {"id": "serial", "name": "资产序列号", "type": "input", "required": True},
                            {"id": "asset_code", "name": "资产编号", "type": "input", "required": True},
                            {"id": "reset_proof", "name": "恢复出厂设置证明", "type": "image", "required": True},
                            {"id": "courier", "name": "快递单号", "type": "input", "required": True},
                        ],
                    },
                    {
                        "approval_name": "Laptop Purchase Request While Offboarding",
                        "form": [
                            {"id": "resign_date", "name": "Resignation Date", "type": "date", "required": True},
                            {"id": "serial", "name": "Buyback Device Serial Number", "type": "input", "required": True},
                            {"id": "asset_code", "name": "回购设备资产号", "type": "input", "required": True},
                            {"id": "amount", "name": "Amount", "type": "amount", "required": True},
                        ],
                    },
                ]
            ),
        ):
            response = await client.get(
                "/api/v1/approval/asset/templates",
                headers=employee_headers,
            )

        assert response.status_code == 200
        templates = {item["template_key"]: item["widgets"] for item in response.json()["data"]}
        # Laptop replacement: every applicant-facing widget flows
        # through, including the two static description text blocks
        # (FY-year spec sheet + delivery-time notice) and the
        # ``机型配置`` input. ``Payment Information`` is now also
        # applicant-visible (2026-05-18): the frontend auto-fills it
        # read-only from the asset's ``purchase_entity`` →
        # asset_legal_entities bank info so the IT admin doesn't have
        # to retype on every approval. (``Payment Amount`` is no longer
        # approver-only as of 2026-06-24 — it's the buyback price,
        # auto-filled when 是否回购 = 是 — but this fixture's form has no
        # such widget, so the assertion below is unaffected.)
        laptop_widgets = templates["laptop_replacement"]
        assert [widget["name"] for widget in laptop_widgets] == [
            "说明 1",
            "申请类型",
            "设备序列号",
            "Base Location",
            "申请电脑机型",
            "机型配置",
            "使用电脑说明",
            "是否回购旧电脑",
            "说明 2",
            "Payment Information",
        ]
        # Text widgets carry their multi-line display content in the
        # new ``description`` field so the frontend can render them
        # read-only above/below the input fields.
        spec_block = next(w for w in laptop_widgets if w["id"] == "spec_block")
        assert spec_block["type"] == "text"
        assert "FY27" in (spec_block["description"] or "")
        delivery_notice = next(
            w for w in laptop_widgets if w["id"] == "delivery_notice"
        )
        assert delivery_notice["type"] == "text"
        assert "加急采购" in (delivery_notice["description"] or "")
        assert [widget["name"] for widget in templates["asset_return"]] == [
            "离职日期",
            "归还方式",
            "资产序列号",
            "资产编号",
            "恢复出厂设置证明",
        ]
        # 2026-05-12: "Amount" is now applicant-visible on the
        # offboarding_buyback template. The frontend renders it
        # read-only and fills it from the buyback-quote endpoint, but
        # the widget itself still flows through the same template list.
        assert [widget["name"] for widget in templates["offboarding_buyback"]] == [
            "Resignation Date",
            "Buyback Device Serial Number",
            "Amount",
        ]

    @pytest.mark.asyncio
    async def test_employee_can_create_feishu_backed_asset_request(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ):
        """Employee request should create both Helix request and Feishu approval binding.

        Cycle-2 audit fix: ``_assert_asset_caller_authorized`` now requires
        the caller to own / currently borrow the asset for the
        ``asset_return`` template — assign the asset to the test employee
        so the legitimate path still passes.
        """
        employee, _ = employee_user
        asset = AssetItem(
            asset_tag="RET-001",
            category="laptop",
            status="assigned",
            assigned_to=employee.id,
        )
        db_session.add(asset)
        await db_session.commit()

        with (
            patch(
                "app.approval.service.FeishuApprovalClient.get_approval_definition",
                new=AsyncMock(
                    return_value={
                        "approval_name": "资产归还",
                        "form": [
                            {"id": "date_widget", "name": "离职日期", "type": "date", "required": True},
                            {"id": "return_type", "name": "归还方式", "type": "radioV2", "required": True},
                            {"id": "asset_code", "name": "资产编号", "type": "input", "required": True},
                        ],
                    }
                ),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.resolve_user_id_by_email",
                new=AsyncMock(return_value="feishu_user_001"),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.create_instance",
                new=AsyncMock(
                    return_value={
                        "instance_code": "feishu_instance_001",
                        "instance_status": "PENDING",
                    }
                ),
            ),
        ):
            response = await client.post(
                "/api/v1/approval/asset/requests",
                headers=employee_headers,
                json={
                    "template_key": "asset_return",
                    "asset_id": str(asset.id),
                    "reason": "离职归还电脑",
                    "answers": {
                        "date_widget": "2026-03-10",
                        "return_type": "onsite",
                        "asset_code": "RET-001",
                    },
                },
            )

        assert response.status_code == 200
        data = response.json()["data"]
        assert data["request_type"] == "asset_return"
        assert data["approval_instance_code"] == "feishu_instance_001"
        assert data["approval_status"] == "pending"

        request_row = await db_session.execute(
            select(AssetEmployeeRequest).where(
                AssetEmployeeRequest.request_number == data["request_number"]
            )
        )
        employee_request = request_row.scalar_one()
        assert employee_request.detail["feishu"]["instance_code"] == "feishu_instance_001"

    @pytest.mark.asyncio
    async def test_laptop_replacement_skips_required_check_on_hidden_widgets(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ):
        """Regression — applicant-hidden widgets must NOT trigger
        ``APPROVAL_FIELD_REQUIRED`` on submit.

        Feishu marks widgets ``required=True`` statically while gating
        them with ``display_condition`` ("Police Report Receipt" only
        when Type=Lost, "Model configuration" only when Non-standard
        model, the buyback payment cluster only when Yes-buyback, …).
        The applicant-facing template returned by
        ``/approval/asset/templates`` already filters those out; the
        submit path must do the same — otherwise every conditional
        ``laptop_replacement`` / ``computer_borrow`` / ``asset_return``
        / ``offboarding_buyback`` submission 400s with "Missing required
        approval field: <hidden widget>", and the user only sees the
        generic "资产创建失败" toast (bug 2026-05-11, production users
        affected).
        """
        employee, _ = employee_user
        asset = AssetItem(
            asset_tag="LR-001",
            category="laptop",
            status="assigned",
            assigned_to=employee.id,
            brand="MacBook",
            model="Mac Pro",
            serial_number="HHMH40HFP9",
        )
        db_session.add(asset)
        await db_session.commit()

        captured: dict[str, str] = {}

        async def _capture_create_instance(self, **kwargs):
            captured["form"] = kwargs.get("form", "")
            return {
                "instance_code": "feishu_instance_lr_001",
                "instance_status": "PENDING",
            }

        with (
            patch(
                "app.approval.service.FeishuApprovalClient.get_approval_definition",
                new=AsyncMock(
                    return_value={
                        "approval_name": "Laptop Replacement Request",
                        "form": [
                            # Static description text — must flow
                            # through to the applicant but never
                            # serialised back into the submission.
                            {
                                "id": "spec_block",
                                "name": "说明 1",
                                "type": "text",
                                "value": (
                                    "FY27 财年公司电脑机型说明：\n"
                                    "• MacBook Air：13 英寸，M5 芯片，16GB 内存，512GB 存储。\n"
                                    "• MacBook Pro：14 英寸，M5 芯片，32GB 内存，1TB 存储。\n"
                                    "• ThinkPad X1：14 英寸，Ultra 7 芯片，32GB 内存，1TB SSD 存储。"
                                ),
                            },
                            # Applicant-fillable widgets (everything
                            # that is NOT in the approver-only denylist).
                            {"id": "request_type", "name": "申请类型", "type": "radioV2", "required": True},
                            {"id": "device_sn", "name": "设备序列号", "type": "input", "required": True},
                            {"id": "base_location", "name": "Base Location", "type": "radioV2", "required": True},
                            {"id": "model_choice", "name": "申请电脑机型", "type": "radioV2", "required": True},
                            {"id": "model_config", "name": "机型配置", "type": "input", "required": True},
                            {"id": "reason", "name": "使用电脑说明", "type": "textarea", "required": True},
                            {"id": "buyback", "name": "是否回购旧电脑", "type": "radioV2", "required": True},
                            # Payment Information is applicant-visible
                            # (2026-05-18): auto-filled from the asset's
                            # ``purchase_entity`` legal entity. Other
                            # approver-only / system-managed widgets MUST
                            # NOT leak into the applicant form nor the
                            # Feishu submission — listed in
                            # ``_approver_only_widget_names``.
                            {"id": "payment_info", "name": "Payment Information", "type": "textarea", "required": True},
                            {"id": "lost_type", "name": "Type", "type": "radioV2", "required": True},
                            {"id": "police_receipt", "name": "Police Report Receipt", "type": "attachmentV2", "required": True},
                            {"id": "payment_amount", "name": "Payment Amount", "type": "amount", "required": True},
                            {"id": "asset_code", "name": "IT资产编码", "type": "input", "required": True},
                        ],
                    }
                ),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.resolve_user_id_by_email",
                new=AsyncMock(return_value="feishu_user_lr"),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.create_instance",
                new=_capture_create_instance,
            ),
        ):
            response = await client.post(
                "/api/v1/approval/asset/requests",
                headers=employee_headers,
                json={
                    "template_key": "laptop_replacement",
                    "asset_id": str(asset.id),
                    "reason": "M5 14寸 32G",
                    "answers": {
                        "request_type": "retirement_replacement",
                        "device_sn": "HHMH40HFP9",
                        "base_location": "Guangzhou",
                        "model_choice": "non_standard",
                        "model_config": "M5 14寸 32G 1TB",
                        "reason": "M5 14寸 32G",
                        # No buyback → the (now applicant-visible) Payment
                        # Amount widget is dropped server-side, so it
                        # neither requires a value nor leaks into Feishu.
                        "buyback": "no",
                    },
                },
            )

        assert response.status_code == 200, response.text
        data = response.json()["data"]
        assert data["request_type"] == "laptop_replacement"
        assert data["approval_instance_code"] == "feishu_instance_lr_001"

        # The serialized Feishu form must contain ONLY the applicant-
        # fillable widget IDs. Approver-only widgets (lost_type /
        # police_receipt / asset_code) must not leak in. ``payment_amount``
        # is no longer approver-only — it's the buyback price — but it's
        # dropped here because buyback was NOT chosen. Static description
        # text (spec_block) is also stripped at serialise time — it's
        # display-only. ``payment_info`` is
        # included now (2026-05-18): the auto-fill stamps a placeholder
        # for assets with no resolvable purchase_entity, so the
        # required-field guard passes and the approver sees a clear
        # note to backfill before re-submitting.
        import json as _json
        submitted = _json.loads(captured["form"])
        submitted_ids = {item["id"] for item in submitted}
        assert submitted_ids == {
            "request_type", "device_sn", "base_location",
            "model_choice", "model_config", "reason", "buyback",
            "payment_info",
        }, submitted_ids
        payment_payload = next(item for item in submitted if item["id"] == "payment_info")
        assert "采购主体" in payment_payload["value"], payment_payload

    @pytest.mark.asyncio
    async def test_payment_information_unconditional_overwrite_blocks_tamper(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ):
        """Regression — Payment Information MUST be overwritten with the
        server-resolved entity text even when the client supplies a
        forged value.

        Threat model: the frontend renders the widget read-only
        (``disabled`` TextArea), but ``disabled`` is a UI affordance,
        NOT an authoritative server trust boundary. A malicious
        client (curl, MCP agent, replay) can POST arbitrary text under
        the widget id; without the unconditional overwrite, the
        approver's Feishu card would display forged bank info that
        LOOKS like an autofill from purchase_entity. Code-review pass
        2 caught this on 2026-05-19.
        """
        from app.domains.asset.models import AssetItem, AssetLegalEntity

        employee, _ = employee_user
        entity = AssetLegalEntity(
            name="PingCAP CN Test Entity",
            bank_name="ICBC Beijing",
            bank_account="6222 0233 2333 9999",
            bank_account_name="北京平凯星辰科技有限公司",
            currency="CNY",
        )
        db_session.add(entity)
        await db_session.flush()
        asset = AssetItem(
            asset_tag="LR-TAMPER-01",
            category="laptop",
            status="assigned",
            assigned_to=employee.id,
            purchase_entity="PingCAP CN Test Entity",
        )
        db_session.add(asset)
        await db_session.commit()

        captured: dict[str, str] = {}

        async def _capture(self, **kwargs):
            captured["form"] = kwargs.get("form", "")
            return {"instance_code": "fi-tamper-01", "instance_status": "PENDING"}

        with (
            patch(
                "app.approval.service.FeishuApprovalClient.get_approval_definition",
                new=AsyncMock(return_value={
                    "approval_name": "Laptop Replacement Request",
                    "form": [
                        {"id": "buyback", "name": "是否回购旧电脑", "type": "radioV2", "required": True,
                         "option": {"options": [{"value": "yes", "text": "Yes"}, {"value": "no", "text": "No"}]}},
                        {"id": "payment_info", "name": "Payment Information", "type": "textarea", "required": True},
                    ],
                }),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.resolve_user_id_by_email",
                new=AsyncMock(return_value="feishu_user_tamper"),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.create_instance",
                new=_capture,
            ),
        ):
            response = await client.post(
                "/api/v1/approval/asset/requests",
                headers=employee_headers,
                json={
                    "template_key": "laptop_replacement",
                    "asset_id": str(asset.id),
                    "reason": "tamper test",
                    "answers": {
                        "buyback": "no",
                        # Forged bank info — must be discarded by the
                        # unconditional server-side overwrite.
                        "payment_info": "Account 6666 0000 EVIL — Attacker Bank",
                    },
                },
            )
        assert response.status_code == 200, response.text

        import json as _json
        submitted = _json.loads(captured["form"])
        payment_payload = next(
            item for item in submitted if item["id"] == "payment_info"
        )
        # MUST contain the server-resolved entity text, NOT the forged value.
        assert "EVIL" not in payment_payload["value"], (
            f"Server trusted client-supplied Payment Information value: "
            f"{payment_payload['value']!r} — tamper guard broken."
        )
        assert "ICBC Beijing" in payment_payload["value"]
        assert "PingCAP CN Test Entity" in payment_payload["value"]

    @pytest.mark.asyncio
    async def test_owner_entity_single_select_autofilled_from_asset(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ):
        """所属主体 单选 must be server-stamped from the asset's resolved
        legal entity (mirrors the 公司主体 line of Payment Information),
        overwriting any client-supplied value so the constrained option
        stays consistent with the bank dump the approver sees and a
        forged-payload entity swap is discarded. (2026-06-17)
        """
        from app.domains.asset.models import AssetItem, AssetLegalEntity

        employee, _ = employee_user
        entity = AssetLegalEntity(
            name="PINGCAP PTE. LTD.",
            bank_name="DBS Singapore",
            currency="SGD",
        )
        db_session.add(entity)
        await db_session.flush()
        asset = AssetItem(
            asset_tag="LR-ENTITY-01",
            category="laptop",
            status="assigned",
            assigned_to=employee.id,
            purchase_entity="PINGCAP PTE. LTD.",
        )
        db_session.add(asset)
        await db_session.commit()

        captured: dict[str, str] = {}

        async def _capture(self, **kwargs):
            captured["form"] = kwargs.get("form", "")
            return {"instance_code": "fi-entity-01", "instance_status": "PENDING"}

        with (
            patch(
                "app.approval.service.FeishuApprovalClient.get_approval_definition",
                new=AsyncMock(return_value={
                    "approval_name": "Laptop Replacement Request",
                    "form": [
                        {"id": "buyback", "name": "是否回购旧电脑", "type": "radioV2", "required": True,
                         "option": {"options": [{"value": "yes", "text": "Yes"}, {"value": "no", "text": "No"}]}},
                        {"id": "owner_entity", "name": "所属主体", "type": "radioV2", "required": True,
                         "option": {"options": [
                             {"value": "PINGCAP (US), INC.", "text": "PINGCAP (US), INC."},
                             {"value": "PINGCAP PTE. LTD.", "text": "PINGCAP PTE. LTD."},
                         ]}},
                    ],
                }),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.resolve_user_id_by_email",
                new=AsyncMock(return_value="feishu_user_entity"),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.create_instance",
                new=_capture,
            ),
        ):
            response = await client.post(
                "/api/v1/approval/asset/requests",
                headers=employee_headers,
                json={
                    "template_key": "laptop_replacement",
                    "asset_id": str(asset.id),
                    "reason": "entity autofill",
                    "answers": {
                        "buyback": "no",
                        # Forged / wrong entity — the server must overwrite
                        # it with the asset's resolved owning entity.
                        "owner_entity": "PINGCAP (US), INC.",
                    },
                },
            )
        assert response.status_code == 200, response.text

        import json as _json
        submitted = _json.loads(captured["form"])
        owner_payload = next(
            item for item in submitted if item["id"] == "owner_entity"
        )
        assert owner_payload["value"] == "PINGCAP PTE. LTD.", (
            "所属主体 was not stamped from the asset's resolved entity: "
            f"{owner_payload['value']!r}"
        )

    @pytest.mark.asyncio
    async def test_upload_endpoint_widget_type_aware_mime_validation(
        self,
        client: AsyncClient,
        employee_headers: dict,
    ):
        """MIME allowlist depends on widget_type.

        - ``image`` widget → Feishu /im/v1/images for photos, /im/v1/files
          for videos (when the Feishu admin enabled the 视频 checkbox).
          Photos accepted: JPG/PNG/GIF/BMP/WebP. Videos accepted: MP4 /
          QuickTime / WebM / 3GP / AVI / MKV / M4V. HEIC / HEIF / TIFF /
          PDF / Word MUST be rejected at the Helix boundary with
          ``UPLOAD_TYPE_NOT_ALLOWED`` so the user sees a clear hint
          instead of Feishu's opaque 230002.
        - ``attachmentV2`` widget → Feishu /drive/v1/medias/upload_all,
          accepts the full set: image / video / PDF / Word / Excel /
          PPT / HEIC / etc.

        Empty widget_type / unknown widget_type returns
        ``UNSUPPORTED_UPLOAD_WIDGET``.
        """
        # image widget — HEIC rejected with clear hint
        resp = await client.post(
            "/api/v1/approval/asset/uploads?widget_type=image",
            headers=employee_headers,
            files={"file": ("phone.heic", b"\x00\x00\x00\x20ftypheic\x00\x00", "image/heic")},
        )
        assert resp.status_code == 400, resp.text
        body = resp.json()
        assert body["errors"][0]["code"] == "UPLOAD_TYPE_NOT_ALLOWED"
        assert "image/heic" in body["errors"][0]["message"]
        assert "兼容性最佳" in body["errors"][0]["message"]

        # image widget — PDF rejected (would 502 at Feishu's images
        # endpoint with a non-localised error code).
        resp = await client.post(
            "/api/v1/approval/asset/uploads?widget_type=image",
            headers=employee_headers,
            files={"file": ("doc.pdf", b"%PDF-1.7\n", "application/pdf")},
        )
        assert resp.status_code == 400
        assert resp.json()["errors"][0]["code"] == "UPLOAD_TYPE_NOT_ALLOWED"

        # attachmentV2 widget — HEIC + PDF + Word + WebP all accepted
        # (Feishu's /drive/v1/medias/upload_all is permissive). Mock
        # the downstream Feishu call so we don't try to hit the real
        # API — the test is about the MIME gate, not the upload.
        with patch(
            "app.approval.feishu_client.FeishuApprovalClient.upload_attachment_bytes",
            new=AsyncMock(return_value={"file_token": "tok-abc"}),
        ):
            for filename, content_type in [
                ("phone.heic", "image/heic"),
                ("doc.pdf", "application/pdf"),
                ("form.docx", "application/vnd.openxmlformats-officedocument.wordprocessingml.document"),
                ("screenshot.webp", "image/webp"),
            ]:
                resp = await client.post(
                    "/api/v1/approval/asset/uploads?widget_type=attachmentV2",
                    headers=employee_headers,
                    files={"file": (filename, b"payload", content_type)},
                )
                assert resp.status_code == 200, (
                    f"attachmentV2 rejected legal content_type {content_type}: "
                    f"{resp.status_code} {resp.text}"
                )

        # Unknown widget_type returns 400 UNSUPPORTED_UPLOAD_WIDGET.
        resp = await client.post(
            "/api/v1/approval/asset/uploads?widget_type=mysteryWidget",
            headers=employee_headers,
            files={"file": ("a.png", b"\x89PNG\r\n", "image/png")},
        )
        assert resp.status_code == 400
        assert resp.json()["errors"][0]["code"] == "UNSUPPORTED_UPLOAD_WIDGET"

    @pytest.mark.asyncio
    async def test_image_widget_accepts_video_routes_to_files_endpoint(
        self,
        client: AsyncClient,
        employee_headers: dict,
    ):
        """The buyback / asset-return ``恢复出厂设置证明`` widget is
        configured on the Feishu side as ``image`` with the 视频 checkbox
        enabled — both photos AND videos are valid attachments. The
        router must:

        1. Allow ``video/*`` MIME on widget_type=image (previously rejected
           with ``UPLOAD_TYPE_NOT_ALLOWED`` so the user saw 不支持的文件类型
           (video/mp4) and had no path to satisfy the form's "上传视频"
           requirement).
        2. Route the upload to Feishu's ``/im/v1/files`` endpoint via
           ``upload_video`` (not ``upload_image`` which is photo-only and
           would 400 at the Feishu boundary).
        3. Return the ``file_key`` Feishu produces — the form serialiser
           dispatches by token prefix to emit ``{"file_key": ...}`` vs
           ``{"image_key": ...}``.
        """
        with patch(
            "app.approval.feishu_client.FeishuApprovalClient.upload_video_bytes",
            new=AsyncMock(return_value={"file_key": "file_v2_factory_reset_clip"}),
        ) as upload_video_mock:
            resp = await client.post(
                "/api/v1/approval/asset/uploads?widget_type=image",
                headers=employee_headers,
                files={"file": ("reset.mp4", b"\x00\x00\x00\x20ftypisom", "video/mp4")},
            )
        assert resp.status_code == 200, resp.text
        body = resp.json()
        assert body["data"]["token"] == "file_v2_factory_reset_clip"
        assert body["data"]["widget_type"] == "image"
        upload_video_mock.assert_awaited_once()

        # video/quicktime (iPhone .mov) also accepted on the image widget.
        with patch(
            "app.approval.feishu_client.FeishuApprovalClient.upload_video_bytes",
            new=AsyncMock(return_value={"file_key": "file_v2_mov_clip"}),
        ):
            resp = await client.post(
                "/api/v1/approval/asset/uploads?widget_type=image",
                headers=employee_headers,
                files={"file": ("reset.mov", b"\x00\x00\x00\x14ftypqt", "video/quicktime")},
            )
        assert resp.status_code == 200, resp.text

    @pytest.mark.asyncio
    async def test_upload_quota_rejects_before_feishu_or_storage_calls(
        self,
        client: AsyncClient,
        employee_headers: dict,
    ):
        quota = AsyncMock(side_effect=AppException(
            code="UPLOAD_DAILY_QUOTA_EXCEEDED",
            message="daily upload byte quota exceeded",
            status_code=429,
            headers={"Retry-After": "86400"},
        ))
        feishu = AsyncMock(return_value={"image_key": "must-not-run"})
        storage = AsyncMock(return_value="must-not-run")
        with (
            patch("app.core.storage_uploads.enforce_approval_upload_quota", quota),
            patch(
                "app.approval.feishu_client.FeishuApprovalClient.upload_image_bytes",
                feishu,
            ),
            patch("app.core.storage.put_object_bytes", storage),
        ):
            response = await client.post(
                "/api/v1/approval/asset/uploads?widget_type=image&flow=supplement",
                headers=employee_headers,
                files={"file": ("proof.png", b"png-bytes", "image/png")},
            )

        assert response.status_code == 429
        assert response.headers["Retry-After"] == "86400"
        quota.assert_awaited_once()
        assert quota.await_args.args[1] == len(b"png-bytes")
        feishu.assert_not_awaited()
        storage.assert_not_awaited()

    @pytest.mark.asyncio
    async def test_image_widget_rejects_oversize_video(
        self,
        client: AsyncClient,
        employee_headers: dict,
    ):
        """Videos on image widget get the configured 50 MB ceiling
        (vs 10 MB for photos on /im/v1/images). Over-cap upload must 413
        with the per-widget limit named in the message — keeping the
        photo limit text on a video upload would mislead users into
        compressing under 10 MB when the form accepts 30 MB.
        """
        # Patch the ceiling to 5 MB so the test doesn't have to stream
        # 51 MB through httpx, then upload 6 MB so the gate fires. The
        # regression target is the dispatch logic (video → video
        # ceiling, message names the video number, not the 10 MB photo
        # number).
        with patch("app.approval.router.MAX_VIDEO_UPLOAD_SIZE", 5 * 1024 * 1024):
            resp = await client.post(
                "/api/v1/approval/asset/uploads?widget_type=image",
                headers=employee_headers,
                files={"file": ("big.mp4", b"\x00" * (6 * 1024 * 1024), "video/mp4")},
            )
        assert resp.status_code == 413, resp.text
        body = resp.json()
        assert body["errors"][0]["code"] == "UPLOAD_TOO_LARGE"
        assert "5MB" in body["errors"][0]["message"]

    def test_image_widget_serializes_as_raw_string_array(self):
        """Feishu's ``POST /approval/v4/instances`` accepts image
        widget ``value`` only as a **raw JSON array of plain
        file-code strings**. Four shapes were tried live before
        landing this contract:

        1. ``"value": "[{\"image_key\":\"...\"}]"`` → 1390001 控件值不是数组
        2. ``"value": [{"image_key": "..."}]`` → 1390001 Widget value is not string
        3. ``"value": "[\"img_v3_...\"]"`` → 1390001 控件值不是数组
        4. ``"value": ["img_v3_..."]`` → ACCEPTED (current)

        Feishu's 1390001 error has two wordings:
          - "控件值不是数组" — value field itself is not a JSON
            array (e.g. a string was passed).
          - "Widget value is not string" — value IS an array but
            its elements are not plain strings (e.g. wrapped
            objects).

        Read together: value must be ``[str, str, ...]`` —
        raw array, plain string elements.
        """
        from app.approval.service import ApprovalService

        result = ApprovalService._serialize_widget_value(
            "image",
            ["img_v3_photo_alpha", "file_v2_factory_reset_clip", "img_v3_photo_beta"],
        )
        assert isinstance(result, list), (
            f"image widget value must be a raw list of plain strings "
            f"(Feishu 1390001 控件值不是数组 if stringified); got "
            f"{type(result).__name__}: {result!r}"
        )
        assert result == [
            "img_v3_photo_alpha",
            "file_v2_factory_reset_clip",
            "img_v3_photo_beta",
        ]
        for entry in result:
            assert isinstance(entry, str), (
                f"image widget value array elements must be plain "
                f"strings (Feishu 1390001 'Widget value is not "
                f"string' if wrapped in {{image_key:...}} objects); "
                f"got: {entry!r}"
            )

        # Single token (legacy non-list form) still works.
        photo_only = ApprovalService._serialize_widget_value("image", "img_v3_solo")
        assert photo_only == ["img_v3_solo"]
        video_only = ApprovalService._serialize_widget_value("image", "file_v2_solo_video")
        assert video_only == ["file_v2_solo_video"]

    def test_attachmentv2_widget_serializes_as_raw_string_array(self):
        """Same wire contract as image — attachmentV2 ``value`` is a
        raw JSON array of plain file_token strings. Feishu rejects
        both stringified arrays and wrapped-object elements with
        1390001 variants.
        """
        from app.approval.service import ApprovalService

        result = ApprovalService._serialize_widget_value(
            "attachmentV2", ["file_token_alpha", "file_token_beta"],
        )
        assert isinstance(result, list)
        assert result == ["file_token_alpha", "file_token_beta"]
        for entry in result:
            assert isinstance(entry, str)

        single = ApprovalService._serialize_widget_value(
            "attachmentV2", "file_token_solo",
        )
        assert single == ["file_token_solo"]

    @pytest.mark.asyncio
    async def test_create_instance_retries_without_interval_on_form_rejection(self):
        """If Feishu rejects the borrow window, the submit path
        resubmits the identical form minus the interval widget rather
        than failing the applicant. The 2026-08-04 amount-widget
        regression blocked every non-CNY buyback for nine days because
        no such degradation existed.
        """
        import json as _json

        from app.approval.service import ApprovalService
        from app.core.exceptions import AppException

        form = _json.dumps([
            {"id": "date_interval", "type": "dateInterval",
             "value": {"start": "2026-05-18T00:00:00+08:00",
                       "end": "2026-05-24T00:00:00+08:00",
                       "interval": 7, "timezoneOffset": -480}},
            {"id": "scenario", "type": "textarea", "value": "客户演示"},
        ])
        payload = {
            "approval_code": "CB-CODE",
            "feishu_user_id": "feishu_user_cb",
            "form_payload": form,
        }
        seen: list[str] = []

        class _Client:
            async def create_instance(self, *, approval_code, user_id, form, instance_uuid):
                seen.append(form)
                if len(seen) == 1:
                    raise AppException(
                        code="FEISHU_API_ERROR",
                        message=(
                            "Feishu API error 1395006: validate form error "
                            "控件值不合法或者为空。控件= date_interval"
                        ),
                        status_code=502,
                    )
                return {"instance_code": "fi-cb-retry", "instance_status": "PENDING"}

        service = ApprovalService.__new__(ApprovalService)
        instance = await service._create_instance_with_form_fallback(
            client=_Client(), payload=payload, provider_uuid="uuid-1",
        )

        assert instance["instance_code"] == "fi-cb-retry"
        assert len(seen) == 2
        assert [item["id"] for item in _json.loads(seen[1])] == ["scenario"]
        # The degraded form is persisted on the intent payload so a
        # later recovery tick replays what Feishu actually accepted.
        assert [
            item["id"] for item in _json.loads(payload["form_payload"])
        ] == ["scenario"]

    @pytest.mark.asyncio
    async def test_create_instance_reraises_non_form_rejections(self):
        """Only value-validation rejections degrade the form. A 60012
        duplicate-UUID collision must keep reaching the intent
        machine's recovery branch untouched.
        """
        import json as _json

        from app.approval.service import ApprovalService
        from app.core.exceptions import AppException

        payload = {
            "approval_code": "CB-CODE",
            "feishu_user_id": "feishu_user_cb",
            "form_payload": _json.dumps([
                {"id": "date_interval", "type": "dateInterval", "value": {}},
            ]),
        }
        calls: list[str] = []

        class _Client:
            async def create_instance(self, *, approval_code, user_id, form, instance_uuid):
                calls.append(form)
                raise AppException(
                    code="FEISHU_API_ERROR",
                    message="Feishu API error 60012: duplicated instance uuid",
                    status_code=502,
                )

        service = ApprovalService.__new__(ApprovalService)
        with pytest.raises(AppException) as excinfo:
            await service._create_instance_with_form_fallback(
                client=_Client(), payload=payload, provider_uuid="uuid-1",
            )
        assert "60012" in excinfo.value.message
        assert len(calls) == 1

    def test_date_interval_wire_value_matches_feishu_native_shape(self):
        """``dateInterval`` value shape, read off a real Feishu-native
        computer_borrow instance on 2026-08-14:

        ``{"start": "2026-02-25T00:00:00+08:00",
           "end": "2026-06-01T00:00:00+08:00",
           "interval": 97, "timezoneOffset": -480}``

        The three things the pre-2026-08-14 attempts got wrong (and
        1395006'd on) are each asserted: the duration key is
        ``interval`` not ``duration``, it is an inclusive NUMBER of
        days, and ``timezoneOffset`` is present.
        """
        from app.approval.service import ApprovalService

        wire = ApprovalService._date_interval_wire_value({
            "start": "2026-02-25 00:00:00",
            "end": "2026-06-01 00:00:00",
            "duration": "97",
        })
        assert wire == {
            "start": "2026-02-25T00:00:00+08:00",
            "end": "2026-06-01T00:00:00+08:00",
            "interval": 97,
            "timezoneOffset": -480,
        }

        # Mobile submits bare dates; still lands tz-stamped.
        assert ApprovalService._date_interval_wire_value({
            "start": "2026-03-11", "end": "2027-03-11",
        }) == {
            "start": "2026-03-11T00:00:00+08:00",
            "end": "2027-03-11T00:00:00+08:00",
            "interval": 366,
            "timezoneOffset": -480,
        }

        # An RFC3339 boundary from an MCP / replay caller is respected,
        # not double-stamped.
        already_aware = ApprovalService._date_interval_wire_value({
            "start": "2026-03-10T15:00:00+08:00",
            "end": "2026-03-12T15:00:00+08:00",
        })
        assert already_aware["start"] == "2026-03-10T15:00:00+08:00"
        assert already_aware["interval"] == 3

    def test_date_interval_wire_value_drops_unusable_windows(self):
        """A half-filled / reversed / unparseable window serialises to
        ``""`` so ``_serialize_feishu_form`` drops the widget instead
        of 400'ing the applicant. Feishu accepts a submission that
        omits the widget even though the template marks it required —
        that tolerance is what kept borrow requests submittable during
        the 2026-05-19..2026-08-14 SKIP era.
        """
        from app.approval.service import ApprovalService

        for answer in (
            {"start": "2026-05-18 00:00:00"},                  # no end
            {"end": "2026-05-18 00:00:00"},                    # no start
            {"start": "2026-05-24", "end": "2026-05-18"},      # reversed
            {"start": "not-a-date", "end": "2026-05-18"},      # garbage
            {},
            None,
            42,
        ):
            assert ApprovalService._date_interval_wire_value(answer) == "", answer

    def test_serialize_form_omits_unusable_date_interval(self):
        """The widget is required in the Feishu template, but an
        unusable window must not raise APPROVAL_FIELD_REQUIRED — that
        would newly block submits that worked before this widget
        started shipping.
        """
        import json as _json

        from app.approval.schemas import ApprovalTemplateWidget
        from app.approval.service import ApprovalService

        widgets = [
            ApprovalTemplateWidget(
                id="date_interval", name="DateInterval",
                type="dateInterval", required=True,
            ),
            ApprovalTemplateWidget(
                id="scenario", name="Requirement Scenario",
                type="textarea", required=True,
            ),
        ]
        service = ApprovalService.__new__(ApprovalService)
        parsed = _json.loads(service._serialize_feishu_form(
            widgets, {"scenario": "客户演示"},
        ))
        assert [item["id"] for item in parsed] == ["scenario"]

    def test_form_without_date_interval_strips_only_the_interval(self):
        """Submit-path fallback: if Feishu ever rejects the borrow
        window we resubmit the same form minus that widget, so the
        worst case is a card missing the dates rather than a borrow
        request nobody can file.
        """
        import json as _json

        from app.approval.service import ApprovalService

        form = _json.dumps([
            {"id": "date_interval", "type": "dateInterval",
             "value": {"start": "x", "end": "y", "interval": 2}},
            {"id": "scenario", "type": "textarea", "value": "客户演示"},
        ])
        stripped = ApprovalService._form_without_date_interval(form)
        assert _json.loads(stripped) == [
            {"id": "scenario", "type": "textarea", "value": "客户演示"},
        ]
        # Nothing to strip → None, so the caller re-raises the original
        # Feishu rejection instead of retrying an identical payload.
        assert ApprovalService._form_without_date_interval(
            _json.dumps([{"id": "scenario", "type": "textarea", "value": "x"}]),
        ) is None
        assert ApprovalService._form_without_date_interval("not json") is None

    def test_image_widget_value_round_trips_through_form_payload(self):
        """End-to-end ``_serialize_feishu_form`` → outer
        ``json.dumps`` → wire re-parse: on the wire each image
        widget's ``value`` MUST be a raw JSON array of plain
        strings. This pair of assertions catches every 1390001
        rejection path observed in production:

        - "控件值不是数组" — if value is a string (attempts 1, 3).
        - "Widget value is not string" — if value elements aren't
          strings (attempt 2).
        """
        import json as _json

        from app.approval.schemas import ApprovalTemplateWidget
        from app.approval.service import ApprovalService

        widgets = [
            ApprovalTemplateWidget(
                id="reset_proof", name="恢复出厂设置证明", type="image",
                required=True,
            ),
        ]
        service = ApprovalService.__new__(ApprovalService)
        form_str = service._serialize_feishu_form(
            widgets, {"reset_proof": ["img_v3_xyz", "file_v2_clip"]},
        )
        parsed = _json.loads(form_str)
        assert len(parsed) == 1
        entry = parsed[0]
        assert entry["id"] == "reset_proof"
        assert entry["type"] == "image"
        # Catches attempts 1 & 3 (stringified value).
        assert isinstance(entry["value"], list), (
            f"Feishu rejects string value with 1390001 控件值不是数组. "
            f"Got: {type(entry['value']).__name__}: {entry['value']!r}"
        )
        # Catches attempt 2 (wrapped objects as array elements).
        for element in entry["value"]:
            assert isinstance(element, str), (
                f"Feishu rejects non-string array elements with 1390001 "
                f"'Widget value is not string'. Got: {element!r}"
            )
        assert entry["value"] == ["img_v3_xyz", "file_v2_clip"]

    @pytest.mark.asyncio
    async def test_computer_borrow_ships_date_interval_and_stashes_dates_in_detail(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ):
        """Regression — computer_borrow MUST ship the borrow window to
        Feishu in the object shape Feishu itself stores, AND keep
        stashing the dates in ``EmployeeRequest.detail.borrow_dates``
        for the auto-execute callback path to read.

        Bug history: 2026-05-19 production users hit ``Feishu API error
        1395006: validate form error 控件值不合法或者为空。控件=
        widget17326081030290001`` on every computer_borrow submit
        (the shapes tried then keyed the duration as ``duration`` and
        omitted ``timezoneOffset``), so the widget was skipped outright.
        The cost surfaced 2026-08-14: the approver's Feishu card showed
        the reason and pick-up location but no dates at all. The shape
        asserted below was read off a real Feishu-native borrow
        instance via ``GET /approval/v4/instances/{code}``.
        """
        from app.domains.asset.models import AssetItem

        employee, _ = employee_user
        asset = AssetItem(
            asset_tag="BORROW-CB-001",
            category="laptop",
            status="idle",
        )
        db_session.add(asset)
        await db_session.commit()
        asset_id = asset.id

        form_definition = [
            {"id": "date_interval", "name": "DateInterval", "type": "dateInterval", "required": True},
            {"id": "scenario", "name": "Requirement Scenario", "type": "textarea", "required": True},
        ]

        captured: dict[str, str] = {}

        async def _capture(self, **kwargs):
            captured["form"] = kwargs.get("form", "")
            return {"instance_code": "fi-cb-001", "instance_status": "PENDING"}

        with (
            patch(
                "app.approval.service.FeishuApprovalClient.get_approval_definition",
                new=AsyncMock(return_value={
                    "approval_name": "Computer Borrowing",
                    "form": form_definition,
                }),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.resolve_user_id_by_email",
                new=AsyncMock(return_value="feishu_user_cb"),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.create_instance",
                new=_capture,
            ),
        ):
            response = await client.post(
                "/api/v1/approval/asset/requests",
                headers=employee_headers,
                json={
                    "template_key": "computer_borrow",
                    "asset_id": str(asset_id),
                    "reason": "borrow for demo",
                    "answers": {
                        "date_interval": {
                            "start": "2026-05-18 00:00:00",
                            "end": "2026-05-24 00:00:00",
                            "duration": "7",
                        },
                        "scenario": "客户演示",
                    },
                },
            )
        assert response.status_code == 200, response.text

        # 1. The Feishu form payload MUST carry the borrow window in
        #    the exact object shape Feishu stores natively — whole
        #    form item asserted, not just ``value`` (a partial
        #    assertion is how the 2026-08-04 amount-widget regression
        #    stayed green while production 1395006'd).
        import json as _json
        submitted = _json.loads(captured["form"])
        submitted_ids = {item["id"] for item in submitted}
        assert "scenario" in submitted_ids
        interval_item = next(
            item for item in submitted if item["id"] == "date_interval"
        )
        assert interval_item == {
            "id": "date_interval",
            "type": "dateInterval",
            "value": {
                "start": "2026-05-18T00:00:00+08:00",
                "end": "2026-05-24T00:00:00+08:00",
                # Inclusive day count — 05-18..05-24 is 7 days, the same
                # arithmetic the Helix form's 时长（天）field does and
                # the same convention Feishu's own instances carry.
                "interval": 7,
                # JS getTimezoneOffset() convention: minutes west of
                # UTC, so Asia/Shanghai is -480.
                "timezoneOffset": -480,
            },
        }

        # 2. EmployeeRequest.detail.borrow_dates MUST still carry the
        #    dates: Feishu's callback form_items arrives partial under
        #    load, and the submit path degrades to omitting the widget
        #    if Feishu ever rejects the value.
        data = response.json()["data"]
        request_row = await db_session.execute(
            select(AssetEmployeeRequest).where(
                AssetEmployeeRequest.request_number == data["request_number"]
            )
        )
        employee_request = request_row.scalar_one()
        borrow_dates = (employee_request.detail or {}).get("borrow_dates") or {}
        assert borrow_dates.get("start") == "2026-05-18 00:00:00"
        assert borrow_dates.get("end") == "2026-05-24 00:00:00"

    @pytest.mark.asyncio
    async def test_display_condition_hides_widget_when_condition_unmet(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ):
        """``display_condition`` gates conditional widgets at submit time.

        Bug class: pre-2026-05-12, ``_serialize_feishu_form`` only
        checked ``widget.required`` against the answer map without
        honouring ``display_condition`` — so every "No buyback"
        submission 400'd with APPROVAL_FIELD_REQUIRED for the unused
        photo widget. The denylist patched this for known
        approver-only widgets, but a general conditional-required
        gate is the only correct fix.
        """
        from app.domains.asset.models import AssetItem

        employee, _ = employee_user
        asset = AssetItem(
            asset_tag="DC-OFF-001",
            asset_name="MBP 14",
            category="laptop",
            status="assigned",
            assigned_to=employee.id,
            brand="MacBook",
            serial_number="DC0001",
        )
        db_session.add(asset)
        await db_session.commit()
        asset_id = asset.id

        form_definition = [
            {"id": "buyback", "name": "是否回购旧电脑", "type": "radioV2", "required": True},
            {
                "id": "buyback_photo",
                "name": "回购照片",
                "type": "attachmentV2",
                "required": True,
                "display_condition": {
                    "conditions": [
                        {"widget_id": "buyback", "operator": "equal", "value": "yes"},
                    ],
                },
            },
        ]

        captured: dict[str, str] = {}

        async def _capture(self, **kwargs):
            captured["form"] = kwargs.get("form", "")
            return {"instance_code": "fi-dc-001", "instance_status": "PENDING"}

        with (
            patch(
                "app.approval.service.FeishuApprovalClient.get_approval_definition",
                new=AsyncMock(return_value={
                    "approval_name": "Laptop Replacement Request",
                    "form": form_definition,
                }),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.resolve_user_id_by_email",
                new=AsyncMock(return_value="feishu_user_dc"),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.create_instance",
                new=_capture,
            ),
        ):
            response = await client.post(
                "/api/v1/approval/asset/requests",
                headers=employee_headers,
                json={
                    "template_key": "laptop_replacement",
                    "asset_id": str(asset_id),
                    "reason": "no buyback",
                    "answers": {"buyback": "no"},
                },
            )
            assert response.status_code == 200, response.text
            import json as _json
            submitted_ids = {item["id"] for item in _json.loads(captured["form"])}
            # Hidden widget must NOT leak into the Feishu submission.
            assert "buyback_photo" not in submitted_ids
            assert "buyback" in submitted_ids
            # Snapshot must omit the hidden widget for forensic
            # accuracy — admin shouldn't see a buyback_photo row on a
            # "no buyback" record.
            snapshot = response.json()["data"]["detail"]["feishu"]["form_snapshot"]
            snap_ids = {e["id"] for e in snapshot}
            assert "buyback_photo" not in snap_ids
            assert "buyback" in snap_ids

    @pytest.mark.asyncio
    async def test_display_condition_required_fires_when_condition_met(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ):
        """Counterpart of the previous test — when the gate IS met
        (``buyback=yes``), the required-check on the gated widget
        fires normally. Without this assertion, a buggy evaluator
        that always returns False would silently pass both tests.
        """
        from app.domains.asset.models import AssetItem

        employee, _ = employee_user
        asset = AssetItem(
            asset_tag="DC-ON-001",
            asset_name="MBP 14",
            category="laptop",
            status="assigned",
            assigned_to=employee.id,
            brand="MacBook",
            serial_number="DC0002",
        )
        db_session.add(asset)
        await db_session.commit()
        asset_id = asset.id

        form_definition = [
            {"id": "buyback", "name": "是否回购旧电脑", "type": "radioV2", "required": True},
            {
                "id": "buyback_photo",
                "name": "回购照片",
                "type": "attachmentV2",
                "required": True,
                "display_condition": {
                    "conditions": [
                        {"widget_id": "buyback", "operator": "equal", "value": "yes"},
                    ],
                },
            },
        ]

        with (
            patch(
                "app.approval.service.FeishuApprovalClient.get_approval_definition",
                new=AsyncMock(return_value={
                    "approval_name": "Laptop Replacement Request",
                    "form": form_definition,
                }),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.resolve_user_id_by_email",
                new=AsyncMock(return_value="feishu_user_dc2"),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.create_instance",
                new=AsyncMock(return_value={"instance_code": "fi-dc-002", "instance_status": "PENDING"}),
            ),
        ):
            response = await client.post(
                "/api/v1/approval/asset/requests",
                headers=employee_headers,
                json={
                    "template_key": "laptop_replacement",
                    "asset_id": str(asset_id),
                    "reason": "yes buyback no photo",
                    "answers": {"buyback": "yes"},
                },
            )
            assert response.status_code == 400, response.text
            errors = response.json().get("errors") or []
            assert any(
                "回购照片" in (e.get("message") or "")
                or e.get("code") == "APPROVAL_FIELD_REQUIRED"
                for e in errors
            )

    @pytest.mark.asyncio
    async def test_radiov2_rejects_value_outside_option_set(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ):
        """Backstop for Feishu API 1390001 — radioV2 widget rejects
        values not in the registered option list.

        Real bug 2026-05-18: ``buildAssetPrefill`` pushed
        ``asset.model = "Mac Pro"`` (free-form CMDB text) straight
        into the laptop_replacement "申请电脑机型" widget (options:
        MacBook Air / MacBook Pro / ThinkPad X1 / 非标机型). Feishu
        rejected the create_instance call with 1390001 — opaque to
        the caller and requiring a Feishu console round-trip to
        diagnose. ``_serialize_feishu_form`` now rejects with a
        clean 400 APPROVAL_FIELD_INVALID_OPTION naming the widget,
        the offending value, AND the legal options, BEFORE we even
        call Feishu. Defence-in-depth — the frontend prefill has
        been hardened in parallel, but this guard also protects MCP
        / direct API / replay paths.
        """
        from app.domains.asset.models import AssetItem

        employee, _ = employee_user
        asset = AssetItem(
            asset_tag="MAC-PRO-001",
            asset_name="Mac Pro (old data)",
            category="laptop",
            status="assigned",
            assigned_to=employee.id,
            brand="Apple",
            model="Mac Pro",  # mimic the prod data shape that triggered the bug
            serial_number="MP0001",
        )
        db_session.add(asset)
        await db_session.commit()
        asset_id = asset.id

        form_definition = [
            {
                "id": "model_choice",
                "name": "申请电脑机型",
                "type": "radioV2",
                "required": True,
                "option": [
                    {"value": "MacBook Air", "text": "MacBook Air"},
                    {"value": "MacBook Pro", "text": "MacBook Pro"},
                    {"value": "ThinkPad X1", "text": "ThinkPad X1"},
                    {"value": "非标机型", "text": "非标机型"},
                ],
            },
        ]

        with (
            patch(
                "app.approval.service.FeishuApprovalClient.get_approval_definition",
                new=AsyncMock(return_value={
                    "approval_name": "Laptop Replacement Request",
                    "form": form_definition,
                }),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.resolve_user_id_by_email",
                new=AsyncMock(return_value="feishu_user_mp"),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.create_instance",
                new=AsyncMock(return_value={"instance_code": "should-not-be-called", "instance_status": "PENDING"}),
            ) as create_mock,
        ):
            response = await client.post(
                "/api/v1/approval/asset/requests",
                headers=employee_headers,
                json={
                    "template_key": "laptop_replacement",
                    "asset_id": str(asset_id),
                    "reason": "want new laptop",
                    "answers": {"model_choice": "Mac Pro"},  # not in option set
                },
            )

        assert response.status_code == 400, response.text
        errors = response.json().get("errors") or []
        assert any(
            e.get("code") == "APPROVAL_FIELD_INVALID_OPTION"
            and "Mac Pro" in (e.get("message") or "")
            and "申请电脑机型" in (e.get("message") or "")
            for e in errors
        ), errors
        # Feishu must NOT have been called — the guard fires before submission.
        create_mock.assert_not_called()

    @pytest.mark.asyncio
    async def test_radiov2_accepts_value_in_option_set(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ):
        """Happy-path counterpart — a radioV2 answer matching one of
        the registered option values is accepted and serialised
        through to Feishu unchanged. Without this, a buggy guard
        that always raised would still pass the rejection test above.
        """
        from app.domains.asset.models import AssetItem

        employee, _ = employee_user
        asset = AssetItem(
            asset_tag="MAC-OK-001",
            asset_name="MBP",
            category="laptop",
            status="assigned",
            assigned_to=employee.id,
            brand="Apple",
            model="MacBook Pro",
            serial_number="MOK001",
        )
        db_session.add(asset)
        await db_session.commit()
        asset_id = asset.id

        form_definition = [
            {
                "id": "model_choice",
                "name": "申请电脑机型",
                "type": "radioV2",
                "required": True,
                "option": [
                    {"value": "MacBook Air", "text": "MacBook Air"},
                    {"value": "MacBook Pro", "text": "MacBook Pro"},
                ],
            },
        ]

        captured: dict[str, str] = {}

        async def _capture(self, **kwargs):
            captured["form"] = kwargs.get("form", "")
            return {"instance_code": "fi-mp-ok", "instance_status": "PENDING"}

        with (
            patch(
                "app.approval.service.FeishuApprovalClient.get_approval_definition",
                new=AsyncMock(return_value={
                    "approval_name": "Laptop Replacement Request",
                    "form": form_definition,
                }),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.resolve_user_id_by_email",
                new=AsyncMock(return_value="feishu_user_ok"),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.create_instance",
                new=_capture,
            ),
        ):
            response = await client.post(
                "/api/v1/approval/asset/requests",
                headers=employee_headers,
                json={
                    "template_key": "laptop_replacement",
                    "asset_id": str(asset_id),
                    "reason": "want MBP",
                    "answers": {"model_choice": "MacBook Pro"},  # matches option.value
                },
            )

        assert response.status_code == 200, response.text
        import json as _json
        submitted = _json.loads(captured["form"])
        model_entry = next(item for item in submitted if item["id"] == "model_choice")
        assert model_entry["value"] == "MacBook Pro"

    @pytest.mark.asyncio
    async def test_feishu_callback_updates_request_and_asset_metadata(
        self,
        client: AsyncClient,
        employee_user,
        db_session: AsyncSession,
    ):
        """Feishu callback should sync approval status into Helix request and asset tables."""
        employee, _ = employee_user
        asset = AssetItem(
            asset_tag="BUYBACK-001",
            category="laptop",
            status="assigned",
            assigned_to=employee.id,
            custom_fields={},
        )
        db_session.add(asset)
        await db_session.flush()

        request_row = AssetEmployeeRequest(
            request_number="REQ-TEST-001",
            request_type="offboarding_buyback",
            status="pending",
            requester_id=employee.id,
            asset_id=asset.id,
            reason="离职回购",
            detail={
                "feishu": {
                    "instance_code": "feishu_instance_buyback",
                    "approval_code": "3B6AAF34-4762-4A24-AE3A-8E035B7BEDA8",
                    "approval_name": "Laptop Purchase Request While Offboarding",
                    "instance_status": "PENDING",
                }
            },
        )
        db_session.add(request_row)
        await db_session.flush()
        db_session.add(
            ApprovalRequest(
                external_id="feishu_instance_buyback",
                approval_type="offboarding_buyback",
                applicant_id=employee.id,
                status="pending",
                source_payload={
                    "employee_request_id": str(request_row.id),
                    "asset_id": str(asset.id),
                    "approval_code": "3B6AAF34-4762-4A24-AE3A-8E035B7BEDA8",
                },
            )
        )
        await db_session.commit()

        with patch(
            "app.approval.service.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(
                return_value={
                    "instance_code": "feishu_instance_buyback",
                    "status": "APPROVED",
                    "form": [
                        {"name": "Resignation Date", "value": "2026-03-10"},
                        {"name": "Buyback Device Serial Number", "value": "SN-BUYBACK-001"},
                        {"name": "回购设备资产号", "value": "BUYBACK-001"},
                        {"name": "资产所属主体", "value": "PINGCAP (US), INC."},
                        {"name": "Amount", "value": '{"amount":"1000","currency":"CNY"}'},
                        {"name": "Transfer Account", "value": "Bank / Account / Name"},
                    ],
                }
            ),
        ):
            response = await client.post(
                "/api/v1/approval/feishu/callback",
                headers=_fresh_feishu_headers(),
                json={
                    "token": _TEST_FEISHU_VERIFICATION_TOKEN,
                    "header": {"event_type": "approval_instance"},
                    "event": {"instance_code": "feishu_instance_buyback"},
                },
            )

        assert response.status_code == 200
        assert response.json()["code"] == 0

        await db_session.refresh(request_row)
        await db_session.refresh(asset)
        assert request_row.status == "approved"
        assert request_row.detail["feishu"]["instance_status"] == "APPROVED"
        approval_record = asset.custom_fields["feishu_approvals"]["offboarding_buyback"]
        assert approval_record["instance_code"] == "feishu_instance_buyback"
        assert approval_record["resignation_date"] == "2026-03-10"
        assert approval_record["device_serial_number"] == "SN-BUYBACK-001"
        assert approval_record["asset_tag"] == "BUYBACK-001"
        assert approval_record["owner_entity"] == "PINGCAP (US), INC."
        assert approval_record["buyback_amount"] == {"amount": "1000", "currency": "CNY"}
        assert "transfer_account" not in approval_record

        # Auto-execute: buyback approval must now flip the asset to
        # disposed/sold and clear the assignee. The leaver held the
        # laptop at approval time (status=assigned), so this exercises
        # the dispose_asset(force=True) path.
        assert asset.status == "disposed"
        assert asset.disposal_type == "sold"
        assert asset.assigned_to is None
        assert asset.disposed_at is not None
        op_row = (await db_session.execute(
            select(AssetOperation).where(
                AssetOperation.asset_id == asset.id,
                AssetOperation.operation_type == "disposal",
            )
        )).scalar_one()
        assert "offboarding buyback" in (op_row.notes or "").lower()

    @pytest.mark.asyncio
    async def test_feishu_callback_form_as_json_string(
        self,
        client: AsyncClient,
        employee_user,
        db_session: AsyncSession,
    ):
        """Regression (prod incident 2026-07-03): the REAL Feishu
        ``GET /approval/v4/instances/{code}`` ships ``form`` as a
        JSON-encoded STRING. ``_extract_asset_relevant_fields`` iterated
        it as a list → ``AttributeError: 'str' object has no attribute
        'get'`` on EVERY asset-linked APPROVED callback → webhook 500 →
        Feishu gave up → rows sat 待处理 until an admin opened the
        detail drawer. The sibling test above uses a list-shaped mock,
        which is why CI stayed green while prod was broken — this test
        pins the real shape.
        """
        import json as _json
        employee, _ = employee_user
        asset = AssetItem(
            asset_tag="BUYBACK-STR-001",
            category="laptop",
            status="assigned",
            assigned_to=employee.id,
            custom_fields={},
        )
        db_session.add(asset)
        await db_session.flush()

        request_row = AssetEmployeeRequest(
            request_number="REQ-TEST-STR-001",
            request_type="offboarding_buyback",
            status="pending",
            requester_id=employee.id,
            asset_id=asset.id,
            reason="离职回购",
            detail={
                "feishu": {
                    "instance_code": "feishu_instance_buyback_str",
                    "approval_code": "3B6AAF34-4762-4A24-AE3A-8E035B7BEDA8",
                    "instance_status": "PENDING",
                }
            },
        )
        db_session.add(request_row)
        await db_session.flush()
        db_session.add(
            ApprovalRequest(
                external_id="feishu_instance_buyback_str",
                approval_type="offboarding_buyback",
                applicant_id=employee.id,
                status="pending",
                source_payload={
                    "employee_request_id": str(request_row.id),
                    "asset_id": str(asset.id),
                    "approval_code": "3B6AAF34-4762-4A24-AE3A-8E035B7BEDA8",
                },
            )
        )
        await db_session.commit()

        with patch(
            "app.approval.service.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(
                return_value={
                    "instance_code": "feishu_instance_buyback_str",
                    "status": "APPROVED",
                    # The real API shape: form is a JSON STRING.
                    "form": _json.dumps([
                        {"name": "Resignation Date", "value": "2026-06-30"},
                        {"name": "Computer serial number", "value": "SN-STR-001"},
                        {"name": "Computer asset number", "value": "BUYBACK-STR-001"},
                        {
                            "name": "Repurchase price",
                            "type": "amount",
                            "value": '{"amount":"2500","currency":"CNY"}',
                        },
                    ]),
                }
            ),
        ):
            response = await client.post(
                "/api/v1/approval/feishu/callback",
                headers=_fresh_feishu_headers(),
                json={
                    "token": _TEST_FEISHU_VERIFICATION_TOKEN,
                    "header": {"event_type": "approval_instance"},
                    "event": {"instance_code": "feishu_instance_buyback_str"},
                },
            )

        assert response.status_code == 200
        assert response.json()["code"] == 0

        await db_session.refresh(request_row)
        await db_session.refresh(asset)
        # The status flip AND the metadata extraction both survive the
        # string-shaped form.
        assert request_row.status == "approved"
        approval_record = asset.custom_fields["feishu_approvals"]["offboarding_buyback"]
        assert approval_record["device_serial_number"] == "SN-STR-001"
        assert approval_record["asset_tag"] == "BUYBACK-STR-001"
        assert approval_record["buyback_amount"] == {"amount": "2500", "currency": "CNY"}
        # Auto-execute ran too — the whole point of the fix is that the
        # side-effects fire on the webhook instead of never.
        assert asset.status == "disposed"

    @pytest.mark.asyncio
    async def test_feishu_callback_auto_executes_computer_borrow(
        self,
        client: AsyncClient,
        employee_user,
        db_session: AsyncSession,
    ):
        """Feishu approval of a computer_borrow request must auto-borrow
        the asset to the requester with the dates from the form."""
        employee, _ = employee_user
        asset = AssetItem(
            asset_tag="BORROW-LOANER-01",
            category="laptop",
            status="idle",
            custom_fields={},
        )
        db_session.add(asset)
        await db_session.flush()

        request_row = AssetEmployeeRequest(
            request_number="REQ-BORROW-001",
            request_type="computer_borrow",
            status="pending",
            requester_id=employee.id,
            asset_id=asset.id,
            reason="临时借用",
            detail={"feishu": {"instance_code": "feishu_instance_borrow"}},
        )
        db_session.add(request_row)
        await db_session.flush()
        db_session.add(ApprovalRequest(
            external_id="feishu_instance_borrow",
            approval_type="computer_borrow",
            applicant_id=employee.id,
            status="pending",
            source_payload={
                "employee_request_id": str(request_row.id),
                "asset_id": str(asset.id),
            },
        ))
        await db_session.commit()

        # Use future dates — AssetBorrowRequest validates expected_return_date > today.
        from datetime import date as _date
        from datetime import timedelta as _td
        start_date = _date.today() + _td(days=1)
        end_date = _date.today() + _td(days=10)
        with patch(
            "app.approval.service.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(return_value={
                "instance_code": "feishu_instance_borrow",
                "status": "APPROVED",
                "form": [
                    {
                        "name": "DateInterval",
                        "value": (
                            f'{{"start":"{start_date}","end":"{end_date}",'
                            f'"duration":9}}'
                        ),
                    },
                    {"name": "Requirement Scenario", "value": "客户演示"},
                ],
            }),
        ):
            response = await client.post(
                "/api/v1/approval/feishu/callback",
                headers=_fresh_feishu_headers(),
                json={
                    "token": _TEST_FEISHU_VERIFICATION_TOKEN,
                    "header": {"event_type": "approval_instance"},
                    "event": {"instance_code": "feishu_instance_borrow"},
                },
            )
        assert response.status_code == 200

        await db_session.refresh(asset)
        assert asset.status == "borrowed"
        # PR #164 (2026-05-04): borrow now writes the borrower to
        # ``borrowed_by`` and PRESERVES ``assigned_to`` as the original
        # owner (NULL for borrow-from-idle stock).
        assert asset.borrowed_by == employee.id
        assert asset.assigned_to is None
        op = (await db_session.execute(
            select(AssetOperation).where(
                AssetOperation.asset_id == asset.id,
                AssetOperation.operation_type == "borrow",
            )
        )).scalar_one()
        assert op.to_user_id == employee.id
        assert (op.detail or {}).get("borrow_date") == start_date.isoformat()
        assert (op.detail or {}).get("expected_return_date") == end_date.isoformat()

    @pytest.mark.asyncio
    async def test_feishu_callback_auto_executes_asset_return(
        self,
        client: AsyncClient,
        employee_user,
        db_session: AsyncSession,
    ):
        """Feishu approval of an asset_return request must auto-return
        the asset to stock (status=idle, assigned_to=None)."""
        employee, _ = employee_user
        asset = AssetItem(
            asset_tag="RETURN-TEST-01",
            category="laptop",
            status="assigned",
            assigned_to=employee.id,
            custom_fields={},
        )
        db_session.add(asset)
        await db_session.flush()

        request_row = AssetEmployeeRequest(
            request_number="REQ-RETURN-001",
            request_type="asset_return",
            status="pending",
            requester_id=employee.id,
            asset_id=asset.id,
            reason="离职归还",
            detail={"feishu": {"instance_code": "feishu_instance_return"}},
        )
        db_session.add(request_row)
        await db_session.flush()
        db_session.add(ApprovalRequest(
            external_id="feishu_instance_return",
            approval_type="asset_return",
            applicant_id=employee.id,
            status="pending",
            source_payload={
                "employee_request_id": str(request_row.id),
                "asset_id": str(asset.id),
            },
        ))
        await db_session.commit()

        with patch(
            "app.approval.service.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(return_value={
                "instance_code": "feishu_instance_return",
                "status": "APPROVED",
                "form": [
                    {"name": "离职日期", "value": "2026-04-15"},
                    {"name": "归还方式", "value": "现场归还"},
                    {"name": "资产编号", "value": "RETURN-TEST-01"},
                    {"name": "资产序列号", "value": "SN-RET-01"},
                ],
            }),
        ):
            response = await client.post(
                "/api/v1/approval/feishu/callback",
                headers=_fresh_feishu_headers(),
                json={
                    "token": _TEST_FEISHU_VERIFICATION_TOKEN,
                    "header": {"event_type": "approval_instance"},
                    "event": {"instance_code": "feishu_instance_return"},
                },
            )
        assert response.status_code == 200

        await db_session.refresh(asset)
        assert asset.status == "idle"
        assert asset.assigned_to is None
        op = (await db_session.execute(
            select(AssetOperation).where(
                AssetOperation.asset_id == asset.id,
                AssetOperation.operation_type == "return",
            )
        )).scalar_one()
        assert op.from_user_id == employee.id
        assert (op.detail or {}).get("return_date") == "2026-04-15"

    @pytest.mark.asyncio
    async def test_buyback_auto_execute_idempotent_when_already_disposed(
        self,
        client: AsyncClient,
        employee_user,
        db_session: AsyncSession,
    ):
        """If the asset is already disposed (e.g. callback replay), the
        buyback auto-dispose must skip cleanly without raising."""
        from datetime import UTC, datetime
        employee, _ = employee_user
        asset = AssetItem(
            asset_tag="BUYBACK-REPLAY",
            category="laptop",
            status="disposed",
            disposal_type="sold",
            disposed_at=datetime.now(UTC),
            custom_fields={},
        )
        db_session.add(asset)
        await db_session.flush()

        request_row = AssetEmployeeRequest(
            request_number="REQ-REPLAY",
            request_type="offboarding_buyback",
            status="pending",
            requester_id=employee.id,
            asset_id=asset.id,
            reason="离职回购",
            detail={"feishu": {"instance_code": "feishu_instance_replay"}},
        )
        db_session.add(request_row)
        await db_session.flush()
        db_session.add(ApprovalRequest(
            external_id="feishu_instance_replay",
            approval_type="offboarding_buyback",
            applicant_id=employee.id,
            status="pending",
            source_payload={
                "employee_request_id": str(request_row.id),
                "asset_id": str(asset.id),
            },
        ))
        await db_session.commit()

        with patch(
            "app.approval.service.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(return_value={
                "instance_code": "feishu_instance_replay",
                "status": "APPROVED",
                "form": [{"name": "Amount", "value": '{"amount":"500","currency":"CNY"}'}],
            }),
        ):
            response = await client.post(
                "/api/v1/approval/feishu/callback",
                headers=_fresh_feishu_headers(),
                json={
                    "token": _TEST_FEISHU_VERIFICATION_TOKEN,
                    "header": {"event_type": "approval_instance"},
                    "event": {"instance_code": "feishu_instance_replay"},
                },
            )
        assert response.status_code == 200

        # No second disposal operation row — we skipped silently.
        disposal_ops = (await db_session.execute(
            select(AssetOperation).where(
                AssetOperation.asset_id == asset.id,
                AssetOperation.operation_type == "disposal",
            )
        )).scalars().all()
        assert len(disposal_ops) == 0

    @pytest.mark.asyncio
    async def test_feishu_v1_callback_accepted_via_body_token(
        self,
        client: AsyncClient,
        employee_user,
        db_session: AsyncSession,
    ):
        """Feishu v1 events (e.g. ``approval_instance`` v1.0) don't send an
        X-Lark-Signature header — they authenticate via the verification token
        embedded in the payload body. The router must accept this path even
        when ``event_encrypt_key`` is configured."""
        employee, _ = employee_user
        db_session.add(
            ApprovalRequest(
                external_id="feishu_v1_instance",
                approval_type="laptop_replacement",
                applicant_id=employee.id,
                status="pending",
                source_payload={},
            )
        )
        await db_session.commit()

        # Full config — the service builds a FeishuApprovalClient even
        # on the webhook path (for get_instance_detail), which requires
        # app_id + app_secret even though the outbound HTTP call is mocked.
        fake_config = {
            "app_id": "cli_test",
            "app_secret": "secret_test",
            "event_encrypt_key": "test-encrypt-key",
            "event_verification_token": "v1-token",
        }

        with (
            patch(
                "app.approval.router.get_feishu_config",
                new=AsyncMock(return_value=fake_config),
            ),
            patch(
                "app.approval.service.get_feishu_config",
                new=AsyncMock(return_value=fake_config),
            ),
            patch(
                "app.approval.service.FeishuApprovalClient.get_instance_detail",
                new=AsyncMock(
                    return_value={
                        "instance_code": "feishu_v1_instance",
                        "status": "APPROVED",
                        "form": [],
                    }
                ),
            ),
        ):
            response = await client.post(
                "/api/v1/approval/feishu/callback",
                headers=_fresh_feishu_headers(),
                json={
                    "token": "v1-token",
                    "type": "event_callback",
                    "event": {
                        "type": "approval_instance",
                        "instance_code": "feishu_v1_instance",
                        "status": "APPROVED",
                    },
                },
            )

        assert response.status_code == 200, response.text
        assert response.json()["code"] == 0

    @pytest.mark.asyncio
    async def test_feishu_v1_callback_rejected_on_bad_token(
        self,
        client: AsyncClient,
        db_session: AsyncSession,
    ):
        """Wrong token in the body must be rejected when a verification token
        is configured and no signature header is present."""
        fake_config = {
            "event_encrypt_key": "",
            "event_verification_token": "expected-token",
        }

        with (
            patch(
                "app.approval.router.get_feishu_config",
                new=AsyncMock(return_value=fake_config),
            ),
            patch(
                "app.approval.service.get_feishu_config",
                new=AsyncMock(return_value=fake_config),
            ),
        ):
            response = await client.post(
                "/api/v1/approval/feishu/callback",
                headers=_fresh_feishu_headers(),
                json={
                    "token": "wrong-token",
                    "type": "event_callback",
                    "event": {
                        "type": "approval_instance",
                        "instance_code": "whatever",
                    },
                },
            )

        assert response.status_code == 401

    @pytest.mark.asyncio
    async def test_software_seat_callback_activates_allocation(
        self,
        client: AsyncClient,
        employee_user,
        db_session: AsyncSession,
    ):
        """Feishu approval of a software_seat request should flip the
        corresponding SoftwareSeatAllocation from pending → active."""
        employee, _ = employee_user

        software = AssetSoftwareCatalog(name="OPEN AI")
        db_session.add(software)
        await db_session.flush()

        plan = SoftwareLicensePlan(
            software_id=software.id,
            name="Team",
            price=Decimal("20.00"),
            billing_cycle="monthly",
            status="active",
        )
        db_session.add(plan)
        await db_session.flush()

        allocation = SoftwareSeatAllocation(
            plan_id=plan.id,
            user_id=employee.id,
            status="pending",
            reason="管理员必须使用",
        )
        db_session.add(allocation)
        await db_session.flush()

        employee_request = AssetEmployeeRequest(
            request_number="REQ-SEAT-001",
            request_type="software_seat",
            status="pending",
            requester_id=employee.id,
            reason="管理员必须使用",
            detail={
                "feishu": {
                    "instance_code": "feishu_seat_instance",
                    "approval_code": "SEAT_APPROVAL_CODE",
                    "instance_status": "PENDING",
                }
            },
        )
        db_session.add(employee_request)
        await db_session.flush()

        db_session.add(
            ApprovalRequest(
                external_id="feishu_seat_instance",
                approval_type="software_seat",
                applicant_id=employee.id,
                status="pending",
                source_payload={
                    "employee_request_id": str(employee_request.id),
                    "approval_code": "SEAT_APPROVAL_CODE",
                    "seat_allocation_id": str(allocation.id),
                },
            )
        )
        await db_session.commit()

        with patch(
            "app.approval.service.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(
                return_value={
                    "instance_code": "feishu_seat_instance",
                    "status": "APPROVED",
                    "form": [],
                }
            ),
        ):
            response = await client.post(
                "/api/v1/approval/feishu/callback",
                headers=_fresh_feishu_headers(),
                json={
                    "token": _TEST_FEISHU_VERIFICATION_TOKEN,
                    "header": {"event_type": "approval_instance"},
                    "event": {"instance_code": "feishu_seat_instance"},
                },
            )

        assert response.status_code == 200, response.text
        assert response.json()["code"] == 0

        await db_session.refresh(allocation)
        await db_session.refresh(employee_request)
        assert allocation.status == "active"
        assert employee_request.status == "approved"
        assert employee_request.detail["feishu"]["instance_status"] == "APPROVED"