132 lines
3.9 KiB
Python
132 lines
3.9 KiB
Python
"""申请人进度通知（每步审批结果）单测。"""
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app.core.config import settings
|
|
from app.services.oa_wechat_notify import (
|
|
RESULT_FAIL,
|
|
RESULT_OTHER,
|
|
RESULT_PASS,
|
|
RESULT_TRANSFER,
|
|
_send_applicant_progress,
|
|
map_oa_action_to_result,
|
|
)
|
|
|
|
|
|
def test_map_oa_action_to_result():
|
|
assert map_oa_action_to_result("approve") == RESULT_PASS
|
|
assert map_oa_action_to_result("reject") == RESULT_FAIL
|
|
assert map_oa_action_to_result("transfer") == RESULT_TRANSFER
|
|
assert map_oa_action_to_result("return") == RESULT_OTHER
|
|
assert map_oa_action_to_result("add_sign") == RESULT_OTHER
|
|
assert map_oa_action_to_result("unknown") == RESULT_OTHER
|
|
|
|
|
|
def test_send_applicant_progress_skips_without_mp():
|
|
db = MagicMock()
|
|
inst = MagicMock()
|
|
inst.id = "i1"
|
|
inst.app_id = "app1"
|
|
inst.title = "请假"
|
|
inst.initiator_user_id = "u-ini"
|
|
inst.del_flag = 0
|
|
|
|
ini = MagicMock()
|
|
ini.nick_name = "张三"
|
|
ini.wx_mp_openid = ""
|
|
|
|
app = MagicMock()
|
|
app.app_name = "请假申请"
|
|
|
|
actor = MagicMock()
|
|
actor.nick_name = "李四"
|
|
|
|
def query_side_effect(model):
|
|
q = MagicMock()
|
|
name = getattr(model, "__name__", "") or str(model)
|
|
if "OaInstance" in name:
|
|
q.filter.return_value.first.return_value = inst
|
|
elif "OaApp" in name:
|
|
q.filter.return_value.first.return_value = app
|
|
elif "SysUser" in name:
|
|
# initiator then actor
|
|
q.filter.return_value.first.side_effect = [ini, actor]
|
|
else:
|
|
q.filter.return_value.first.return_value = None
|
|
return q
|
|
|
|
db.query.side_effect = query_side_effect
|
|
|
|
with patch(
|
|
"app.services.oa_wechat_notify.send_oa_applicant_progress_notify"
|
|
) as send:
|
|
_send_applicant_progress(
|
|
db,
|
|
instance_id="i1",
|
|
actor_user_id="u-actor",
|
|
result=RESULT_PASS,
|
|
task_id="t1",
|
|
)
|
|
send.assert_not_called()
|
|
|
|
|
|
def test_send_applicant_progress_payload_args():
|
|
db = MagicMock()
|
|
inst = MagicMock()
|
|
inst.id = "i1"
|
|
inst.app_id = "app1"
|
|
inst.title = "请假"
|
|
inst.initiator_user_id = "u-ini"
|
|
|
|
ini = MagicMock()
|
|
ini.nick_name = "张三"
|
|
ini.wx_mp_openid = "mp-ini"
|
|
|
|
app = MagicMock()
|
|
app.app_name = "请假申请"
|
|
|
|
actor = MagicMock()
|
|
actor.nick_name = "李四"
|
|
|
|
user_queue = [ini, actor]
|
|
|
|
def query_side_effect(model):
|
|
q = MagicMock()
|
|
name = getattr(model, "__name__", "") or str(model)
|
|
if "OaInstance" in name:
|
|
q.filter.return_value.first.return_value = inst
|
|
elif "OaApp" in name:
|
|
q.filter.return_value.first.return_value = app
|
|
elif "SysUser" in name:
|
|
q.filter.return_value.first.side_effect = lambda: user_queue.pop(0)
|
|
else:
|
|
q.filter.return_value.first.return_value = None
|
|
return q
|
|
|
|
db.query.side_effect = query_side_effect
|
|
|
|
with patch(
|
|
"app.services.oa_wechat_notify.send_oa_applicant_progress_notify"
|
|
) as send:
|
|
_send_applicant_progress(
|
|
db,
|
|
instance_id="i1",
|
|
actor_user_id="u-actor",
|
|
result=RESULT_PASS,
|
|
task_id="t1",
|
|
)
|
|
send.assert_called_once()
|
|
kwargs = send.call_args.kwargs
|
|
assert kwargs["mp_openid"] == "mp-ini"
|
|
assert kwargs["business_name"] == "张三 - 请假申请"
|
|
assert kwargs["process_name"] == "请假申请"
|
|
assert kwargs["approver_name"] == "李四"
|
|
assert kwargs["result"] == RESULT_PASS
|
|
assert "mode=view" in kwargs["pagepath"]
|
|
assert "instanceId=i1" in kwargs["pagepath"]
|
|
assert (
|
|
"%2Fpages%2Foa%2Fmine%2Findex" in kwargs["pagepath"]
|
|
or "/pages/oa/mine/index" in kwargs["pagepath"]
|
|
)
|
|
# 我发起的查看：不强制带 taskId
|
|
assert "taskId=0" not in kwargs["pagepath"]
|