"""Tests for `POST /api/v1/asset/employee-requests/{id}/supplement-uploads`.

The endpoint lets a requester (or asset admin) attach files to an
in-flight Feishu approval whose upload widgets were left empty at
submission time — the laptop_replacement 回购流程 case where
``转账成功截图`` / ``上传数据清理视频证明`` are no longer required
in the Feishu form but the approver still needs to see them before
deciding. Helix posts the files as a comment-with-attachment on the
Feishu approval instance.
"""
from __future__ import annotations

import json as _json_mod
import uuid
from unittest.mock import AsyncMock, patch

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

from app.domains.asset.models import AssetEmployeeRequest

_FAKE_FEISHU_CFG = {
    "app_id": "cli_test",
    "app_secret": "secret_test",
    "event_verification_token": "test-feishu-token",
}


@pytest.fixture(autouse=True)
def _patch_feishu_config(monkeypatch):
    """Stub Feishu config + tenant token so the client never tries to
    actually hit Feishu in tests. The per-test `add_instance_comment`
    mock then captures the call we want to assert on.

    ``get_instance_detail`` defaults to an empty dict — no pending
    提交人本人 task, so the service takes the comment channel. The
    form-write tests override it with a ``side_effect`` pair
    (pre-approve detail, post-approve detail).
    """
    async def _fake_cfg(**_kwargs):
        return dict(_FAKE_FEISHU_CFG)

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

    async def _fake_token(self):
        return "fake-tenant-token"

    monkeypatch.setattr(
        "app.approval.feishu_client.FeishuApprovalClient.get_tenant_access_token",
        _fake_token,
    )

    async def _fake_instance_detail(self, instance_code):
        return {}

    monkeypatch.setattr(
        "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
        _fake_instance_detail,
    )


def _make_pending_request(
    *,
    requester_id: uuid.UUID,
    snapshot_widgets: list[dict] | None = None,
    status: str = "pending",
) -> AssetEmployeeRequest:
    """Build a laptop_replacement employee_request with a captured
    Feishu form snapshot. The snapshot mirrors what
    ``ApprovalService.create_asset_approval_request`` writes at submit
    time so the supplement-uploads endpoint sees a realistic shape.
    """
    if snapshot_widgets is None:
        snapshot_widgets = [
            {
                "id": "widget_buyback_radio",
                "name": "是否回购",
                "type": "radioV2",
                "required": True,
                "answer": "m28se01b-z05xuezwbv8-0",  # Yes
            },
            {
                "id": "widget_transfer_screenshot",
                "name": "转账成功截图",
                "type": "image",
                "required": False,
                "answer": None,
            },
            {
                "id": "widget_clearance_video",
                "name": "上传数据清理视频证明",
                "type": "attachmentV2",
                "required": False,
                "answer": None,
            },
        ]
    return AssetEmployeeRequest(
        request_number=f"REQ-SUPP-{uuid.uuid4().hex[:8]}",
        request_type="laptop_replacement",
        status=status,
        requester_id=requester_id,
        reason="Battery dead, swapping device",
        detail={
            "feishu": {
                "instance_code": "FAKE-INSTANCE-001",
                "approval_code": "FAKE-LAPTOP-APPROVAL",
                "form_snapshot": snapshot_widgets,
            },
        },
    )


@pytest.mark.asyncio
async def test_supplement_uploads_happy_path_posts_comment_and_updates_snapshot(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    employee_headers,
):
    employee, _ = employee_user
    req = _make_pending_request(requester_id=employee.id)
    db_session.add(req)
    await db_session.commit()

    comment_mock = AsyncMock(return_value={"comment_id": "fake-comment-1"})
    with (
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.add_instance_comment",
            new=comment_mock,
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.resolve_user_id_by_email",
            new=AsyncMock(return_value="fake-feishu-user"),
        ),
    ):
        resp = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=employee_headers,
            json={
                "uploads": {
                    "widget_transfer_screenshot": ["img_key_aaa"],
                    "widget_clearance_video": ["file_token_bbb", "file_token_ccc"],
                },
                "note": "transfer cleared today; cleanup recorded yesterday",
            },
        )

    assert resp.status_code == 200, resp.text

    # Feishu add_instance_comment must have been called with the right
    # instance_code. Bare-token entries carry no viewable url, so they
    # CANNOT ride the v4 comment's url-based ``files`` schema — the
    # comment ships text-only with an honest "not attached" notice
    # (token-shaped attachments are silently dropped by Feishu, which
    # is worse: it reads as delivered when it wasn't).
    assert comment_mock.call_count == 1
    call_kwargs = comment_mock.call_args.kwargs
    assert call_kwargs["instance_code"] == "FAKE-INSTANCE-001"
    assert call_kwargs["user_id"] == "fake-feishu-user"
    assert call_kwargs["attachments"] is None
    assert "申请人补传" in call_kwargs["content_text"]
    assert "未能附带" in call_kwargs["content_text"]

    # form_snapshot answer is appended with the new tokens.
    await db_session.refresh(req)
    feishu = req.detail["feishu"]
    by_id = {w["id"]: w for w in feishu["form_snapshot"]}
    assert by_id["widget_transfer_screenshot"]["answer"] == ["img_key_aaa"]
    assert by_id["widget_clearance_video"]["answer"] == [
        "file_token_bbb",
        "file_token_ccc",
    ]
    # Supplements audit trail recorded.
    assert len(feishu["supplements"]) == 1
    supplement = feishu["supplements"][0]
    assert supplement["actor_email"] == employee.work_email
    assert supplement["note"].startswith("transfer cleared")
    assert set(supplement["uploads"].keys()) == {
        "widget_transfer_screenshot",
        "widget_clearance_video",
    }
    assert supplement["uploads"]["widget_clearance_video"]["count"] == 2


@pytest.mark.asyncio
async def test_supplement_uploads_rejects_non_owner(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    admin_user,
    employee_headers,
):
    """A different non-admin employee cannot supplement someone else's
    request — surfaces as 403 EMPLOYEE_REQUEST_NOT_OWNER. (admin_user
    here just provides a separate identity to OWN the request; the
    caller is ``employee_user`` whose headers we send.)
    """
    other_owner, _ = admin_user  # any non-employee identity will do
    req = _make_pending_request(requester_id=other_owner.id)
    db_session.add(req)
    await db_session.commit()

    with (
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.add_instance_comment",
            new=AsyncMock(),
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.resolve_user_id_by_email",
            new=AsyncMock(return_value="fake-feishu-user"),
        ),
    ):
        resp = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=employee_headers,
            json={
                "uploads": {"widget_transfer_screenshot": ["img_key_aaa"]},
            },
        )

    assert resp.status_code == 403, resp.text
    assert "EMPLOYEE_REQUEST_NOT_OWNER" in resp.text


@pytest.mark.asyncio
async def test_supplement_uploads_rejects_terminal_status(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    employee_headers,
):
    employee, _ = employee_user
    req = _make_pending_request(requester_id=employee.id, status="approved")
    db_session.add(req)
    await db_session.commit()

    with patch(
        "app.approval.feishu_client.FeishuApprovalClient.add_instance_comment",
        new=AsyncMock(),
    ):
        resp = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=employee_headers,
            json={
                "uploads": {"widget_transfer_screenshot": ["img_key_aaa"]},
            },
        )

    assert resp.status_code == 409, resp.text
    assert "EMPLOYEE_REQUEST_INVALID_STATE" in resp.text


@pytest.mark.asyncio
async def test_supplement_uploads_rejects_unknown_widget(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    employee_headers,
):
    employee, _ = employee_user
    req = _make_pending_request(requester_id=employee.id)
    db_session.add(req)
    await db_session.commit()

    with patch(
        "app.approval.feishu_client.FeishuApprovalClient.add_instance_comment",
        new=AsyncMock(),
    ):
        resp = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=employee_headers,
            json={
                "uploads": {"widget_made_up_by_attacker": ["img_key_aaa"]},
            },
        )

    assert resp.status_code == 400, resp.text
    assert "EMPLOYEE_REQUEST_UNKNOWN_WIDGET" in resp.text


@pytest.mark.asyncio
async def test_supplement_uploads_rejects_non_upload_widget(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    employee_headers,
):
    """Targeting the radioV2 widget (是否回购) with a supplement upload
    must 400 — only image / attachmentV2 widgets are eligible. Without
    this guard a client could overwrite a non-upload form answer via
    the supplement path.
    """
    employee, _ = employee_user
    req = _make_pending_request(requester_id=employee.id)
    db_session.add(req)
    await db_session.commit()

    with patch(
        "app.approval.feishu_client.FeishuApprovalClient.add_instance_comment",
        new=AsyncMock(),
    ):
        resp = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=employee_headers,
            json={
                "uploads": {"widget_buyback_radio": ["m28se01b-z05xuezwbv8-1"]},
            },
        )

    assert resp.status_code == 400, resp.text
    assert "EMPLOYEE_REQUEST_NOT_UPLOAD_WIDGET" in resp.text


@pytest.mark.asyncio
async def test_supplement_uploads_admin_can_act_on_other_request(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    admin_user,
    admin_headers,
):
    """Asset admins can supplement on behalf of any requester — same
    auth model as ``cancel_employee_request``. The comment shows up
    under the admin's Feishu identity (their email is what we hand to
    ``resolve_user_id_by_email``), which is honest about who actually
    attached the files.
    """
    employee, _ = employee_user
    req = _make_pending_request(requester_id=employee.id)
    db_session.add(req)
    await db_session.commit()

    comment_mock = AsyncMock(return_value={"comment_id": "fake-comment-2"})
    with (
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.add_instance_comment",
            new=comment_mock,
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.resolve_user_id_by_email",
            new=AsyncMock(return_value="admin-feishu-user"),
        ),
    ):
        resp = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=admin_headers,
            json={
                "uploads": {"widget_transfer_screenshot": ["admin_uploaded_img"]},
            },
        )

    assert resp.status_code == 200, resp.text
    assert comment_mock.call_args.kwargs["user_id"] == "admin-feishu-user"

    await db_session.refresh(req)
    admin_person, _ = admin_user
    supplement = req.detail["feishu"]["supplements"][0]
    assert supplement["actor_email"] == admin_person.work_email


@pytest.mark.asyncio
async def test_supplement_uploads_legacy_row_falls_back_to_instance_form(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    employee_headers,
):
    """Legacy rows (pre-2026-05-12) have no captured ``form_snapshot``
    but the frontend still renders the supplement panel from
    ``latest_instance_detail.form`` — the backend must validate against
    the same source instead of 409ing (the REQ-20260512-1A65 bug). The
    synthesized snapshot is persisted so the row self-heals.
    """
    import json as _json

    employee, _ = employee_user
    instance_form = _json.dumps([
        {"id": "widget_clearance_video", "name": "上传数据清理视频", "type": "image"},
        {"id": "widget_reason", "name": "原因", "type": "input"},
    ])
    req = AssetEmployeeRequest(
        request_number=f"REQ-SUPP-{uuid.uuid4().hex[:8]}",
        request_type="laptop_replacement",
        status="pending",
        requester_id=employee.id,
        reason="legacy row without captured snapshot",
        detail={
            "feishu": {
                "instance_code": "FAKE-INSTANCE-LEGACY",
                "approval_code": "FAKE-LAPTOP-APPROVAL",
                # no form_snapshot key at all — legacy shape
                "latest_instance_detail": {"form": instance_form},
            },
        },
    )
    db_session.add(req)
    await db_session.commit()

    comment_mock = AsyncMock(return_value={"comment_id": "fake-comment-3"})
    with (
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.add_instance_comment",
            new=comment_mock,
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.resolve_user_id_by_email",
            new=AsyncMock(return_value="fake-feishu-user"),
        ),
    ):
        resp = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=employee_headers,
            json={
                "uploads": {"widget_clearance_video": ["img_key_legacy"]},
            },
        )

    assert resp.status_code == 200, resp.text
    assert comment_mock.call_count == 1

    # The synthesized snapshot is backfilled with the new answer so the
    # next open / supplement takes the captured path directly.
    await db_session.refresh(req)
    by_id = {w["id"]: w for w in req.detail["feishu"]["form_snapshot"]}
    assert by_id["widget_clearance_video"]["answer"] == ["img_key_legacy"]
    assert by_id["widget_reason"]["type"] == "input"


@pytest.mark.asyncio
async def test_supplement_uploads_legacy_row_without_instance_form_still_409s(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    employee_headers,
):
    """No snapshot AND no instance form ⇒ there is nothing to validate
    widget ids against; keep the hard 409.
    """
    employee, _ = employee_user
    req = AssetEmployeeRequest(
        request_number=f"REQ-SUPP-{uuid.uuid4().hex[:8]}",
        request_type="laptop_replacement",
        status="pending",
        requester_id=employee.id,
        reason="legacy row, no feishu form at all",
        detail={
            "feishu": {
                "instance_code": "FAKE-INSTANCE-BARE",
                "approval_code": "FAKE-LAPTOP-APPROVAL",
            },
        },
    )
    db_session.add(req)
    await db_session.commit()

    with patch(
        "app.approval.feishu_client.FeishuApprovalClient.add_instance_comment",
        new=AsyncMock(),
    ):
        resp = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=employee_headers,
            json={"uploads": {"widget_clearance_video": ["img_key_x"]}},
        )

    assert resp.status_code == 409, resp.text
    assert "EMPLOYEE_REQUEST_NO_FORM_SNAPSHOT" in resp.text


@pytest.mark.asyncio
async def test_supplement_uploads_rich_entries_route_videos_and_persist_urls(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    employee_headers,
):
    """Rich ``{token, name, url, object_key, size}`` entries: the comment
    attachments must use the documented v4 URL-based ``files`` schema
    (``{url, fileSize, title, type}`` — verified against prod 2026-06-11:
    token-shaped attachments and the ``images`` key are silently
    dropped by Feishu), with ``type`` classified from the filename;
    full metadata must persist in the supplement history so the files
    stay openable from Helix afterwards.
    """
    employee, _ = employee_user
    req = _make_pending_request(requester_id=employee.id)
    db_session.add(req)
    await db_session.commit()

    comment_mock = AsyncMock(return_value={"comment_id": "fake-comment-4"})
    with (
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.add_instance_comment",
            new=comment_mock,
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.resolve_user_id_by_email",
            new=AsyncMock(return_value="fake-feishu-user"),
        ),
    ):
        resp = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=employee_headers,
            json={
                "uploads": {
                    "widget_transfer_screenshot": [
                        {
                            "token": "tok_video",
                            "name": "WangFan电脑初始化视频.mp4",
                            "url": "https://feishu.example/stream/video",
                            "object_key": "approval_supplement/2026/06/11/u/aa-video.mp4",
                            "size": 31457280,
                        },
                        {
                            "token": "tok_photo",
                            "name": "电脑SN.jpg",
                            "url": "https://feishu.example/stream/photo",
                            "object_key": "approval_supplement/2026/06/11/u/bb-sn.jpg",
                            "size": 2048000,
                        },
                    ],
                },
            },
        )

    assert resp.status_code == 200, resp.text
    call_kwargs = comment_mock.call_args.kwargs
    # Documented v4 comment files schema — URL-based, type classified
    # from filename (.mp4 → attachment, .jpg → image).
    assert call_kwargs["attachments"] == [
        {
            "url": "https://feishu.example/stream/video",
            "title": "WangFan电脑初始化视频.mp4",
            "type": "attachment",
            "fileSize": 31457280,
        },
        {
            "url": "https://feishu.example/stream/photo",
            "title": "电脑SN.jpg",
            "type": "image",
            "fileSize": 2048000,
        },
    ]
    assert "未能附带" not in call_kwargs["content_text"]

    await db_session.refresh(req)
    feishu = req.detail["feishu"]
    # Snapshot mirror keeps bare tokens (display counting only).
    by_id = {w["id"]: w for w in feishu["form_snapshot"]}
    assert by_id["widget_transfer_screenshot"]["answer"] == [
        "tok_video", "tok_photo",
    ]
    # History keeps the full per-file metadata: the ephemeral Feishu
    # url AND the durable Helix S3 object_key the presign endpoint
    # serves after the Feishu signature lapses.
    files = feishu["supplements"][0]["uploads"]["widget_transfer_screenshot"]["files"]
    assert files == [
        {
            "token": "tok_video",
            "name": "WangFan电脑初始化视频.mp4",
            "url": "https://feishu.example/stream/video",
            "object_key": "approval_supplement/2026/06/11/u/aa-video.mp4",
            "size": 31457280,
        },
        {
            "token": "tok_photo",
            "name": "电脑SN.jpg",
            "url": "https://feishu.example/stream/photo",
            "object_key": "approval_supplement/2026/06/11/u/bb-sn.jpg",
            "size": 2048000,
        },
    ]


@pytest.mark.asyncio
async def test_supplement_file_presign_authz_and_idor(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    employee_headers,
):
    """The presign endpoint must only mint URLs for object_keys that
    actually belong to the request's supplement history (no IDOR), and
    only for the requester / asset-read roles.
    """
    employee, _ = employee_user
    req = _make_pending_request(requester_id=employee.id)
    detail = dict(req.detail)
    detail["feishu"] = {
        **detail["feishu"],
        "supplements": [{
            "actor_id": str(employee.id),
            "at": "2026-06-11T05:00:00+00:00",
            "uploads": {
                "widget_transfer_screenshot": {
                    "name": "转账成功截图", "type": "image", "count": 1,
                    "files": [{
                        "token": "tok_x",
                        "name": "a.jpg",
                        "url": None,
                        "object_key": "approval_supplement/2026/06/11/u/cc-a.jpg",
                        "size": 1,
                    }],
                },
            },
        }],
    }
    req.detail = detail
    db_session.add(req)
    await db_session.commit()

    base = f"/api/v1/assets/employee-requests/{req.id}/supplement-files/presign"

    # A key that is NOT in this request's history → 403, even for the owner.
    resp = await client.get(
        base, headers=employee_headers,
        params={"object_key": "approval_supplement/2026/06/11/u/zz-other.jpg"},
    )
    assert resp.status_code == 403, resp.text
    assert "SUPPLEMENT_FILE_UNKNOWN" in resp.text

    # The legit key → presigned URL for the owner (S3 presign is a
    # local signature; storage config rows are seeded so no AWS call
    # happens).
    presign_mock = AsyncMock(return_value="https://s3.example/signed")
    with patch(
        "app.core.storage.presign_get",
        new=presign_mock,
    ):
        resp = await client.get(
            base, headers=employee_headers,
            params={"object_key": "approval_supplement/2026/06/11/u/cc-a.jpg"},
        )
    assert resp.status_code == 200, resp.text
    assert resp.json()["data"]["url"] == "https://s3.example/signed"


# ─────────────────────────────────────────────────────────────────────
# Form-write path: flow parked on the requester's own 提交人本人 node
# (upload widget marked editable in the template) ⇒ supplement writes
# the files into the form widget via tasks/approve + form and completes
# the node; comment only carries what didn't stick.
# ─────────────────────────────────────────────────────────────────────

def _live_detail(
    *,
    video_values: list | None = None,
    screenshot_values: list | None = None,
    task_user: str = "fake-feishu-user",
    task_status: str = "PENDING",
) -> dict:
    """Shape of GET /approval/v4/instances/{code} the form-write path
    consumes: ``task_list`` to locate the requester's pending node and
    ``form`` (JSON string) for the per-widget value counts.
    """
    return {
        "approval_code": "FAKE-LAPTOP-APPROVAL",
        "task_list": [
            {
                "id": "task-self-1",
                "status": task_status,
                "user_id": task_user,
                "node_name": "上传旧电脑恢复出厂设置",
            },
        ],
        "form": _json_mod.dumps([
            {
                "id": "widget_transfer_screenshot",
                "type": "image",
                "value": screenshot_values or [],
            },
            {
                "id": "widget_clearance_video",
                "type": "attachmentV2",
                "value": video_values or [],
            },
        ]),
    }


@pytest.mark.asyncio
async def test_supplement_form_write_at_own_node_writes_widget_and_completes_node(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    employee_headers,
):
    employee, _ = employee_user
    req = _make_pending_request(requester_id=employee.id)
    db_session.add(req)
    await db_session.commit()

    pre = _live_detail(video_values=[])
    post = _live_detail(video_values=["file_token_bbb", "file_token_ccc"])
    detail_mock = AsyncMock(side_effect=[pre, post])
    approve_mock = AsyncMock(return_value={})
    comment_mock = AsyncMock(return_value={"comment_id": "should-not-fire"})
    with (
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
            new=detail_mock,
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.approve_task",
            new=approve_mock,
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.add_instance_comment",
            new=comment_mock,
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.resolve_user_id_by_email",
            new=AsyncMock(return_value="fake-feishu-user"),
        ),
    ):
        resp = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=employee_headers,
            json={
                "uploads": {
                    "widget_clearance_video": [
                        "file_token_bbb", "file_token_ccc",
                    ],
                },
                "note": "video recorded today",
            },
        )

    assert resp.status_code == 200, resp.text

    # tasks/approve fired as the requester on their pending node, with
    # the upload codes as the widget's form value.
    assert approve_mock.call_count == 1
    kwargs = approve_mock.call_args.kwargs
    assert kwargs["approval_code"] == "FAKE-LAPTOP-APPROVAL"
    assert kwargs["instance_code"] == "FAKE-INSTANCE-001"
    assert kwargs["user_id"] == "fake-feishu-user"
    assert kwargs["task_id"] == "task-self-1"
    assert _json_mod.loads(kwargs["form"]) == [
        {
            "id": "widget_clearance_video",
            "type": "attachmentV2",
            "value": ["file_token_bbb", "file_token_ccc"],
        },
    ]
    assert "video recorded today" in kwargs["comment"]

    # Verified write ⇒ no comment fallback.
    assert comment_mock.call_count == 0

    await db_session.refresh(req)
    feishu = req.detail["feishu"]
    supplement = feishu["supplements"][0]
    assert supplement["delivery"] == "form"
    assert supplement["completed_task"] == {
        "task_id": "task-self-1",
        "node_name": "上传旧电脑恢复出厂设置",
    }
    assert (
        supplement["uploads"]["widget_clearance_video"]["delivery"] == "form"
    )
    # Post-approve instance snapshot persisted so the progress view
    # reflects the completed node + filled widget without a refetch.
    assert feishu["latest_instance_detail"] == post


@pytest.mark.asyncio
async def test_supplement_form_write_merges_existing_answer_codes(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    employee_headers,
):
    """approve+form may REPLACE the widget value, so codes already in
    the snapshot answer (submit-time uploads / earlier supplements)
    must ride along; http(s) display URLs from synthesized snapshots
    must NOT (Feishu rejects URLs as widget codes).
    """
    employee, _ = employee_user
    req = _make_pending_request(
        requester_id=employee.id,
        snapshot_widgets=[
            {
                "id": "widget_transfer_screenshot",
                "name": "转账成功截图",
                "type": "image",
                "required": False,
                "answer": ["OLD-CODE-1", "https://feishu.example/old-display"],
            },
        ],
    )
    db_session.add(req)
    await db_session.commit()

    pre = _live_detail(screenshot_values=["existing"])
    post = _live_detail(screenshot_values=["existing", "new"])
    approve_mock = AsyncMock(return_value={})
    with (
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(side_effect=[pre, post]),
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.approve_task",
            new=approve_mock,
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.add_instance_comment",
            new=AsyncMock(),
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.resolve_user_id_by_email",
            new=AsyncMock(return_value="fake-feishu-user"),
        ),
    ):
        resp = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=employee_headers,
            json={"uploads": {"widget_transfer_screenshot": ["img_key_new"]}},
        )

    assert resp.status_code == 200, resp.text
    form = _json_mod.loads(approve_mock.call_args.kwargs["form"])
    assert form == [
        {
            "id": "widget_transfer_screenshot",
            "type": "image",
            "value": ["OLD-CODE-1", "img_key_new"],
        },
    ]


@pytest.mark.asyncio
async def test_supplement_form_write_unverified_widget_falls_back_to_comment(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    employee_headers,
):
    """Two widgets supplemented at the requester's node, but only the
    video widget's value grew after approve (the screenshot widget is
    not editable at this node ⇒ Feishu silently ignored it). The
    screenshot files must fall back to the comment channel; delivery
    is recorded as ``mixed``.
    """
    employee, _ = employee_user
    req = _make_pending_request(requester_id=employee.id)
    db_session.add(req)
    await db_session.commit()

    pre = _live_detail()
    post = _live_detail(video_values=["file_token_bbb"])  # screenshot didn't grow
    comment_mock = AsyncMock(return_value={"comment_id": "fallback-1"})
    with (
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(side_effect=[pre, post]),
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.approve_task",
            new=AsyncMock(return_value={}),
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.add_instance_comment",
            new=comment_mock,
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.resolve_user_id_by_email",
            new=AsyncMock(return_value="fake-feishu-user"),
        ),
    ):
        resp = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=employee_headers,
            json={
                "uploads": {
                    "widget_clearance_video": ["file_token_bbb"],
                    "widget_transfer_screenshot": ["img_key_aaa"],
                },
            },
        )

    assert resp.status_code == 200, resp.text
    # Comment fallback carries ONLY the widget the form write missed.
    assert comment_mock.call_count == 1
    text = comment_mock.call_args.kwargs["content_text"]
    assert "转账成功截图" in text
    assert "上传数据清理视频证明" not in text

    await db_session.refresh(req)
    supplement = req.detail["feishu"]["supplements"][0]
    assert supplement["delivery"] == "mixed"
    assert supplement["uploads"]["widget_clearance_video"]["delivery"] == "form"
    assert (
        supplement["uploads"]["widget_transfer_screenshot"]["delivery"]
        == "comment"
    )


@pytest.mark.asyncio
async def test_supplement_form_write_approve_failure_falls_back_to_comment(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    employee_headers,
):
    employee, _ = employee_user
    req = _make_pending_request(requester_id=employee.id)
    db_session.add(req)
    await db_session.commit()

    comment_mock = AsyncMock(return_value={"comment_id": "fallback-2"})
    with (
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(return_value=_live_detail()),
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.approve_task",
            new=AsyncMock(side_effect=RuntimeError("feishu 500")),
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.add_instance_comment",
            new=comment_mock,
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.resolve_user_id_by_email",
            new=AsyncMock(return_value="fake-feishu-user"),
        ),
    ):
        resp = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=employee_headers,
            json={"uploads": {"widget_clearance_video": ["file_token_bbb"]}},
        )

    assert resp.status_code == 200, resp.text
    assert comment_mock.call_count == 1

    await db_session.refresh(req)
    supplement = req.detail["feishu"]["supplements"][0]
    assert supplement["delivery"] == "comment"
    assert supplement["completed_task"] is None


@pytest.mark.asyncio
async def test_supplement_admin_actor_never_takes_form_path(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    admin_user,
    admin_headers,
):
    """An admin supplementing on behalf must NOT approve a node in the
    requester's name — even if the flow is parked on the requester's
    node, the admin path goes straight to the comment channel.
    """
    employee, _ = employee_user
    req = _make_pending_request(requester_id=employee.id)
    db_session.add(req)
    await db_session.commit()

    detail_mock = AsyncMock(return_value=_live_detail())
    approve_mock = AsyncMock(return_value={})
    comment_mock = AsyncMock(return_value={"comment_id": "admin-1"})
    with (
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
            new=detail_mock,
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.approve_task",
            new=approve_mock,
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.add_instance_comment",
            new=comment_mock,
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.resolve_user_id_by_email",
            new=AsyncMock(return_value="admin-feishu-user"),
        ),
    ):
        resp = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=admin_headers,
            json={"uploads": {"widget_transfer_screenshot": ["img_key_aaa"]}},
        )

    assert resp.status_code == 200, resp.text
    assert detail_mock.call_count == 0
    assert approve_mock.call_count == 0
    assert comment_mock.call_count == 1


# ─────────── post-merge audit 2026-06-12 regressions ───────────


def test_find_pending_self_task_bails_on_ambiguous_nodes():
    """``approve_task`` ADVANCES whichever pending node we pick, so when
    the requester operates more than one pending node at once we must
    NOT guess — picking a genuine approval-decision node would silently
    self-approve. Ambiguous → None → comment-only fallback.
    """
    from app.domains.asset.service import AssetService

    one = {
        "task_list": [
            {"id": "t1", "status": "PENDING", "user_id": "u-me"},
            {"id": "t2", "status": "PENDING", "user_id": "u-other"},
        ]
    }
    # Exactly one self-task → returned.
    assert AssetService._find_pending_self_task(one, "u-me")["id"] == "t1"

    two_self = {
        "task_list": [
            {"id": "t1", "status": "PENDING", "user_id": "u-me"},
            {"id": "t2", "status": "PENDING", "user_id": "u-me"},
        ]
    }
    # Two pending self-tasks → ambiguous → None.
    assert AssetService._find_pending_self_task(two_self, "u-me") is None

    # Completed self-tasks don't count toward the ambiguity tally.
    one_done = {
        "task_list": [
            {"id": "t1", "status": "PENDING", "user_id": "u-me"},
            {"id": "t2", "status": "APPROVED", "user_id": "u-me"},
        ]
    }
    assert AssetService._find_pending_self_task(one_done, "u-me")["id"] == "t1"


@pytest.mark.asyncio
async def test_supplement_replay_is_idempotent_no_duplicate_comment(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_user,
    employee_headers,
):
    """Re-submitting the identical upload tokens (double-click / retry)
    is a no-op: no second Feishu comment, no doubled snapshot answer,
    no extra supplement-history entry.
    """
    employee, _ = employee_user
    req = _make_pending_request(requester_id=employee.id)
    db_session.add(req)
    await db_session.commit()

    comment_mock = AsyncMock(return_value={"comment_id": "fake-comment-1"})
    payload = {
        "uploads": {
            "widget_transfer_screenshot": ["img_key_aaa"],
            "widget_clearance_video": ["file_token_bbb"],
        },
        "note": "first submit",
    }
    with (
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.add_instance_comment",
            new=comment_mock,
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.resolve_user_id_by_email",
            new=AsyncMock(return_value="fake-feishu-user"),
        ),
    ):
        r1 = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=employee_headers,
            json=payload,
        )
        # Exact replay of the same tokens.
        r2 = await client.post(
            f"/api/v1/assets/employee-requests/{req.id}/supplement-uploads",
            headers=employee_headers,
            json=payload,
        )

    assert r1.status_code == 200, r1.text
    assert r2.status_code == 200, r2.text
    # The comment was posted exactly once — the replay short-circuited.
    assert comment_mock.call_count == 1

    await db_session.refresh(req)
    feishu = req.detail["feishu"]
    # Snapshot answers not doubled.
    by_id = {w["id"]: w for w in feishu["form_snapshot"]}
    assert by_id["widget_transfer_screenshot"]["answer"] == ["img_key_aaa"]
    assert by_id["widget_clearance_video"]["answer"] == ["file_token_bbb"]
    # Only one supplement-history entry.
    assert len(feishu["supplements"]) == 1