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", "thing1"), patch.object(
        settings, "WX_MP_TPL_FIELD_APPLICANT", "thing2"
    ), patch.object(settings, "WX_MP_TPL_FIELD_NODE", "thing3"), patch.object(
        settings, "WX_MP_TPL_FIELD_TIME", "time4"
    ):
        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",
        )

    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"]["thing1"]["value"] == long_title[:20]
    assert payload["data"]["thing2"]["value"] == "张三"
    assert payload["data"]["thing3"]["value"] == "主管审批"
    assert payload["data"]["time4"]["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"
    ):
        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",
        )

    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 "thing1" in payload["data"]
    assert payload["data"]["thing1"]["value"] == "报销"