zkxw/backend/tests/test_oa_engine.py

"""OA 引擎单元测试（纯逻辑 + 轻量 mock）"""
import json
import unittest
from datetime import datetime
from unittest.mock import MagicMock, patch

from app.services.oa_engine import (
    OaEngineError,
    _find_node,
    _flow_node_key_order,
    _match_condition_group,
    _sort_tasks_for_detail,
    advance_from,
)


class TestOaEngineHelpers(unittest.TestCase):
    def test_find_node(self):
        root = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "child": {"id": "e1", "type": "end"},
            },
        }
        self.assertEqual(_find_node(root, "a1")["id"], "a1")
        self.assertEqual(_find_node(root, "e1")["id"], "e1")
        self.assertIsNone(_find_node(root, "x"))

    def test_condition_match(self):
        groups = [
            {
                "name": "gt3",
                "when": "{days} > 3",
                "child": {"id": "n1", "type": "end"},
            },
            {"name": "default", "rules": [], "child": {"id": "n2", "type": "end"}},
        ]
        g = _match_condition_group(groups, {"days": 5})
        self.assertEqual(g["name"], "gt3")
        g2 = _match_condition_group(groups, {"days": 1})
        self.assertEqual(g2["name"], "default")

    def test_flow_node_key_order_cc_before_approval(self):
        root = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "c1",
                "type": "cc",
                "child": {
                    "id": "a1",
                    "type": "approval",
                    "child": {"id": "e1", "type": "end"},
                },
            },
        }
        order = _flow_node_key_order(root)
        self.assertLess(order["c1"], order["a1"])

    def test_sort_tasks_by_flow_when_same_create_time(self):
        """抄送先于审批设计时，即使 create_time / order_num 相同也应按树序展示。"""
        root = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "c1",
                "type": "cc",
                "child": {
                    "id": "a1",
                    "type": "approval",
                    "child": {"id": "e1", "type": "end"},
                },
            },
        }
        ts = datetime(2026, 7, 15, 14, 5, 7)
        approval = MagicMock(node_key="a1", order_num=0, create_time=ts, id="AAAA")
        cc = MagicMock(node_key="c1", order_num=0, create_time=ts, id="BBBB")
        # 故意把审批放在列表前面，模拟 create_time 同秒时 PK 乱序
        sorted_tasks = _sort_tasks_for_detail([approval, cc], flow_root=root)
        self.assertEqual([t.node_key for t in sorted_tasks], ["c1", "a1"])


class TestAdvanceFrom(unittest.TestCase):
    def test_linear_creates_approval_then_stops(self):
        flow = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "负责人",
                "approverConfig": {"assigneeType": "dept_leader"},
                "child": {"id": "e1", "type": "end"},
            },
        }
        db = MagicMock()
        inst = MagicMock(
            id="inst1",
            tenant_no="zkxw",
            initiator_user_id="u0",
            initiator_dept_id="d1",
            form_data_json="{}",
            status="running",
        )
        ver = MagicMock(
            flow_tree_json=json.dumps(flow),
            advanced_settings_json=json.dumps({"supervisorFallbackToDeptLeader": True}),
        )
        fake_task = MagicMock()
        with patch(
            "app.services.oa_engine._next_task_order_num", return_value=1
        ), patch(
            "app.services.oa_engine._create_tasks_for_node",
            return_value=[fake_task],
        ) as mock_create:
            advance_from(db, inst=inst, ver=ver, from_node=None, user_id="u0")
            mock_create.assert_called_once()
            self.assertEqual(inst.status, "running")
            self.assertEqual(mock_create.call_args.kwargs.get("order_num_start"), 1)

    def test_complete_path_to_end_via_cc(self):
        flow = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "c1",
                "type": "cc",
                "name": "抄送",
                "approverConfig": {"assigneeType": "users", "userIds": ["u9"]},
                "child": {"id": "e1", "type": "end"},
            },
        }
        db = MagicMock()
        inst = MagicMock(
            id="inst1",
            tenant_no="zkxw",
            initiator_user_id="u0",
            initiator_dept_id="d1",
            form_data_json="{}",
            status="running",
            update_time=datetime.now(),
        )
        ver = MagicMock(
            flow_tree_json=json.dumps(flow),
            advanced_settings_json="{}",
        )
        with patch(
            "app.services.oa_engine._next_task_order_num", return_value=1
        ), patch(
            "app.services.oa_engine._create_tasks_for_node",
            return_value=[MagicMock()],
        ):
            advance_from(db, inst=inst, ver=ver, from_node=None, user_id="u0")
        self.assertEqual(inst.status, "completed")

    def test_empty_assignee_raises(self):
        flow = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "approverConfig": {"assigneeType": "dept_leader"},
                "child": {"id": "e1", "type": "end"},
            },
        }
        db = MagicMock()
        inst = MagicMock(
            id="inst1",
            tenant_no="zkxw",
            initiator_user_id="u0",
            initiator_dept_id="d1",
            form_data_json="{}",
            status="running",
        )
        ver = MagicMock(flow_tree_json=json.dumps(flow), advanced_settings_json="{}")
        with patch(
            "app.services.oa_engine._next_task_order_num", return_value=1
        ), patch("app.services.oa_engine._create_tasks_for_node", return_value=[]):
            with self.assertRaises(OaEngineError):
                advance_from(db, inst=inst, ver=ver, from_node=None, user_id="u0")


class TestAddSign(unittest.TestCase):
    def test_add_sign_rejects_when_disabled(self):
        from app.services.oa_engine import add_sign

        db = MagicMock()
        task = MagicMock(
            id="t1",
            assignee_user_id="u1",
            task_type="approval",
            status="pending",
            instance_id="i1",
            node_key="a1",
            node_name="审批",
            sign_mode="or",
            comment_text=None,
        )
        inst = MagicMock(id="i1", status="running", initiator_dept_id="d1")
        ver = MagicMock(
            id="v1",
            advanced_settings_json=json.dumps({"allowAddSign": False}),
            del_flag=0,
        )
        db.query.return_value.filter.return_value.first.side_effect = [task, inst, ver]
        with self.assertRaises(OaEngineError) as ctx:
            add_sign(
                db,
                "t1",
                tenant_no="zkxw",
                user_id="u1",
                user_ids=["u2"],
            )
        self.assertIn("不允许加签", str(ctx.exception))

    def test_add_sign_requires_users(self):
        from app.services.oa_engine import add_sign

        db = MagicMock()
        task = MagicMock(
            id="t1",
            assignee_user_id="u1",
            task_type="approval",
            status="pending",
            instance_id="i1",
        )
        inst = MagicMock(id="i1", status="running")
        ver = MagicMock(advanced_settings_json="{}", del_flag=0)
        db.query.return_value.filter.return_value.first.side_effect = [task, inst, ver]
        with self.assertRaises(OaEngineError) as ctx:
            add_sign(db, "t1", tenant_no="zkxw", user_id="u1", user_ids=[])
        self.assertIn("加签人", str(ctx.exception))


class TestCompleteTaskOrSign(unittest.TestCase):
    """回归：Session autoflush=False 时，或签取消同节点待办不可覆盖本人已同意状态。"""

    def setUp(self):
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker
        from app.core.database import Base
        from app.models.models import OaAppVersion, OaInstance, OaTask

        self.engine = create_engine("sqlite:///:memory:")
        # 仅建引擎相关表
        OaAppVersion.__table__.create(self.engine)
        OaInstance.__table__.create(self.engine)
        OaTask.__table__.create(self.engine)
        Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
        self.db = Session()
        self.Base = Base
        self.OaAppVersion = OaAppVersion
        self.OaInstance = OaInstance
        self.OaTask = OaTask

        flow = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "审批人",
                "approverConfig": {"signMode": "or"},
                "child": {"id": "e1", "type": "end"},
            },
        }
        now = datetime.now()
        self.db.add(
            OaAppVersion(
                id="v1",
                app_id="app1",
                version_no=1,
                status="published",
                is_latest_published="1",
                form_schema_json="{}",
                flow_tree_json=json.dumps(flow),
                advanced_settings_json="{}",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.add(
            OaInstance(
                id="i1",
                app_id="app1",
                app_version_id="v1",
                title="单人审批",
                initiator_user_id="u0",
                initiator_dept_id="d1",
                form_data_json="{}",
                status="running",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.add(
            OaTask(
                id="t1",
                instance_id="i1",
                node_key="a1",
                node_name="审批人",
                task_type="approval",
                assignee_user_id="u1",
                status="pending",
                sign_mode="or",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.commit()

    def tearDown(self):
        self.db.close()
        self.engine.dispose()

    def test_approve_keeps_approved_not_done(self):
        from app.services.oa_engine import complete_task

        with patch(
            "app.services.oa_engine.get_instance_detail",
            return_value={"id": "i1", "status": "completed"},
        ):
            complete_task(
                self.db,
                "t1",
                tenant_no="zkxw",
                user_id="u1",
                action="approve",
                comment="同意",
            )

        task = self.db.query(self.OaTask).filter(self.OaTask.id == "t1").first()
        self.assertEqual(task.status, "approved")
        self.assertEqual(task.comment_text, "同意")

        # 已处理列表应能查到
        from app.services.oa_engine import list_tasks

        done = list_tasks(self.db, tenant_no="zkxw", user_id="u1", kind="done")
        self.assertEqual(len(done), 1)
        self.assertEqual(done[0]["status"], "approved")

    def test_or_sign_cancels_siblings_only(self):
        from app.services.oa_engine import complete_task

        now = datetime.now()
        self.db.add(
            self.OaTask(
                id="t2",
                instance_id="i1",
                node_key="a1",
                node_name="审批人",
                task_type="approval",
                assignee_user_id="u2",
                status="pending",
                sign_mode="or",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.commit()

        with patch(
            "app.services.oa_engine.get_instance_detail",
            return_value={"id": "i1", "status": "completed"},
        ):
            complete_task(
                self.db,
                "t1",
                tenant_no="zkxw",
                user_id="u1",
                action="approve",
                comment="同意",
            )

        t1 = self.db.query(self.OaTask).filter(self.OaTask.id == "t1").first()
        t2 = self.db.query(self.OaTask).filter(self.OaTask.id == "t2").first()
        self.assertEqual(t1.status, "approved")
        self.assertEqual(t2.status, "done")


class TestBuildProgress(unittest.TestCase):
    def test_cc_before_approval_and_pending_ahead(self):
        from app.services.oa_engine import build_progress_items

        root = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "c1",
                "type": "cc",
                "name": "抄送人",
                "child": {
                    "id": "a1",
                    "type": "approval",
                    "name": "审批人",
                    "child": {
                        "id": "a2",
                        "type": "approval",
                        "name": "总经理",
                        "child": {"id": "e1", "type": "end"},
                    },
                },
            },
        }
        ts = datetime(2026, 7, 15, 14, 5, 7)
        # 审批任务 create 同秒但列表故意审批在前
        t_approval = MagicMock(
            id="tA",
            node_key="a1",
            node_name="审批人",
            task_type="approval",
            status="pending",
            assignee_user_id="u2",
            order_num=0,
            create_time=ts,
            update_time=ts,
            comment_text=None,
            resolve_source="users",
        )
        t_cc = MagicMock(
            id="tC",
            node_key="c1",
            node_name="抄送人",
            task_type="cc",
            status="done",
            assignee_user_id="u9",
            order_num=0,
            create_time=ts,
            update_time=ts,
            comment_text=None,
            resolve_source="users",
        )
        inst = MagicMock(
            status="running",
            initiator_user_id="u0",
            create_time=ts,
            update_time=ts,
        )
        items = build_progress_items(
            inst=inst,
            flow_root=root,
            tasks=[t_approval, t_cc],
            name_map={"u0": "发起人甲", "u2": "审批乙", "u9": "抄送丙"},
            events=[],  # P0 可空
        )
        kinds = [x["kind"] for x in items]
        self.assertEqual(kinds[0], "submit")
        # 已产生：cc 再 approval
        self.assertEqual(
            [x["nodeKey"] for x in items if x["kind"] == "task"],
            ["c1", "a1"],
        )
        # running 时未到达 a2
        pending = [x for x in items if x["kind"] == "pending_node"]
        self.assertTrue(any(p["nodeKey"] == "a2" for p in pending))

    def test_terminal_status_excludes_pending_nodes(self):
        from app.services.oa_engine import build_progress_items

        root = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "审批人",
                "child": {
                    "id": "a2",
                    "type": "approval",
                    "name": "总经理",
                    "child": {"id": "e1", "type": "end"},
                },
            },
        }
        ts = datetime(2026, 7, 15, 14, 5, 7)
        t_approval = MagicMock(
            id="tA",
            node_key="a1",
            node_name="审批人",
            task_type="approval",
            status="rejected",
            assignee_user_id="u2",
            order_num=0,
            create_time=ts,
            update_time=ts,
            comment_text="不同意",
            resolve_source="users",
        )
        for status in ("rejected", "completed", "cancelled"):
            inst = MagicMock(
                status=status,
                initiator_user_id="u0",
                create_time=ts,
                update_time=ts,
            )
            items = build_progress_items(
                inst=inst,
                flow_root=root,
                tasks=[t_approval],
                name_map={"u0": "发起人甲", "u2": "审批乙"},
                events=[],
            )
            pending = [x for x in items if x["kind"] == "pending_node"]
            self.assertEqual(pending, [], f"status={status} should not have pending_node")
            terminal = [x for x in items if x["kind"] == "terminal"]
            self.assertEqual(len(terminal), 1, f"status={status} should have one terminal")


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