Files
zkxw/backend/tests/test_wechat_mp_message.py
T

122 lines
4.8 KiB
Python

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"] == "报销"