zkxw/backend/tests/test_oa_wechat_notify_hook.py

"""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"]