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_condition_match_rules_list(self):
        groups = [
            {
                "name": "gt3",
                "rules": [{"field": "days", "op": ">", "value": 3}],
                "child": {"id": "n1", "type": "end"},
            },
            {"name": "default", "isDefault": True, "rules": [], "child": {"id": "n2", "type": "end"}},
        ]
        self.assertEqual(_match_condition_group(groups, {"days": 5})["name"], "gt3")
        self.assertEqual(_match_condition_group(groups, {"days": 1})["name"], "default")

    def test_condition_match_rules_and(self):
        groups = [
            {
                "name": "both",
                "rules": [
                    {"field": "days", "op": ">", "value": 3},
                    {"field": "leaveType", "op": "==", "value": "事假"},
                ],
                "child": {"id": "n1", "type": "end"},
            },
            {"name": "default", "rules": [], "isDefault": True, "child": {"id": "n2", "type": "end"}},
        ]
        self.assertEqual(
            _match_condition_group(groups, {"days": 5, "leaveType": "事假"})["name"],
            "both",
        )
        self.assertEqual(
            _match_condition_group(groups, {"days": 5, "leaveType": "病假"})["name"],
            "default",
        )

    def test_condition_contains_person(self):
        groups = [
            {
                "name": "hit",
                "rules": [
                    {"field": "Fperson", "op": "contains", "value": ["u1", "u2"]}
                ],
                "child": {"id": "n1", "type": "end"},
            },
            {"name": "default", "rules": [], "child": {"id": "n2", "type": "end"}},
        ]
        g = _match_condition_group(
            groups,
            {"Fperson": [{"id": "u2", "name": "李四"}]},
        )
        self.assertEqual(g["name"], "hit")
        g2 = _match_condition_group(
            groups,
            {"Fperson": [{"id": "u9", "name": "其他人"}]},
        )
        self.assertEqual(g2["name"], "default")

    def test_condition_contains_person_single_object(self):
        groups = [
            {
                "name": "hit",
                "rules": [{"field": "Fperson", "op": "contains", "value": ["u1"]}],
                "child": {"id": "n1", "type": "end"},
            },
            {"name": "default", "isDefault": True, "child": {"id": "n2", "type": "end"}},
        ]
        g = _match_condition_group(
            groups, {"Fperson": {"id": "u1", "name": "张三"}}
        )
        self.assertEqual(g["name"], "hit")

    def test_condition_contains_dept(self):
        groups = [
            {
                "name": "hit",
                "rules": [
                    {"field": "Fdept", "op": "contains", "value": ["d1", "d2"]}
                ],
                "child": {"id": "n1", "type": "end"},
            },
            {"name": "default", "rules": [], "child": {"id": "n2", "type": "end"}},
        ]
        g = _match_condition_group(
            groups,
            {"Fdept": [{"id": "d2", "name": "研发"}]},
        )
        self.assertEqual(g["name"], "hit")
        g2 = _match_condition_group(
            groups,
            {"Fdept": [{"id": "d9", "name": "其他"}]},
        )
        self.assertEqual(g2["name"], "default")

    def test_condition_contains_owner_dept_single_object(self):
        groups = [
            {
                "name": "hit",
                "rules": [{"field": "Fod", "op": "contains", "value": ["d1"]}],
                "child": {"id": "n1", "type": "end"},
            },
            {"name": "default", "isDefault": True, "child": {"id": "n2", "type": "end"}},
        ]
        g = _match_condition_group(
            groups, {"Fod": {"id": "d1", "name": "研发"}}
        )
        self.assertEqual(g["name"], "hit")

    def test_condition_contains_initiator(self):
        groups = [
            {
                "name": "hit",
                "rules": [
                    {
                        "field": "__initiator__",
                        "op": "contains",
                        "value": ["u1", "u2"],
                    }
                ],
                "child": {"id": "n1", "type": "end"},
            },
            {"name": "default", "rules": [], "child": {"id": "n2", "type": "end"}},
        ]
        g = _match_condition_group(groups, {}, initiator_user_id="u2")
        self.assertEqual(g["name"], "hit")
        g2 = _match_condition_group(groups, {}, initiator_user_id="u9")
        self.assertEqual(g2["name"], "default")

    def test_condition_initiator_and_form_field(self):
        groups = [
            {
                "name": "both",
                "rules": [
                    {"field": "__initiator__", "op": "contains", "value": ["u1"]},
                    {"field": "days", "op": ">", "value": 3},
                ],
                "child": {"id": "n1", "type": "end"},
            },
            {"name": "default", "isDefault": True, "child": {"id": "n2", "type": "end"}},
        ]
        self.assertEqual(
            _match_condition_group(
                groups, {"days": 5}, initiator_user_id="u1"
            )["name"],
            "both",
        )
        self.assertEqual(
            _match_condition_group(
                groups, {"days": 5}, initiator_user_id="u9"
            )["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 OaApp, OaAppVersion, OaInstance, OaInstanceEvent, OaTask, SysUser

        self.engine = create_engine("sqlite:///:memory:")
        # 仅建引擎相关表
        OaApp.__table__.create(self.engine)
        OaAppVersion.__table__.create(self.engine)
        OaInstance.__table__.create(self.engine)
        OaTask.__table__.create(self.engine)
        OaInstanceEvent.__table__.create(self.engine)
        SysUser.__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")["list"]
        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,
            round_no=1,
            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,
            round_no=1,
            create_time=ts,
            update_time=ts,
            comment_text=None,
            resolve_source="users",
        )
        inst = MagicMock(
            status="running",
            initiator_user_id="u0",
            current_round=1,
            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_running_shows_all_subsequent_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": "a3",
                        "type": "approval",
                        "name": "三级审批",
                        "child": {"id": "e1", "type": "end"},
                    },
                },
            },
        }
        ts = datetime(2026, 7, 17, 11, 0, 0)
        t1 = MagicMock(
            id="t1",
            node_key="a1",
            node_name="一级审批",
            task_type="approval",
            status="pending",
            assignee_user_id="u1",
            order_num=0,
            round_no=1,
            create_time=ts,
            update_time=ts,
            comment_text=None,
            resolve_source="users",
            sign_mode="or",
        )
        inst = MagicMock(
            status="running",
            initiator_user_id="u0",
            current_round=1,
            create_time=ts,
            update_time=ts,
            form_data_json="{}",
            tenant_no="t1",
            initiator_dept_id="",
        )
        items = build_progress_items(
            inst=inst,
            flow_root=root,
            tasks=[t1],
            name_map={"u0": "发起人", "u1": "甲"},
            events=[],
        )
        pending = [x for x in items if x["kind"] == "pending_node"]
        self.assertEqual([p["nodeKey"] for p in pending], ["a2", "a3", "e1"])
        self.assertEqual([p["title"] for p in pending], ["二级审批", "三级审批", "结束"])
        self.assertEqual(pending[-1]["taskType"], "end")

    def test_or_sign_same_node_merged_into_one_card(self):
        """或签同节点三人只投影一张进度卡片，people 含各人状态。"""
        from app.services.oa_engine import build_progress_items

        root = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "部门审批",
                "child": {"id": "e1", "type": "end"},
            },
        }
        ts = datetime(2026, 7, 17, 10, 0, 0)
        ts2 = datetime(2026, 7, 17, 10, 5, 0)

        def mk(tid, uid, status, comment=None, update=None):
            return MagicMock(
                id=tid,
                node_key="a1",
                node_name="部门审批",
                task_type="approval",
                status=status,
                assignee_user_id=uid,
                order_num=0,
                round_no=1,
                create_time=ts,
                update_time=update or ts,
                comment_text=comment,
                resolve_source="users",
                sign_mode="or",
            )

        tasks = [
            mk("t1", "u1", "approved", "同意", ts2),
            mk("t2", "u2", "done"),
            mk("t3", "u3", "done"),
        ]
        inst = MagicMock(
            status="completed",
            initiator_user_id="u0",
            current_round=1,
            create_time=ts,
            update_time=ts2,
        )
        items = build_progress_items(
            inst=inst,
            flow_root=root,
            tasks=tasks,
            name_map={"u0": "发起人", "u1": "甲", "u2": "乙", "u3": "丙"},
            events=[],
        )
        task_items = [x for x in items if x["kind"] == "task"]
        self.assertEqual(len(task_items), 1)
        card = task_items[0]
        self.assertEqual(card["nodeKey"], "a1")
        self.assertEqual(card["status"], "approved")
        self.assertEqual(card["signMode"], "or")
        self.assertEqual(len(card["people"]), 3)
        self.assertEqual(
            [(p["nickName"], p["status"]) for p in card["people"]],
            [("甲", "approved"), ("乙", "done"), ("丙", "done")],
        )
        self.assertIn("甲：同意", card["commentText"])

    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,
            round_no=1,
            create_time=ts,
            update_time=ts,
            comment_text="不同意",
            resolve_source="users",
        )
        for status in ("rejected", "completed", "cancelled", "returned"):
            inst = MagicMock(
                status=status,
                initiator_user_id="u0",
                current_round=1,
                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"]
            # 打回仅展示审批人节点「已打回」，不追加独立终态行
            expect_n = 0 if status == "returned" else 1
            self.assertEqual(
                len(terminal), expect_n, f"status={status} terminal count"
            )

    def test_cancelled_terminal_attributes_to_initiator(self):
        """撤销终态归因发起人；任务 cancelled 仍投影为节点，由前端映射「无需处理」。"""
        from app.services.oa_engine import build_progress_items

        root = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "直接主管",
                "child": {"id": "e1", "type": "end"},
            },
        }
        ts = datetime(2026, 8, 13, 12, 47, 46)
        t_pending_cancelled = MagicMock(
            id="tA",
            node_key="a1",
            node_name="直接主管",
            task_type="approval",
            status="cancelled",
            assignee_user_id="u2",
            order_num=0,
            round_no=1,
            create_time=datetime(2026, 8, 13, 11, 17, 38),
            update_time=ts,
            comment_text=None,
            resolve_source="direct_supervisor",
            sign_mode="and",
        )
        inst = MagicMock(
            status="cancelled",
            initiator_user_id="u0",
            current_round=1,
            create_time=datetime(2026, 8, 13, 11, 17, 38),
            update_time=ts,
        )
        items = build_progress_items(
            inst=inst,
            flow_root=root,
            tasks=[t_pending_cancelled],
            name_map={"u0": "吴子龙", "u2": "张主管"},
            events=[],
        )
        task_card = next(x for x in items if x["kind"] == "task")
        self.assertEqual(task_card["title"], "直接主管")
        self.assertEqual(task_card["status"], "cancelled")
        self.assertEqual(task_card["people"][0]["status"], "cancelled")
        self.assertEqual(task_card["timestamp"], "", "cancelled 节点勿带撤销时间")

        terminal = next(x for x in items if x["kind"] == "terminal")
        self.assertEqual(terminal["status"], "cancelled")
        self.assertEqual(terminal["title"], "已撤销")
        self.assertEqual(terminal["assigneeUserId"], "u0")
        self.assertEqual(terminal["assigneeUserName"], "吴子龙")
        self.assertEqual(terminal["commentText"], "发起人撤销")
        self.assertTrue(terminal["timestamp"])

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

        root = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "审批人",
                "child": {"id": "e1", "type": "end"},
            },
        }
        ts = datetime(2026, 7, 15, 14, 58, 8)
        t_approval = MagicMock(
            id="tA",
            node_key="a1",
            node_name="审批人",
            task_type="approval",
            status="returned",
            assignee_user_id="u2",
            order_num=1,
            round_no=1,
            create_time=ts,
            update_time=ts,
            comment_text="打回",
            resolve_source="users",
        )
        ret_ev = MagicMock(
            event_type="return",
            round_no=1,
            actor_user_id="u2",
            create_time=ts,
            comment_text="打回",
            node_key="a1",
            task_id="tA",
        )
        inst = MagicMock(
            status="returned",
            initiator_user_id="u0",
            current_round=1,
            create_time=ts,
            update_time=ts,
        )
        items = build_progress_items(
            inst=inst,
            flow_root=root,
            tasks=[t_approval],
            name_map={"u0": "发起人", "u2": "审批人"},
            events=[ret_ev],
        )
        self.assertFalse(any(x["kind"] == "return" for x in items))
        tasks = [x for x in items if x["kind"] == "task"]
        self.assertEqual(len(tasks), 1)
        self.assertEqual(tasks[0]["status"], "returned")
        self.assertEqual(tasks[0]["commentText"], "打回")
        terminals = [x for x in items if x["kind"] == "terminal"]
        self.assertEqual(terminals, [], "打回不追加独立终态「已打回」行")


class TestReturnAndResubmit(unittest.TestCase):
    def setUp(self):
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker
        from app.models.models import OaApp, OaAppVersion, OaInstance, OaInstanceEvent, OaTask, SysUser

        self.engine = create_engine("sqlite:///:memory:")
        OaApp.__table__.create(self.engine)
        OaAppVersion.__table__.create(self.engine)
        OaInstance.__table__.create(self.engine)
        OaTask.__table__.create(self.engine)
        OaInstanceEvent.__table__.create(self.engine)
        SysUser.__table__.create(self.engine)
        Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
        self.db = Session()
        self.OaInstance = OaInstance
        self.OaTask = OaTask
        self.OaInstanceEvent = OaInstanceEvent

        flow = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "审批人",
                "approverConfig": {
                    "assigneeType": "users",
                    "userIds": ["u1"],
                    "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=json.dumps({"allowReturn": True}),
                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",
                current_round=1,
                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",
                round_no=1,
                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_return_sets_returned_and_event(self):
        from app.services.oa_engine import return_task

        with patch(
            "app.services.oa_engine.get_instance_detail",
            return_value={"id": "i1", "status": "returned"},
        ):
            return_task(
                self.db, "t1", tenant_no="zkxw", user_id="u1", comment="请补充材料"
            )
        inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
        task = self.db.query(self.OaTask).filter_by(id="t1").first()
        ev = (
            self.db.query(self.OaInstanceEvent)
            .filter_by(instance_id="i1", event_type="return")
            .first()
        )
        self.assertEqual(inst.status, "returned")
        self.assertEqual(task.status, "returned")
        self.assertIsNotNone(ev)
        self.assertEqual(ev.comment_text, "请补充材料")

    def test_resubmit_raises_deprecated(self):
        from app.services.oa_engine import OaEngineError, resubmit_instance, return_task

        with patch(
            "app.services.oa_engine.get_instance_detail",
            return_value={"id": "i1", "status": "returned"},
        ):
            return_task(
                self.db, "t1", tenant_no="zkxw", user_id="u1", comment="请补充材料"
            )

        with self.assertRaises(OaEngineError) as ctx:
            resubmit_instance(
                self.db,
                "i1",
                tenant_no="zkxw",
                user_id="u0",
                form_data={"x": 1},
            )
        self.assertIn("废弃", str(ctx.exception))

    def test_done_list_includes_returned(self):
        from app.services.oa_engine import list_tasks, return_task

        with patch(
            "app.services.oa_engine.get_instance_detail",
            return_value={"id": "i1", "status": "returned"},
        ):
            return_task(
                self.db, "t1", tenant_no="zkxw", user_id="u1", comment="请补充材料"
            )
        done = list_tasks(self.db, tenant_no="zkxw", user_id="u1", kind="done")["list"]
        self.assertEqual(len(done), 1)
        self.assertEqual(done[0]["status"], "returned")


class TestListTasksEnrichment(unittest.TestCase):
    """list_tasks 返回整单此刻进度摘要字段。"""

    def setUp(self):
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker
        from app.models.models import OaApp, OaAppVersion, OaInstance, OaTask, SysUser

        self.engine = create_engine("sqlite:///:memory:")
        OaApp.__table__.create(self.engine)
        OaAppVersion.__table__.create(self.engine)
        OaInstance.__table__.create(self.engine)
        OaTask.__table__.create(self.engine)
        SysUser.__table__.create(self.engine)
        Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
        self.db = Session()
        self.OaTask = OaTask
        now = datetime(2026, 7, 15, 10, 0, 0)
        self.db.add(
            OaApp(
                id="app1",
                app_name="请假审批",
                initiator_scope="all",
                admin_scope="all_oa_admin",
                status="1",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.add(
            OaAppVersion(
                id="v1",
                app_id="app1",
                version_no=1,
                status="published",
                is_latest_published="1",
                form_schema_json="{}",
                flow_tree_json="{}",
                advanced_settings_json="{}",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.add(
            SysUser(
                id="u0",
                user_no="initiator",
                nick_name="发起人甲",
                password="x",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.add(
            SysUser(
                id="u1",
                user_no="approver1",
                nick_name="审批人乙",
                password="x",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.add(
            SysUser(
                id="u2",
                user_no="approver2",
                nick_name="审批人丙",
                password="x",
                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="t-done",
                instance_id="i1",
                node_key="a1",
                node_name="一级审批",
                task_type="approval",
                assignee_user_id="u1",
                status="approved",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.add(
            OaTask(
                id="t-pending",
                instance_id="i1",
                node_key="a2",
                node_name="二级审批",
                task_type="approval",
                assignee_user_id="u2",
                status="pending",
                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_done_list_shows_instance_current_progress(self):
        from app.services.oa_engine import list_tasks

        done = list_tasks(self.db, tenant_no="zkxw", user_id="u1", kind="done")["list"]
        self.assertEqual(len(done), 1)
        row = done[0]
        self.assertEqual(row["appName"], "请假审批")
        self.assertEqual(row["initiatorUserName"], "发起人甲")
        self.assertEqual(row["applyTime"], "2026-07-15 10:00:00")
        self.assertEqual(row["instanceStatus"], "running")
        self.assertEqual(row["currentNodeName"], "二级审批")
        self.assertEqual(row["currentApproverNames"], "审批人丙")
        self.assertEqual(row["status"], "approved")

    def test_mine_list_enrichment(self):
        from app.services.oa_engine import list_tasks

        mine = list_tasks(self.db, tenant_no="zkxw", user_id="u0", kind="mine")["list"]
        self.assertEqual(len(mine), 1)
        row = mine[0]
        self.assertEqual(row["appName"], "请假审批")
        self.assertEqual(row["instanceStatus"], "running")
        self.assertEqual(row["currentNodeName"], "二级审批")
        self.assertEqual(row["currentApproverNames"], "审批人丙")

    def test_pending_list_enrichment(self):
        from app.services.oa_engine import list_tasks

        pending = list_tasks(self.db, tenant_no="zkxw", user_id="u2", kind="pending")["list"]
        self.assertEqual(len(pending), 1)
        row = pending[0]
        self.assertEqual(row["appName"], "请假审批")
        self.assertEqual(row["currentNodeName"], "二级审批")

    def test_terminal_instance_labels(self):
        from app.services.oa_engine import list_tasks
        from app.models.models import OaInstance

        cases = [
            ("completed", "已完成"),
            ("rejected", "已拒绝"),
            ("returned", "已打回"),
            ("cancelled", "已撤销"),
        ]
        inst = self.db.query(OaInstance).filter_by(id="i1").first()
        pending = self.db.query(self.OaTask).filter_by(id="t-pending").first()
        for status, expected_label in cases:
            with self.subTest(status=status):
                inst.status = status
                pending.status = "cancelled"
                self.db.commit()

                mine = list_tasks(self.db, tenant_no="zkxw", user_id="u0", kind="mine")["list"]
                row = mine[0]
                self.assertEqual(row["instanceStatus"], status)
                self.assertEqual(row["currentNodeName"], expected_label)
                self.assertEqual(row["currentApproverNames"], "-")

                inst.status = "running"
                pending.status = "pending"
                self.db.commit()


class TestTransferTask(unittest.TestCase):
    def setUp(self):
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker
        from app.models.models import OaApp, OaAppVersion, OaInstance, OaInstanceEvent, OaTask, SysUser

        self.engine = create_engine("sqlite:///:memory:")
        OaApp.__table__.create(self.engine)
        OaAppVersion.__table__.create(self.engine)
        OaInstance.__table__.create(self.engine)
        OaTask.__table__.create(self.engine)
        OaInstanceEvent.__table__.create(self.engine)
        SysUser.__table__.create(self.engine)
        Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
        self.db = Session()
        self.OaTask = OaTask
        self.OaInstance = OaInstance
        self.OaAppVersion = OaAppVersion
        self.SysUser = SysUser
        now = datetime.now()
        self.db.add(
            OaApp(
                id="app1",
                group_id="",
                app_name="测试应用",
                icon_url="",
                description="",
                initiator_scope="all",
                initiator_user_ids="[]",
                admin_scope="all",
                admin_user_ids="[]",
                status="published",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.add(
            OaAppVersion(
                id="v1",
                app_id="app1",
                version_no=1,
                status="published",
                is_latest_published="1",
                form_schema_json="{}",
                flow_tree_json="{}",
                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="t",
                status="running",
                initiator_user_id="u0",
                initiator_dept_id="d1",
                form_data_json="{}",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
                current_round=1,
            )
        )
        for uid, name in (("u1", "李四"), ("u2", "赵六"), ("u3", "钱七")):
            self.db.add(
                SysUser(
                    id=uid,
                    user_no=uid,
                    nick_name=name,
                    email="",
                    phone_no="",
                    password="",
                    status="1",
                    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",
                resolve_source="role",
                sign_mode="and",
                round_no=1,
                tenant_no="zkxw",
                order_num=1,
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.add(
            OaTask(
                id="t2",
                instance_id="i1",
                node_key="a1",
                node_name="审批人",
                task_type="approval",
                assignee_user_id="u3",
                status="pending",
                resolve_source="role",
                sign_mode="and",
                round_no=1,
                tenant_no="zkxw",
                order_num=2,
                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_transfer_rejects_self(self):
        from app.services.oa_engine import OaEngineError, transfer_task

        with self.assertRaises(OaEngineError) as ctx:
            transfer_task(
                self.db, "t1", tenant_no="zkxw", user_id="u1", target_user_id="u1"
            )
        self.assertIn("自己", str(ctx.exception))

    def test_transfer_rejects_existing_pending(self):
        from app.services.oa_engine import OaEngineError, transfer_task

        with self.assertRaises(OaEngineError) as ctx:
            transfer_task(
                self.db, "t1", tenant_no="zkxw", user_id="u1", target_user_id="u3"
            )
        self.assertIn("待办", str(ctx.exception))

    def test_transfer_rejects_non_assignee(self):
        from app.services.oa_engine import OaEngineError, transfer_task

        with self.assertRaises(OaEngineError) as ctx:
            transfer_task(
                self.db, "t1", tenant_no="zkxw", user_id="u2", target_user_id="u2"
            )
        self.assertIn("无权", str(ctx.exception))

    def test_transfer_success_and_keeps_co_approver(self):
        from app.services.oa_engine import list_tasks, transfer_task

        with patch(
            "app.services.oa_engine.get_instance_detail",
            return_value={"id": "i1", "status": "running"},
        ):
            transfer_task(
                self.db,
                "t1",
                tenant_no="zkxw",
                user_id="u1",
                target_user_id="u2",
                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()
        new_task = (
            self.db.query(self.OaTask)
            .filter(
                self.OaTask.instance_id == "i1",
                self.OaTask.assignee_user_id == "u2",
                self.OaTask.status == "pending",
            )
            .first()
        )
        self.assertEqual(t1.status, "transferred")
        self.assertEqual(t1.comment_text, "请协助处理")
        self.assertEqual(t2.status, "pending")
        self.assertIsNotNone(new_task)
        self.assertEqual(new_task.resolve_source, "transfer")
        self.assertEqual(new_task.sign_mode, "and")
        self.assertEqual(new_task.node_key, "a1")

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

        from app.services.oa_engine import build_progress_items

        tasks = (
            self.db.query(self.OaTask)
            .filter(self.OaTask.instance_id == "i1", self.OaTask.del_flag == 0)
            .all()
        )
        inst = self.db.query(self.OaInstance).filter(self.OaInstance.id == "i1").first()
        name_map = {"u1": "李四", "u2": "赵六", "u3": "钱七"}
        items = build_progress_items(
            inst=inst, flow_root=None, tasks=tasks, name_map=name_map
        )
        card = next(i for i in items if i.get("kind") == "task")
        transferred = next(p for p in card["people"] if p["status"] == "transferred")
        self.assertEqual(transferred["nickName"], "李四")
        self.assertEqual(transferred["transferToUserId"], "u2")
        self.assertEqual(transferred["transferToUserName"], "赵六")

    def test_transfer_comment_optional(self):
        from app.services.oa_engine import transfer_task

        with patch(
            "app.services.oa_engine.get_instance_detail",
            return_value={"id": "i1", "status": "running"},
        ):
            transfer_task(
                self.db,
                "t1",
                tenant_no="zkxw",
                user_id="u1",
                target_user_id="u2",
            )

        t1 = self.db.query(self.OaTask).filter(self.OaTask.id == "t1").first()
        self.assertEqual(t1.status, "transferred")
        self.assertIsNone(t1.comment_text)

    def test_transfer_chain(self):
        from app.services.oa_engine import transfer_task

        now = datetime.now()
        self.db.add(
            self.SysUser(
                id="u4",
                user_no="u4",
                nick_name="孙八",
                email="",
                phone_no="",
                password="",
                status="1",
                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": "running"},
        ):
            transfer_task(
                self.db,
                "t1",
                tenant_no="zkxw",
                user_id="u1",
                target_user_id="u2",
            )

        u2_task = (
            self.db.query(self.OaTask)
            .filter(
                self.OaTask.instance_id == "i1",
                self.OaTask.assignee_user_id == "u2",
                self.OaTask.status == "pending",
            )
            .first()
        )
        self.assertIsNotNone(u2_task)

        with patch(
            "app.services.oa_engine.get_instance_detail",
            return_value={"id": "i1", "status": "running"},
        ):
            transfer_task(
                self.db,
                u2_task.id,
                tenant_no="zkxw",
                user_id="u2",
                target_user_id="u4",
            )

        u2_task = self.db.query(self.OaTask).filter(self.OaTask.id == u2_task.id).first()
        u4_task = (
            self.db.query(self.OaTask)
            .filter(
                self.OaTask.instance_id == "i1",
                self.OaTask.assignee_user_id == "u4",
                self.OaTask.status == "pending",
            )
            .first()
        )
        self.assertEqual(u2_task.status, "transferred")
        self.assertIsNotNone(u4_task)
        self.assertEqual(u4_task.resolve_source, "transfer")


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