zkxw/backend/tests/test_oa_initiator_select_by_node.py

"""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()