82 lines
2.5 KiB
Python
82 lines
2.5 KiB
Python
"""OA 微信服务号通知钩子单测。"""
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app.services.oa_wechat_notify import _send_one, build_oa_detail_pagepath
|
|
|
|
|
|
def test_build_pagepath_no_leading_slash():
|
|
p = build_oa_detail_pagepath(instance_id="i1", task_id="t1", mode="approve")
|
|
assert p.startswith("pages/")
|
|
assert "mode=approve" in p
|
|
assert "instanceId=i1" in p
|
|
assert "taskId=t1" in p
|
|
|
|
|
|
def test_send_one_routes_approval_vs_cc():
|
|
db = MagicMock()
|
|
|
|
user = MagicMock()
|
|
user.wx_mp_openid = "mp-openid-1"
|
|
user.nick_name = "张三"
|
|
|
|
inst = MagicMock()
|
|
inst.title = "请假"
|
|
inst.initiator_user_id = "u-ini"
|
|
|
|
ini = MagicMock()
|
|
ini.nick_name = "李四"
|
|
|
|
def query_side_effect(model):
|
|
q = MagicMock()
|
|
name = getattr(model, "__name__", "") or str(model)
|
|
if "SysUser" in name:
|
|
# first call: assignee; second: initiator (approval path may call twice)
|
|
q.filter.return_value.first.side_effect = [user, ini]
|
|
else:
|
|
q.filter.return_value.first.return_value = inst
|
|
return q
|
|
|
|
db.query.side_effect = query_side_effect
|
|
|
|
approval_task = MagicMock()
|
|
approval_task.id = "t-appr"
|
|
approval_task.instance_id = "i1"
|
|
approval_task.assignee_user_id = "u1"
|
|
approval_task.task_type = "approval"
|
|
approval_task.status = "pending"
|
|
approval_task.node_name = "主管审批"
|
|
|
|
with patch(
|
|
"app.services.oa_wechat_notify.send_oa_approval_notify"
|
|
) as send_appr, patch(
|
|
"app.services.oa_wechat_notify.send_oa_cc_notify"
|
|
) as send_cc:
|
|
_send_one(db, approval_task)
|
|
send_appr.assert_called_once()
|
|
send_cc.assert_not_called()
|
|
kwargs = send_appr.call_args.kwargs
|
|
assert kwargs["mp_openid"] == "mp-openid-1"
|
|
assert "mode=approve" in kwargs["pagepath"]
|
|
|
|
# reset query for cc path
|
|
db.query.side_effect = query_side_effect
|
|
|
|
cc_task = MagicMock()
|
|
cc_task.id = "t-cc"
|
|
cc_task.instance_id = "i1"
|
|
cc_task.assignee_user_id = "u1"
|
|
cc_task.task_type = "cc"
|
|
cc_task.status = "done"
|
|
cc_task.node_name = "抄送"
|
|
|
|
with patch(
|
|
"app.services.oa_wechat_notify.send_oa_approval_notify"
|
|
) as send_appr, patch(
|
|
"app.services.oa_wechat_notify.send_oa_cc_notify"
|
|
) as send_cc:
|
|
_send_one(db, cc_task)
|
|
send_cc.assert_called_once()
|
|
send_appr.assert_not_called()
|
|
kwargs = send_cc.call_args.kwargs
|
|
assert "mode=view" in kwargs["pagepath"]
|