zlwuandCursor
5dad1bb0e3
锁定缴费校验失败时整单回滚，避免样本半成功落库；前端禁止删除锁定缴费样本，改码走更正编码。同步落地样本更正、采集微信通知、OA提及未读、细胞产品套餐与相关迁移文档。 Co-authored-by: Cursor <cursoragent@cursor.com>
147 lines
4.5 KiB
Python
147 lines
4.5 KiB
Python
"""OA 评论 @提到 复用抄送模板推送单测。"""
|
|
from unittest.mock import MagicMock, patch
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
from app.services.oa_wechat_notify import (
|
|
RETURN_MENTIONS,
|
|
_build_mention_title,
|
|
_send_mention_one,
|
|
build_oa_detail_pagepath,
|
|
schedule_oa_mention_notifies,
|
|
)
|
|
|
|
|
|
def test_return_mentions_path():
|
|
assert RETURN_MENTIONS == "/pages/oa/mentions/index"
|
|
|
|
|
|
def test_build_mention_title():
|
|
assert _build_mention_title("张三") == "张三 在评论中提到了你"
|
|
assert _build_mention_title("") == "有人在评论中提到了你"
|
|
|
|
|
|
def test_mention_pagepath_mode_view_return_mentions():
|
|
p = build_oa_detail_pagepath(
|
|
instance_id="i1",
|
|
task_id="",
|
|
mode="view",
|
|
return_path=RETURN_MENTIONS,
|
|
)
|
|
q = parse_qs(urlparse("x://" + p).query)
|
|
assert q["mode"] == ["view"]
|
|
assert "taskId" not in q
|
|
assert q["returnPath"] == [RETURN_MENTIONS]
|
|
|
|
|
|
def test_send_mention_one_uses_cc_template_fields():
|
|
db = MagicMock()
|
|
|
|
mentioned = MagicMock()
|
|
mentioned.wx_mp_openid = "mp-m1"
|
|
mentioned.nick_name = "李四"
|
|
|
|
commenter = MagicMock()
|
|
commenter.nick_name = "张三"
|
|
|
|
inst = MagicMock()
|
|
inst.id = "i1"
|
|
inst.app_id = "app1"
|
|
inst.title = "请假"
|
|
inst.del_flag = 0
|
|
|
|
app = MagicMock()
|
|
app.app_name = "请假申请"
|
|
|
|
user_queue = [mentioned, commenter]
|
|
|
|
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_cc_notify") as send:
|
|
_send_mention_one(
|
|
db,
|
|
instance_id="i1",
|
|
mentioned_user_id="u2",
|
|
commenter_user_id="u1",
|
|
)
|
|
send.assert_called_once()
|
|
kwargs = send.call_args.kwargs
|
|
assert kwargs["mp_openid"] == "mp-m1"
|
|
assert kwargs["title"] == "张三 在评论中提到了你"
|
|
assert kwargs["applicant"] == "张三"
|
|
assert kwargs["process_name"] == "请假申请"
|
|
assert "mode=view" in kwargs["pagepath"]
|
|
assert (
|
|
"%2Fpages%2Foa%2Fmentions%2Findex" in kwargs["pagepath"]
|
|
or "/pages/oa/mentions/index" in kwargs["pagepath"]
|
|
)
|
|
|
|
|
|
def test_send_mention_one_skips_without_mp():
|
|
db = MagicMock()
|
|
mentioned = MagicMock()
|
|
mentioned.wx_mp_openid = ""
|
|
mentioned.nick_name = "李四"
|
|
|
|
def query_side_effect(model):
|
|
q = MagicMock()
|
|
name = getattr(model, "__name__", "") or str(model)
|
|
if "SysUser" in name:
|
|
q.filter.return_value.first.return_value = mentioned
|
|
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_cc_notify") as send:
|
|
_send_mention_one(
|
|
db,
|
|
instance_id="i1",
|
|
mentioned_user_id="u2",
|
|
commenter_user_id="u1",
|
|
)
|
|
send.assert_not_called()
|
|
|
|
|
|
def test_schedule_oa_mention_notifies_noop_when_disabled():
|
|
with patch("app.services.oa_wechat_notify.settings") as st, patch(
|
|
"app.services.oa_wechat_notify.threading.Thread"
|
|
) as th:
|
|
st.WX_MP_NOTIFY_ENABLED = False
|
|
schedule_oa_mention_notifies(
|
|
instance_id="i1",
|
|
commenter_user_id="u1",
|
|
mentioned_user_ids=["u2"],
|
|
)
|
|
th.assert_not_called()
|
|
|
|
|
|
def test_schedule_oa_mention_notifies_starts_thread_when_enabled():
|
|
with patch("app.services.oa_wechat_notify.settings") as st, patch(
|
|
"app.services.oa_wechat_notify.threading.Thread"
|
|
) as th:
|
|
st.WX_MP_NOTIFY_ENABLED = True
|
|
schedule_oa_mention_notifies(
|
|
instance_id="i1",
|
|
commenter_user_id="u1",
|
|
mentioned_user_ids=["u2", "u3"],
|
|
)
|
|
th.assert_called_once()
|
|
kwargs = th.call_args.kwargs
|
|
assert kwargs["kwargs"]["instance_id"] == "i1"
|
|
assert kwargs["kwargs"]["commenter_user_id"] == "u1"
|
|
assert kwargs["kwargs"]["mentioned_user_ids"] == ["u2", "u3"]
|