137 lines
5.3 KiB
Python
137 lines
5.3 KiB
Python
"""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()
|