zkxw/backend/tests/test_oa_wechat_notify_hook.py

"""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"] == "请假申请"


def test_schedule_oa_task_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
        from app.services.oa_wechat_notify import schedule_oa_task_notifies

        schedule_oa_task_notifies(["t1"])
        th.assert_not_called()


def test_schedule_oa_applicant_progress_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
        from app.services.oa_wechat_notify import schedule_oa_applicant_progress

        schedule_oa_applicant_progress(
            instance_id="i1", actor_user_id="u1", action="approve"
        )
        th.assert_not_called()


def test_notify_oa_tasks_sync_noop_when_disabled():
    with patch("app.services.oa_wechat_notify.settings") as st, patch(
        "app.services.oa_wechat_notify.SessionLocal"
    ) as session_local:
        st.WX_MP_NOTIFY_ENABLED = False
        from app.services.oa_wechat_notify import notify_oa_tasks_sync

        notify_oa_tasks_sync(["t1"])
        session_local.assert_not_called()