zkxw/backend/tests/test_oa_flow_preview.py

"""OA 发起预览：条件字段门闩与单路 walk"""
import unittest
from unittest.mock import MagicMock, patch

from app.services.oa_engine import (
    INITIATOR_CONDITION_FIELD,
    _FLOW_PREVIEW_PENDING_MSG,
    _collect_condition_fields,
    _condition_fields_ready,
    _form_value_present,
    _iter_resolved_path_nodes,
    _walk_preview_nodes,
)


def _cond_flow():
    return {
        "id": "s",
        "type": "start",
        "name": "发起人",
        "child": {
            "id": "a0",
            "type": "approval",
            "name": "一级审批",
            "approverConfig": {"assigneeType": "users", "userIds": ["u1"], "signMode": "or"},
            "child": {
                "id": "c1",
                "type": "condition",
                "conditionGroups": [
                    {
                        "name": "条件1",
                        "rules": [{"field": "days", "op": ">", "value": 3}],
                        "child": {
                            "id": "a1",
                            "type": "approval",
                            "name": "董事长审批",
                            "approverConfig": {
                                "assigneeType": "users",
                                "userIds": ["u2"],
                                "signMode": "or",
                            },
                            "child": {"id": "e1", "type": "end", "name": "结束"},
                        },
                    },
                    {
                        "name": "默认",
                        "isDefault": True,
                        "rules": [],
                        "child": {"id": "e2", "type": "end", "name": "结束"},
                    },
                ],
            },
        },
    }


class TestFormValuePresent(unittest.TestCase):
    def test_empty(self):
        self.assertFalse(_form_value_present({}, "days"))
        self.assertFalse(_form_value_present({"days": None}, "days"))
        self.assertFalse(_form_value_present({"days": ""}, "days"))
        self.assertFalse(_form_value_present({"days": "  "}, "days"))
        self.assertFalse(_form_value_present({"days": []}, "days"))

    def test_present(self):
        self.assertTrue(_form_value_present({"days": 0}, "days"))
        self.assertTrue(_form_value_present({"ok": False}, "ok"))
        self.assertTrue(_form_value_present({"days": 5}, "days"))
        self.assertTrue(_form_value_present({"t": "事假"}, "t"))


class TestCollectConditionFields(unittest.TestCase):
    def test_linear_no_fields(self):
        root = {
            "id": "s",
            "type": "start",
            "child": {"id": "e", "type": "end"},
        }
        self.assertEqual(_collect_condition_fields(root), [])
        self.assertTrue(_condition_fields_ready(root, {}))

    def test_rules_field(self):
        root = _cond_flow()
        self.assertEqual(_collect_condition_fields(root), ["days"])
        self.assertFalse(_condition_fields_ready(root, {}))
        self.assertTrue(_condition_fields_ready(root, {"days": 5}))

    def test_when_formula_fields(self):
        root = {
            "type": "condition",
            "conditionGroups": [
                {"name": "x", "when": "{days} > 3 and {leaveType} == '事假'", "child": {}},
                {"name": "d", "isDefault": True, "rules": []},
            ],
        }
        self.assertEqual(_collect_condition_fields(root), ["days", "leaveType"])

    def test_initiator_system_field_not_gated_by_form_data(self):
        """系统伪字段 __initiator__ 不依赖 formData，空表单也应 resolved 门闩通过。"""
        root = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "c1",
                "type": "condition",
                "conditionGroups": [
                    {
                        "name": "条件1",
                        "rules": [
                            {
                                "field": INITIATOR_CONDITION_FIELD,
                                "op": "contains",
                                "value": ["u1"],
                            }
                        ],
                        "child": {"id": "e1", "type": "end"},
                    },
                    {
                        "name": "默认",
                        "isDefault": True,
                        "rules": [],
                        "child": {"id": "e2", "type": "end"},
                    },
                ],
            },
        }
        self.assertEqual(_collect_condition_fields(root), [])
        self.assertTrue(_condition_fields_ready(root, {}))

    def test_initiator_plus_form_field_still_gates_form(self):
        root = {
            "type": "condition",
            "conditionGroups": [
                {
                    "name": "条件1",
                    "rules": [
                        {
                            "field": INITIATOR_CONDITION_FIELD,
                            "op": "contains",
                            "value": ["u1"],
                        },
                        {"field": "days", "op": ">", "value": 3},
                    ],
                    "child": {},
                },
                {"name": "默认", "isDefault": True, "rules": []},
            ],
        }
        self.assertEqual(_collect_condition_fields(root), ["days"])
        self.assertFalse(_condition_fields_ready(root, {}))
        self.assertTrue(_condition_fields_ready(root, {"days": 5}))


class TestWalkPreviewResolvedPath(unittest.TestCase):
    def test_hit_condition1(self):
        db = MagicMock()
        steps = []
        with patch("app.services.oa_engine.resolve_assignees", return_value=(["u1"], {})), patch(
            "app.services.oa_engine._preview_people",
            side_effect=lambda _db, ids: [{"id": i, "nickName": i} for i in ids],
        ), patch("app.services.oa_engine._assignee_meta", return_value="指定成员"):
            ok = _walk_preview_nodes(
                db,
                node=_cond_flow(),
                tenant_no="t1",
                initiator_user_id="init",
                initiator_dept_id=None,
                advanced={},
                initiator_select_by_node=None,
                form_data={"days": 5},
                steps=steps,
            )
        self.assertTrue(ok)
        types = [s["type"] for s in steps]
        names = [s["name"] for s in steps]
        self.assertNotIn("branch", types)
        self.assertEqual(names, ["发起人", "一级审批", "董事长审批", "结束"])

    def test_hit_default(self):
        db = MagicMock()
        steps = []
        with patch("app.services.oa_engine.resolve_assignees", return_value=(["u1"], {})), patch(
            "app.services.oa_engine._preview_people",
            side_effect=lambda _db, ids: [{"id": i, "nickName": i} for i in ids],
        ), patch("app.services.oa_engine._assignee_meta", return_value="指定成员"):
            ok = _walk_preview_nodes(
                db,
                node=_cond_flow(),
                tenant_no="t1",
                initiator_user_id="init",
                initiator_dept_id=None,
                advanced={},
                initiator_select_by_node=None,
                form_data={"days": 1},
                steps=steps,
            )
        self.assertTrue(ok)
        names = [s["name"] for s in steps]
        self.assertEqual(names, ["发起人", "一级审批", "结束"])
        self.assertNotIn("董事长审批", names)

    def test_pending_message_constant(self):
        self.assertEqual(_FLOW_PREVIEW_PENDING_MSG, "必填信息填写后，流程将自动显示")

    def test_merge_after_condition_skips_branch_end(self):
        """条件分支叶子 end 后若有汇合主链，预览只保留真正的结束节点。"""
        root = {
            "id": "s",
            "type": "start",
            "name": "发起人",
            "child": {
                "id": "c1",
                "type": "condition",
                "conditionGroups": [
                    {
                        "name": "条件1",
                        "rules": [{"field": "days", "op": ">", "value": 3}],
                        "child": {
                            "id": "a1",
                            "type": "approval",
                            "name": "领导会签",
                            "approverConfig": {
                                "assigneeType": "users",
                                "userIds": ["u2"],
                                "signMode": "and",
                            },
                            "child": {"id": "e_branch", "type": "end", "name": "结束"},
                        },
                    },
                    {
                        "name": "默认",
                        "isDefault": True,
                        "rules": [],
                        "child": {"id": "e_def", "type": "end", "name": "结束"},
                    },
                ],
                "child": {
                    "id": "cc1",
                    "type": "cc",
                    "name": "抄送人",
                    "approverConfig": {
                        "assigneeTypes": ["initiator_select"],
                        "assigneeType": "initiator_select",
                    },
                    "child": {"id": "e_final", "type": "end", "name": "结束"},
                },
            },
        }
        db = MagicMock()
        steps = []
        with patch(
            "app.services.oa_engine.resolve_assignees", return_value=(["u2"], {})
        ), patch(
            "app.services.oa_engine.resolve_cc_assignees", return_value=([], {})
        ), patch(
            "app.services.oa_engine._preview_people",
            side_effect=lambda _db, ids: [{"id": i, "nickName": i} for i in ids],
        ), patch(
            "app.services.oa_engine._assignee_meta", return_value="指定成员"
        ), patch(
            "app.services.oa_engine._assignee_meta_for_cc", return_value="发起人自选"
        ):
            ok = _walk_preview_nodes(
                db,
                node=root,
                tenant_no="t1",
                initiator_user_id="init",
                initiator_dept_id=None,
                advanced={},
                initiator_select_by_node=None,
                form_data={"days": 5},
                steps=steps,
            )
        self.assertTrue(ok)
        names = [s["name"] for s in steps]
        types = [s["type"] for s in steps]
        self.assertEqual(names, ["发起人", "领导会签", "抄送人", "结束"])
        self.assertEqual(types.count("end"), 1)
        self.assertEqual(steps[-1]["id"], "e_final")

        path = _iter_resolved_path_nodes(root, {"days": 5}, initiator_user_id="init")
        self.assertIsNotNone(path)
        path_ids = [n.get("id") for n in path]
        self.assertNotIn("e_branch", path_ids)
        self.assertIn("e_final", path_ids)
        self.assertEqual(sum(1 for n in path if n.get("type") == "end"), 1)


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