zkxw/backend/tests/test_oa_flow_preview.py

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

from app.services.oa_engine import (
    _FLOW_PREVIEW_PENDING_MSG,
    _collect_condition_fields,
    _condition_fields_ready,
    _form_value_present,
    _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"])


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, "必填信息填写后，流程将自动显示")


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