zkxw/backend/tests/test_oa_publish.py

"""OA 发布校验单元测试"""
import unittest
from unittest.mock import MagicMock, patch

from app.services.oa_publish import (
    count_fillable_fields,
    validate_condition_rules_against_form,
    validate_control_types,
    validate_draft_for_publish,
    validate_flow_tree,
)


def _linear_ok_flow():
    return {
        "id": "s1",
        "type": "start",
        "name": "发起人",
        "child": {
            "id": "a1",
            "type": "approval",
            "name": "部门负责人",
            "approverConfig": {"assigneeType": "dept_leader"},
            "child": {"id": "e1", "type": "end", "name": "结束"},
        },
    }


def _condition_flow(rules=None, when=None, is_default=False):
    return {
        "type": "start",
        "name": "发起人",
        "child": {
            "type": "condition",
            "name": "条件",
            "conditionGroups": [
                {
                    "name": "分支1",
                    "rules": rules if rules is not None else [],
                    "when": when,
                    "isDefault": False,
                    "child": {
                        "type": "approval",
                        "name": "审批A",
                        "approverConfig": {"assigneeType": "dept_leader"},
                        "child": {"type": "end", "name": "结束"},
                    },
                },
                {
                    "name": "默认",
                    "rules": [],
                    "isDefault": True,
                    "child": {
                        "type": "approval",
                        "name": "审批B",
                        "approverConfig": {"assigneeType": "dept_leader"},
                        "child": {"type": "end", "name": "结束"},
                    },
                },
            ],
        },
    }


class TestOaPublishValidate(unittest.TestCase):
    def test_count_fillable_fields(self):
        schema = {
            "rule": [
                {"type": "html", "props": {"innerHTML": "说明"}},
                {"type": "input", "field": "title", "title": "标题"},
                {
                    "type": "fcRow",
                    "children": [
                        {"type": "inputNumber", "field": "days", "title": "天数"},
                    ],
                },
            ]
        }
        self.assertEqual(count_fillable_fields(schema), 2)
        self.assertEqual(count_fillable_fields({"rule": []}), 0)

    def test_flow_ok(self):
        self.assertEqual(validate_flow_tree(_linear_ok_flow()), [])

    def test_flow_missing_end(self):
        tree = {
            "type": "start",
            "name": "发起人",
            "child": {
                "type": "approval",
                "name": "审批",
                "approverConfig": {"assigneeType": "users", "userIds": ["u1"]},
            },
        }
        errs = validate_flow_tree(tree)
        self.assertTrue(any("结束" in e or "后续" in e for e in errs))

    def test_approval_users_empty_rejected(self):
        tree = {
            "type": "start",
            "child": {
                "type": "approval",
                "name": "审批",
                "approverConfig": {"assigneeType": "users", "userIds": []},
                "child": {"type": "end"},
            },
        }
        errs = validate_flow_tree(tree)
        self.assertTrue(any("指定成员" in e for e in errs))

    def test_cc_multi_types_ok(self):
        flow = {
            "type": "start",
            "child": {
                "type": "cc",
                "name": "抄送",
                "approverConfig": {
                    "assigneeTypes": ["users", "dept_leader"],
                    "userIds": ["u1"],
                },
                "child": {"type": "end"},
            },
        }
        self.assertEqual(validate_flow_tree(flow), [])

    def test_cc_multi_users_empty_fails(self):
        flow = {
            "type": "start",
            "child": {
                "type": "cc",
                "name": "抄送",
                "approverConfig": {
                    "assigneeTypes": ["users", "initiator_select"],
                    "userIds": [],
                },
                "child": {"type": "end"},
            },
        }
        errs = validate_flow_tree(flow)
        self.assertTrue(any("指定成员" in e for e in errs))

    def test_cc_no_types_fails(self):
        flow = {
            "type": "start",
            "child": {
                "type": "cc",
                "name": "抄送",
                "approverConfig": {},
                "child": {"type": "end"},
            },
        }
        errs = validate_flow_tree(flow)
        self.assertTrue(len(errs) >= 1)

    def test_approval_ignores_assignee_types_array(self):
        flow = {
            "type": "start",
            "child": {
                "type": "approval",
                "name": "审批",
                "approverConfig": {"assigneeTypes": ["dept_leader"]},
                "child": {"type": "end"},
            },
        }
        errs = validate_flow_tree(flow)
        self.assertTrue(len(errs) >= 1)

    def test_cc_legacy_single_assignee_type_ok(self):
        flow = {
            "type": "start",
            "child": {
                "type": "cc",
                "name": "抄送",
                "approverConfig": {"assigneeType": "initiator_select"},
                "child": {"type": "end"},
            },
        }
        self.assertEqual(validate_flow_tree(flow), [])

    def test_condition_branch_incomplete(self):
        tree = {
            "type": "start",
            "child": {
                "type": "condition",
                "name": "条件",
                "conditionGroups": [{"name": "大于3天", "rules": [], "child": None}],
            },
        }
        errs = validate_flow_tree(tree)
        self.assertTrue(any("下游" in e or "结束" in e for e in errs))

    def test_draft_requires_fillable_and_assignee(self):
        db = MagicMock()
        with patch(
            "app.services.oa_publish.validate_posts_referenced",
            return_value=[],
        ):
            errs = validate_draft_for_publish(
                db,
                tenant_no="zkxw",
                form_schema={"rule": [{"type": "html"}]},
                flow_tree={
                    "type": "start",
                    "child": {
                        "type": "approval",
                        "approverConfig": {"assigneeType": "users", "userIds": []},
                        "child": {"type": "end"},
                    },
                },
            )
        self.assertTrue(any("可填字段" in e for e in errs))
        self.assertTrue(any("指定成员" in e for e in errs))

    def test_valid_draft_passes_form_and_flow(self):
        db = MagicMock()
        with patch(
            "app.services.oa_publish.validate_posts_referenced",
            return_value=[],
        ):
            errs = validate_draft_for_publish(
                db,
                tenant_no="zkxw",
                form_schema={"rule": [{"type": "input", "field": "reason"}]},
                flow_tree=_linear_ok_flow(),
            )
        self.assertEqual(errs, [])

    def test_unknown_control_type_rejected(self):
        errs = validate_control_types(
            {"rule": [{"type": "aiMagicBox", "field": "x"}, {"type": "input", "field": "y"}]}
        )
        self.assertTrue(any("aiMagicBox" in e for e in errs))

    def test_upload_is_allowed(self):
        self.assertEqual(
            validate_control_types(
                {"rule": [{"type": "upload", "field": "files", "title": "上传"}]}
            ),
            [],
        )
        self.assertEqual(
            count_fillable_fields(
                {"rule": [{"type": "upload", "field": "files", "title": "上传"}]}
            ),
            1,
        )


class TestConditionPublish(unittest.TestCase):
    def test_when_only_rejected(self):
        errs = validate_flow_tree(_condition_flow(rules=[], when="{days} > 3"))
        self.assertTrue(any("可视化" in e or "重新配置" in e or "rules" in e for e in errs))

    def test_rules_empty_non_default_rejected(self):
        errs = validate_flow_tree(_condition_flow(rules=[]))
        self.assertTrue(any("条件" in e for e in errs))

    def test_rules_ok(self):
        errs = validate_flow_tree(
            _condition_flow(rules=[{"field": "days", "op": ">", "value": 3}])
        )
        self.assertEqual(errs, [])

    def test_invalid_field_reference(self):
        db = MagicMock()
        with patch(
            "app.services.oa_publish.validate_posts_referenced",
            return_value=[],
        ):
            errs = validate_draft_for_publish(
                db,
                tenant_no="zkxw",
                form_schema={
                    "rule": [
                        {
                            "type": "inputNumber",
                            "field": "days",
                            "title": "天数",
                            "useInCondition": True,
                            "$required": True,
                        }
                    ]
                },
                flow_tree=_condition_flow(
                    rules=[{"field": "other", "op": ">", "value": 1}]
                ),
            )
        self.assertTrue(any("无效字段" in e or "other" in e for e in errs))

    def test_field_not_flagged_rejected(self):
        db = MagicMock()
        with patch(
            "app.services.oa_publish.validate_posts_referenced",
            return_value=[],
        ):
            errs = validate_draft_for_publish(
                db,
                tenant_no="zkxw",
                form_schema={
                    "rule": [
                        {
                            "type": "inputNumber",
                            "field": "days",
                            "title": "天数",
                            "useInCondition": False,
                        }
                    ]
                },
                flow_tree=_condition_flow(
                    rules=[{"field": "days", "op": ">", "value": 3}]
                ),
            )
        self.assertTrue(any("无效字段" in e or "参与" in e for e in errs))

    def test_initiator_condition_field_allowed(self):
        errs = validate_condition_rules_against_form(
            {"rule": []},
            _condition_flow(
                rules=[
                    {
                        "field": "__initiator__",
                        "op": "contains",
                        "value": ["u1"],
                    }
                ]
            ),
        )
        self.assertEqual(errs, [])

    def test_initiator_condition_empty_value_rejected(self):
        errs = validate_condition_rules_against_form(
            {"rule": []},
            _condition_flow(
                rules=[
                    {
                        "field": "__initiator__",
                        "op": "contains",
                        "value": [],
                    }
                ]
            ),
        )
        self.assertTrue(any("发起人" in e and ("包含" in e or "用户" in e) for e in errs))

    def test_initiator_condition_bad_op_rejected(self):
        errs = validate_condition_rules_against_form(
            {"rule": []},
            _condition_flow(
                rules=[
                    {
                        "field": "__initiator__",
                        "op": "==",
                        "value": ["u1"],
                    }
                ]
            ),
        )
        self.assertTrue(any("发起人" in e and "包含" in e for e in errs))

    def test_dept_condition_field_allowed(self):
        errs = validate_condition_rules_against_form(
            {
                "rule": [
                    {
                        "type": "dept",
                        "field": "Fdept",
                        "title": "部门",
                        "useInCondition": True,
                        "$required": True,
                        "props": {"multiple": True},
                    }
                ]
            },
            _condition_flow(
                rules=[
                    {"field": "Fdept", "op": "contains", "value": ["d1", "d2"]}
                ]
            ),
        )
        self.assertEqual(errs, [])

    def test_owner_dept_condition_field_allowed(self):
        errs = validate_condition_rules_against_form(
            {
                "rule": [
                    {
                        "type": "ownerDept",
                        "field": "Fod",
                        "title": "归属部门",
                        "useInCondition": True,
                        "$required": True,
                    }
                ]
            },
            _condition_flow(
                rules=[{"field": "Fod", "op": "contains", "value": ["d1"]}]
            ),
        )
        self.assertEqual(errs, [])


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