159 lines
4.9 KiB
Python
159 lines
4.9 KiB
Python
"""OA 微信服务号通知钩子单测。"""
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app.services.oa_wechat_notify import (
|
|
_build_matter_title,
|
|
_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_build_pagepath_three_notify_targets():
|
|
"""三种通知：审批待办 / 抄送查看 / 我发起查看，mode + returnPath 不同。"""
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
from app.services.oa_wechat_notify import (
|
|
RETURN_CC,
|
|
RETURN_MINE,
|
|
RETURN_PENDING,
|
|
)
|
|
|
|
appr = build_oa_detail_pagepath(
|
|
instance_id="i1",
|
|
task_id="t1",
|
|
mode="approve",
|
|
return_path=RETURN_PENDING,
|
|
)
|
|
q1 = parse_qs(urlparse("x://" + appr).query)
|
|
assert q1["mode"] == ["approve"]
|
|
assert q1["taskId"] == ["t1"]
|
|
assert q1["returnPath"] == [RETURN_PENDING]
|
|
|
|
cc = build_oa_detail_pagepath(
|
|
instance_id="i1",
|
|
task_id="t-cc",
|
|
mode="view",
|
|
return_path=RETURN_CC,
|
|
)
|
|
q2 = parse_qs(urlparse("x://" + cc).query)
|
|
assert q2["mode"] == ["view"]
|
|
assert q2["taskId"] == ["t-cc"]
|
|
assert q2["returnPath"] == [RETURN_CC]
|
|
|
|
mine = build_oa_detail_pagepath(
|
|
instance_id="i1",
|
|
task_id="",
|
|
mode="view",
|
|
return_path=RETURN_MINE,
|
|
)
|
|
q3 = parse_qs(urlparse("x://" + mine).query)
|
|
assert q3["mode"] == ["view"]
|
|
assert "taskId" not in q3
|
|
assert q3["returnPath"] == [RETURN_MINE]
|
|
|
|
|
|
def test_build_matter_title_applicant_process():
|
|
assert _build_matter_title(
|
|
applicant="李四", process_name="请假申请", fallback="x"
|
|
) == "李四 - 请假申请"
|
|
assert _build_matter_title(applicant="", process_name="请假申请", fallback="x") == (
|
|
"请假申请"
|
|
)
|
|
assert _build_matter_title(applicant="", process_name="", fallback="审批") == "审批"
|
|
|
|
|
|
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"
|
|
inst.app_id = "app-1"
|
|
|
|
ini = MagicMock()
|
|
ini.nick_name = "李四"
|
|
|
|
app = MagicMock()
|
|
app.app_name = "请假申请"
|
|
|
|
# approval + cc 各查 assignee、initiator
|
|
user_results = [user, ini, user, ini]
|
|
|
|
def query_side_effect(model):
|
|
q = MagicMock()
|
|
name = getattr(model, "__name__", "") or str(model)
|
|
if "SysUser" in name:
|
|
q.filter.return_value.first.side_effect = lambda: user_results.pop(0)
|
|
elif "OaApp" in name:
|
|
q.filter.return_value.first.return_value = app
|
|
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"]
|
|
assert "returnPath=" in kwargs["pagepath"]
|
|
assert "%2Fpages%2Foa%2Fpending%2Findex" in kwargs["pagepath"] or (
|
|
"/pages/oa/pending/index" in kwargs["pagepath"]
|
|
)
|
|
assert kwargs["title"] == "李四 - 请假申请"
|
|
assert kwargs["applicant"] == "李四"
|
|
assert kwargs["process_name"] == "请假申请"
|
|
assert kwargs["node_name"] == "主管审批"
|
|
|
|
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"]
|
|
assert (
|
|
"%2Fpages%2Foa%2Fcc%2Findex" in kwargs["pagepath"]
|
|
or "/pages/oa/cc/index" in kwargs["pagepath"]
|
|
)
|
|
assert kwargs["title"] == "李四 - 请假申请"
|
|
assert kwargs["process_name"] == "请假申请"
|