zkxw/backend/tests/test_wechat_mp_message.py

from unittest.mock import MagicMock, patch

from app.core.config import settings


def test_send_approval_requires_mp_openid():
    from app.services.wechat_mp_message import send_oa_approval_notify

    r = send_oa_approval_notify(
        mp_openid="",
        pagepath="pages/oa/webview/index?instanceId=i1&taskId=t1&mode=approve",
        title="请假",
        applicant="张三",
        node_name="主管审批",
        time_text="2026-08-13 12:00",
    )
    assert r.get("success") is False
    assert r.get("error") == "missing_mp_openid"


def test_send_approval_requires_template_id():
    from app.services.wechat_mp_message import send_oa_approval_notify

    with patch.object(settings, "WX_MP_TEMPLATE_APPROVAL", ""):
        r = send_oa_approval_notify(
            mp_openid="mp_openid_1",
            pagepath="pages/oa/webview/index?instanceId=i1&taskId=t1&mode=approve",
            title="请假",
            applicant="张三",
            node_name="主管审批",
            time_text="2026-08-13 12:00",
        )
    assert r.get("success") is False
    assert r.get("error") == "missing_template_id"


@patch("app.services.wechat_mp_message.get_mp_access_token", return_value="tok_mp")
@patch("app.services.wechat_mp_message.httpx.Client")
def test_send_approval_success_payload_shape(mock_client_cls, _mock_token):
    from app.services.wechat_mp_message import send_oa_approval_notify

    mock_resp = MagicMock()
    mock_resp.raise_for_status = MagicMock()
    mock_resp.json.return_value = {"errcode": 0, "msgid": 1001}
    mock_client = MagicMock()
    mock_client.__enter__.return_value = mock_client
    mock_client.__exit__.return_value = False
    mock_client.post.return_value = mock_resp
    mock_client_cls.return_value = mock_client

    long_title = "这是一个非常非常长的审批标题需要截断处理一下才行"

    with patch.object(settings, "WX_MP_TEMPLATE_APPROVAL", "tpl_approval"), patch.object(
        settings, "WXPAY_MINI_APPID", "wx_mini_appid"
    ), patch.object(settings, "WX_MP_TPL_FIELD_TITLE", "thing12"), patch.object(
        settings, "WX_MP_TPL_FIELD_APPLICANT", "thing21"
    ), patch.object(settings, "WX_MP_TPL_FIELD_NODE", "thing7"), patch.object(
        settings, "WX_MP_TPL_FIELD_TIME", "time3"
    ), patch.object(settings, "WX_MP_TPL_FIELD_PROCESS", "thing48"):
        r = send_oa_approval_notify(
            mp_openid="mp_openid_1",
            pagepath="/pages/oa/webview/index?instanceId=i1&taskId=t1&mode=approve",
            title=long_title,
            applicant="张三",
            node_name="主管审批",
            time_text="2026-08-13 12:00",
            process_name="请假申请",
        )

    assert r == {"success": True, "msgid": 1001}
    assert mock_client.post.call_count == 1
    args, kwargs = mock_client.post.call_args
    assert args[0].startswith(
        "https://api.weixin.qq.com/cgi-bin/message/template/send?access_token="
    )
    payload = kwargs["json"]
    assert payload["touser"] == "mp_openid_1"
    assert payload["template_id"] == "tpl_approval"
    assert payload["miniprogram"] == {
        "appid": "wx_mini_appid",
        "pagepath": "pages/oa/webview/index?instanceId=i1&taskId=t1&mode=approve",
    }
    assert payload["data"]["thing12"]["value"] == long_title[:20]
    assert payload["data"]["thing21"]["value"] == "张三"
    assert payload["data"]["thing7"]["value"] == "主管审批"
    assert payload["data"]["time3"]["value"] == "2026-08-13 12:00"
    assert payload["data"]["thing48"]["value"] == "请假申请"


@patch("app.services.wechat_mp_message.get_mp_access_token", return_value="tok_mp")
@patch("app.services.wechat_mp_message.httpx.Client")
def test_send_approval_skips_blank_field_keys(mock_client_cls, _mock_token):
    """临时模板仅申请人+时间时，TITLE/NODE 留空不进 data。"""
    from app.services.wechat_mp_message import send_oa_approval_notify

    mock_resp = MagicMock()
    mock_resp.raise_for_status = MagicMock()
    mock_resp.json.return_value = {"errcode": 0, "msgid": 3003}
    mock_client = MagicMock()
    mock_client.__enter__.return_value = mock_client
    mock_client.__exit__.return_value = False
    mock_client.post.return_value = mock_resp
    mock_client_cls.return_value = mock_client

    with patch.object(settings, "WX_MP_TEMPLATE_APPROVAL", "tpl_tmp"), patch.object(
        settings, "WXPAY_MINI_APPID", "wx_mini_appid"
    ), patch.object(settings, "WX_MP_TPL_FIELD_TITLE", ""), patch.object(
        settings, "WX_MP_TPL_FIELD_APPLICANT", "thing1"
    ), patch.object(settings, "WX_MP_TPL_FIELD_NODE", ""), patch.object(
        settings, "WX_MP_TPL_FIELD_TIME", "time2"
    ), patch.object(settings, "WX_MP_TPL_FIELD_PROCESS", ""):
        r = send_oa_approval_notify(
            mp_openid="mp_openid_1",
            pagepath="pages/oa/webview/index?instanceId=i1&taskId=t1&mode=approve",
            title="请假",
            applicant="张三",
            node_name="主管审批",
            time_text="2026-08-13 12:00",
        )

    assert r.get("success") is True
    data = mock_client.post.call_args.kwargs["json"]["data"]
    assert data == {
        "thing1": {"value": "张三"},
        "time2": {"value": "2026-08-13 12:00"},
    }


@patch("app.services.wechat_mp_message.get_mp_access_token", return_value="tok_mp")
@patch("app.services.wechat_mp_message.httpx.Client")
def test_send_cc_success_payload_shape(mock_client_cls, _mock_token):
    from app.services.wechat_mp_message import send_oa_cc_notify

    mock_resp = MagicMock()
    mock_resp.raise_for_status = MagicMock()
    mock_resp.json.return_value = {"errcode": 0, "msgid": 2002}
    mock_client = MagicMock()
    mock_client.__enter__.return_value = mock_client
    mock_client.__exit__.return_value = False
    mock_client.post.return_value = mock_resp
    mock_client_cls.return_value = mock_client

    with patch.object(settings, "WX_MP_TEMPLATE_CC", "tpl_cc"), patch.object(
        settings, "WXPAY_MINI_APPID", "wx_mini_appid"
    ), patch.object(settings, "WX_MP_TPL_CC_FIELD_TITLE", "thing5"), patch.object(
        settings, "WX_MP_TPL_CC_FIELD_APPLICANT", "thing2"
    ), patch.object(settings, "WX_MP_TPL_CC_FIELD_TIME", "time3"), patch.object(
        settings, "WX_MP_TPL_CC_FIELD_PROCESS", "thing1"
    ):
        r = send_oa_cc_notify(
            mp_openid="mp_openid_2",
            pagepath="pages/oa/webview/index?instanceId=i2&taskId=t2&mode=view",
            title="李四 - 报销申请",
            applicant="李四",
            node_name="抄送知会",
            time_text="2026-08-13 15:30",
            process_name="报销申请",
        )

    assert r == {"success": True, "msgid": 2002}
    payload = mock_client.post.call_args.kwargs["json"]
    assert payload["touser"] == "mp_openid_2"
    assert payload["template_id"] == "tpl_cc"
    assert payload["miniprogram"]["appid"] == "wx_mini_appid"
    assert payload["miniprogram"]["pagepath"].endswith("mode=view")
    assert payload["data"] == {
        "thing5": {"value": "李四 - 报销申请"},
        "thing2": {"value": "李四"},
        "time3": {"value": "2026-08-13 15:30"},
        "thing1": {"value": "报销申请"},
    }


@patch("app.services.wechat_mp_message.get_mp_access_token", return_value="tok_mp")
@patch("app.services.wechat_mp_message.httpx.Client")
def test_send_applicant_progress_payload_shape(mock_client_cls, _mock_token):
    from app.services.wechat_mp_message import send_oa_applicant_progress_notify

    mock_resp = MagicMock()
    mock_resp.raise_for_status = MagicMock()
    mock_resp.json.return_value = {"errcode": 0, "msgid": 5005}
    mock_client = MagicMock()
    mock_client.__enter__.return_value = mock_client
    mock_client.__exit__.return_value = False
    mock_client.post.return_value = mock_resp
    mock_client_cls.return_value = mock_client

    with patch.object(
        settings, "WX_MP_TEMPLATE_APPLICANT_PROGRESS", "tpl_progress"
    ), patch.object(settings, "WXPAY_MINI_APPID", "wx_mini_appid"), patch.object(
        settings, "WX_MP_TPL_PROGRESS_FIELD_BUSINESS", "thing5"
    ), patch.object(
        settings, "WX_MP_TPL_PROGRESS_FIELD_PROCESS", "thing1"
    ), patch.object(
        settings, "WX_MP_TPL_PROGRESS_FIELD_APPROVER", "thing6"
    ), patch.object(
        settings, "WX_MP_TPL_PROGRESS_FIELD_RESULT", "const4"
    ), patch.object(settings, "WX_MP_TPL_PROGRESS_FIELD_TIME", "time11"):
        r = send_oa_applicant_progress_notify(
            mp_openid="mp_ini",
            pagepath="pages/oa/webview/index?instanceId=i1&taskId=t1&mode=view",
            business_name="张三 - 请假申请",
            process_name="请假申请",
            approver_name="李四",
            result="通过",
            time_text="2026-08-14 14:30",
        )

    assert r == {"success": True, "msgid": 5005}
    payload = mock_client.post.call_args.kwargs["json"]
    assert payload["template_id"] == "tpl_progress"
    assert payload["data"] == {
        "thing5": {"value": "张三 - 请假申请"},
        "thing1": {"value": "请假申请"},
        "thing6": {"value": "李四"},
        "const4": {"value": "通过"},
        "time11": {"value": "2026-08-14 14:30"},
    }


@patch("app.services.wechat_mp_message.get_mp_access_token", return_value="tok_mp")
@patch("app.services.wechat_mp_message.httpx.Client")
def test_send_approval_retries_without_miniprogram_on_40165(
    mock_client_cls, _mock_token
):
    from app.services.wechat_mp_message import send_oa_approval_notify

    mock_resp_fail = MagicMock()
    mock_resp_fail.raise_for_status = MagicMock()
    mock_resp_fail.json.return_value = {
        "errcode": 40165,
        "errmsg": "invalid weapp pagepath",
    }
    mock_resp_ok = MagicMock()
    mock_resp_ok.raise_for_status = MagicMock()
    mock_resp_ok.json.return_value = {"errcode": 0, "msgid": 4004}
    mock_client = MagicMock()
    mock_client.__enter__.return_value = mock_client
    mock_client.__exit__.return_value = False
    mock_client.post.side_effect = [mock_resp_fail, mock_resp_ok]
    mock_client_cls.return_value = mock_client

    with patch.object(settings, "WX_MP_TEMPLATE_APPROVAL", "tpl_approval"), patch.object(
        settings, "WXPAY_MINI_APPID", "wx_mini_appid"
    ), patch.object(settings, "WX_MP_TPL_FIELD_TITLE", ""), patch.object(
        settings, "WX_MP_TPL_FIELD_APPLICANT", "thing1"
    ), patch.object(settings, "WX_MP_TPL_FIELD_NODE", ""), patch.object(
        settings, "WX_MP_TPL_FIELD_TIME", "time2"
    ), patch.object(settings, "WX_MP_TPL_FIELD_PROCESS", ""):
        r = send_oa_approval_notify(
            mp_openid="mp_openid_1",
            pagepath="pages/oa/webview/index?instanceId=i1&taskId=t1&mode=approve",
            title="请假",
            applicant="张三",
            node_name="主管审批",
            time_text="2026-08-13 12:00",
        )

    assert r == {"success": True, "msgid": 4004}
    assert mock_client.post.call_count == 2
    first = mock_client.post.call_args_list[0].kwargs["json"]
    second = mock_client.post.call_args_list[1].kwargs["json"]
    assert "miniprogram" in first
    assert "miniprogram" not in second