112 lines
3.7 KiB
Python
112 lines
3.7 KiB
Python
"""OA 发起人自选：按节点 map"""
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app.services.oa_engine import (
|
|
OaEngineError,
|
|
_create_tasks_for_node,
|
|
_ids_for_initiator_select_node,
|
|
_normalize_initiator_select_by_node,
|
|
)
|
|
|
|
|
|
class TestNormalizeMap(unittest.TestCase):
|
|
def test_normalize_dedupe_preserve_order(self):
|
|
raw = {"n1": ["u1", "u1", " u2 ", ""], "n2": "bad"}
|
|
out = _normalize_initiator_select_by_node(raw)
|
|
self.assertEqual(out["n1"], ["u1", "u2"])
|
|
self.assertNotIn("n2", out)
|
|
|
|
def test_ids_for_node(self):
|
|
m = {"a1": ["u9"]}
|
|
self.assertEqual(_ids_for_initiator_select_node(m, "a1"), ["u9"])
|
|
self.assertEqual(_ids_for_initiator_select_node(m, "missing"), [])
|
|
|
|
|
|
class TestCreateTasksInitiatorSelect(unittest.TestCase):
|
|
def _inst(self):
|
|
inst = MagicMock()
|
|
inst.id = "inst1"
|
|
inst.tenant_no = "zkxw"
|
|
inst.initiator_user_id = "init"
|
|
inst.initiator_dept_id = "d1"
|
|
inst.current_round = 1
|
|
return inst
|
|
|
|
def test_cc_empty_skips_without_resolve(self):
|
|
db = MagicMock()
|
|
node = {
|
|
"id": "cc1",
|
|
"type": "cc",
|
|
"name": "抄送人",
|
|
"approverConfig": {"assigneeType": "initiator_select"},
|
|
}
|
|
ver = MagicMock()
|
|
ver.advanced_settings_json = "{}"
|
|
with patch(
|
|
"app.services.oa_engine.resolve_cc_assignees",
|
|
return_value=([], {"resolve_source": "", "skipped": [{"type": "initiator_select"}]}),
|
|
) as resolve:
|
|
created = _create_tasks_for_node(
|
|
db,
|
|
inst=self._inst(),
|
|
node=node,
|
|
ver=ver,
|
|
user_id="init",
|
|
initiator_select_by_node={},
|
|
)
|
|
resolve.assert_called_once()
|
|
self.assertEqual(created, [])
|
|
|
|
def test_approval_empty_raises(self):
|
|
db = MagicMock()
|
|
node = {
|
|
"id": "a1",
|
|
"type": "approval",
|
|
"name": "审批人",
|
|
"approverConfig": {"assigneeType": "initiator_select", "signMode": "or"},
|
|
}
|
|
ver = MagicMock()
|
|
ver.advanced_settings_json = "{}"
|
|
with self.assertRaises(OaEngineError):
|
|
_create_tasks_for_node(
|
|
db,
|
|
inst=self._inst(),
|
|
node=node,
|
|
ver=ver,
|
|
user_id="init",
|
|
initiator_select_by_node={"a1": []},
|
|
)
|
|
|
|
def test_approval_uses_node_ids(self):
|
|
db = MagicMock()
|
|
node = {
|
|
"id": "a1",
|
|
"type": "approval",
|
|
"name": "审批人",
|
|
"approverConfig": {"assigneeType": "initiator_select", "signMode": "or"},
|
|
}
|
|
ver = MagicMock()
|
|
ver.advanced_settings_json = "{}"
|
|
with patch(
|
|
"app.services.oa_engine.resolve_assignees",
|
|
return_value=(["u1"], {"resolve_source": "initiator_select"}),
|
|
) as resolve:
|
|
with patch("app.services.oa_engine.generate_uuid", return_value="t1"):
|
|
with patch("app.services.oa_engine._next_task_order_num", return_value=1):
|
|
created = _create_tasks_for_node(
|
|
db,
|
|
inst=self._inst(),
|
|
node=node,
|
|
ver=ver,
|
|
user_id="init",
|
|
initiator_select_by_node={"a1": ["u1"], "other": ["u9"]},
|
|
)
|
|
self.assertEqual(len(created), 1)
|
|
kwargs = resolve.call_args.kwargs
|
|
self.assertEqual(kwargs.get("initiator_select_user_ids"), ["u1"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|