"""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_allows_self(self): """部门经理等场景：直属主管可配置为本人，解析应成功。""" db = MagicMock() initiator = MagicMock(supervisor_user_id="manager_self", dept_id="deptA") db.query.return_value.filter.return_value.first.return_value = initiator 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="manager_self", initiator_dept_id="deptA", ) self.assertEqual(users, ["manager_self"]) self.assertEqual(meta.get("resolve_source"), "direct_supervisor") 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()