zkxw/backend/tests/test_oa_applicant_progress_notify.py

"""申请人进度通知（每步审批结果）单测。"""
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"]