zkxw/backend/tests/test_oa_assignee.py

"""OA 审批人解析单元测试"""
import unittest
from unittest.mock import MagicMock, patch

from app.services.oa_assignee import AssigneeResolveError, resolve_assignees


class TestOaAssignee(unittest.TestCase):
    def test_users_resolves_active_ids(self):
        db = MagicMock()
        with patch(
            "app.services.oa_assignee._active_user_id",
            side_effect=lambda _db, uid, _tn: uid if uid else None,
        ):
            users, meta = resolve_assignees(
                db,
                tenant_no="zkxw",
                assignee_config={"assigneeType": "users", "userIds": ["u1", "u1", ""]},
                initiator_user_id="initiator",
            )
        self.assertEqual(users, ["u1"])
        self.assertEqual(meta.get("resolve_source"), "users")

    def test_dept_leader_resolves_by_initiator_dept(self):
        db = MagicMock()
        dept = MagicMock(leader_user_id="leaderA")
        db.query.return_value.filter.return_value.first.return_value = dept
        with patch(
            "app.services.oa_assignee._active_user_id",
            side_effect=lambda _db, uid, _tn: uid if uid else None,
        ):
            users, meta = resolve_assignees(
                db,
                tenant_no="zkxw",
                assignee_config={"assigneeType": "dept_leader"},
                initiator_user_id="zhang",
                initiator_dept_id="deptA",
            )
        self.assertEqual(users, ["leaderA"])
        self.assertEqual(meta.get("resolve_source"), "dept_leader")

    def test_direct_supervisor_fallback_to_dept_leader(self):
        db = MagicMock()
        initiator = MagicMock(supervisor_user_id=None, dept_id="deptA")
        dept = MagicMock(leader_user_id="leaderA")
        db.query.return_value.filter.return_value.first.side_effect = [initiator, dept]
        with patch(
            "app.services.oa_assignee._active_user_id",
            side_effect=lambda _db, uid, _tn: uid if uid else None,
        ):
            users, meta = resolve_assignees(
                db,
                tenant_no="zkxw",
                assignee_config={"assigneeType": "direct_supervisor"},
                initiator_user_id="no_supervisor",
                initiator_dept_id="deptA",
                advanced={"supervisorFallbackToDeptLeader": True},
            )
        self.assertEqual(users, ["leaderA"])
        self.assertEqual(meta.get("resolve_source"), "supervisor_fallback_dept_leader")

    def test_direct_supervisor_no_fallback_raises(self):
        db = MagicMock()
        initiator = MagicMock(supervisor_user_id=None, dept_id="deptA")
        db.query.return_value.filter.return_value.first.return_value = initiator
        with patch(
            "app.services.oa_assignee._active_user_id",
            return_value=None,
        ):
            with self.assertRaises(AssigneeResolveError):
                resolve_assignees(
                    db,
                    tenant_no="zkxw",
                    assignee_config={"assigneeType": "direct_supervisor"},
                    initiator_user_id="no_supervisor",
                    initiator_dept_id="deptA",
                    advanced={"supervisorFallbackToDeptLeader": False},
                )

    def test_oa_role_gm(self):
        db = MagicMock()
        with patch(
            "app.services.oa_assignee.oa_crud.list_oa_role_user_ids",
            return_value=["gm_user", "inactive"],
        ), patch(
            "app.services.oa_assignee._active_user_id",
            side_effect=lambda _db, uid, _tn: uid if uid == "gm_user" else None,
        ):
            users, meta = resolve_assignees(
                db,
                tenant_no="zkxw",
                assignee_config={"assigneeType": "oa_roles", "oaRoleIds": ["gm_role_id"]},
                initiator_user_id="zhang",
            )
        self.assertEqual(users, ["gm_user"])
        self.assertEqual(meta.get("resolve_source"), "oa_roles")

    def test_never_reads_sys_role(self):
        db = MagicMock()
        with patch(
            "app.services.oa_assignee.oa_crud.list_oa_role_user_ids",
            return_value=["oa_only"],
        ) as mock_list, patch(
            "app.services.oa_assignee._active_user_id",
            side_effect=lambda _db, uid, _tn: uid,
        ):
            users, _meta = resolve_assignees(
                db,
                tenant_no="zkxw",
                assignee_config={"assigneeType": "oa_roles", "oaRoleIds": ["r1"]},
                initiator_user_id="zhang",
            )
        self.assertEqual(users, ["oa_only"])
        mock_list.assert_called_once()
        # 业务路径不得查询 sys_role
        for call in db.query.call_args_list:
            args = call[0] if call[0] else ()
            for a in args:
                name = getattr(a, "__name__", None) or getattr(a, "__tablename__", None) or str(a)
                self.assertNotIn("SysRole", str(name))
                self.assertNotIn("sys_role", str(name))

    def test_initiator_select_empty_raises(self):
        db = MagicMock()
        with self.assertRaises(AssigneeResolveError):
            resolve_assignees(
                db,
                tenant_no="zkxw",
                assignee_config={"assigneeType": "initiator_select"},
                initiator_user_id="zhang",
                initiator_select_user_ids=[],
            )


if __name__ == "__main__":
    unittest.main()