zkxw/backend/tests/test_oa_assignee.py

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

from app.services.oa_assignee import (
    AssigneeResolveError,
    normalize_cc_assignee_types,
    resolve_assignees,
    resolve_cc_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_posts_resolve(self):
        db = MagicMock()
        with patch(
            "app.services.oa_assignee.user_crud.list_user_ids_bound_to_posts",
            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": "posts", "postIds": ["post_gm"]},
                initiator_user_id="zhang",
            )
        self.assertEqual(users, ["gm_user"])
        self.assertEqual(meta.get("resolve_source"), "posts")

    def test_never_reads_sys_role(self):
        db = MagicMock()
        with patch(
            "app.services.oa_assignee.user_crud.list_user_ids_bound_to_posts",
            return_value=["post_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": "posts", "postIds": ["p1"]},
                initiator_user_id="zhang",
            )
        self.assertEqual(users, ["post_only"])
        mock_list.assert_called_once()
        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=[],
            )

    def test_normalize_prefers_assignee_types(self):
        self.assertEqual(
            normalize_cc_assignee_types(
                {"assigneeTypes": ["users", "dept_leader", "users", "bad"], "assigneeType": "posts"}
            ),
            ["users", "dept_leader"],
        )

    def test_normalize_fallback_single(self):
        self.assertEqual(
            normalize_cc_assignee_types({"assigneeType": "initiator_select"}),
            ["initiator_select"],
        )
        self.assertEqual(normalize_cc_assignee_types({}), [])

    def test_resolve_cc_union_dedupe(self):
        db = MagicMock()
        with patch(
            "app.services.oa_assignee.resolve_assignees",
            side_effect=[
                (["u1", "u2"], {"resolve_source": "users"}),
                (["u2", "u3"], {"resolve_source": "dept_leader"}),
            ],
        ) as mock_resolve:
            users, meta = resolve_cc_assignees(
                db,
                tenant_no="zkxw",
                assignee_config={
                    "assigneeTypes": ["users", "dept_leader"],
                    "userIds": ["u1", "u2"],
                },
                initiator_user_id="zhang",
                initiator_dept_id="d1",
            )
        self.assertEqual(users, ["u1", "u2", "u3"])
        self.assertEqual(meta.get("resolve_source"), "multi")
        self.assertEqual(meta.get("sources"), ["users", "dept_leader"])
        self.assertEqual(mock_resolve.call_count, 2)

    def test_resolve_cc_skips_failed_strategy(self):
        db = MagicMock()
        with patch(
            "app.services.oa_assignee.resolve_assignees",
            side_effect=[
                AssigneeResolveError("部门负责人未配置"),
                (["u1"], {"resolve_source": "users"}),
            ],
        ):
            users, meta = resolve_cc_assignees(
                db,
                tenant_no="zkxw",
                assignee_config={
                    "assigneeTypes": ["dept_leader", "users"],
                    "userIds": ["u1"],
                },
                initiator_user_id="zhang",
                initiator_dept_id="d1",
            )
        self.assertEqual(users, ["u1"])
        self.assertEqual(meta.get("resolve_source"), "users")
        self.assertEqual(len(meta.get("skipped") or []), 1)

    def test_resolve_cc_all_fail_returns_empty(self):
        db = MagicMock()
        with patch(
            "app.services.oa_assignee.resolve_assignees",
            side_effect=AssigneeResolveError("fail"),
        ):
            users, meta = resolve_cc_assignees(
                db,
                tenant_no="zkxw",
                assignee_config={"assigneeTypes": ["dept_leader"]},
                initiator_user_id="zhang",
                initiator_dept_id="d1",
            )
        self.assertEqual(users, [])
        self.assertTrue(meta.get("skipped"))

    def test_resolve_cc_legacy_single_type(self):
        db = MagicMock()
        with patch(
            "app.services.oa_assignee.resolve_assignees",
            return_value=(["u1"], {"resolve_source": "users"}),
        ) as mock_resolve:
            users, meta = resolve_cc_assignees(
                db,
                tenant_no="zkxw",
                assignee_config={"assigneeType": "users", "userIds": ["u1"]},
                initiator_user_id="zhang",
            )
        self.assertEqual(users, ["u1"])
        self.assertEqual(meta.get("resolve_source"), "users")
        single_cfg = mock_resolve.call_args.kwargs["assignee_config"]
        self.assertEqual(single_cfg.get("assigneeType"), "users")


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