265 lines
11 KiB
Python
265 lines
11 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", "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
|