zkxw/backend/tests/test_oa_approver_dedup.py

"""OA：审批人去重 — 同一实例内已同意的人，后续节点系统自动通过。"""
import json
import unittest
from datetime import datetime
from unittest.mock import patch

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


def _resolve_from_config(_db, *, assignee_config, **_kwargs):
    ids = list(assignee_config.get("userIds") or [])
    return ids, {"resolve_source": "users"}


class TestApproverDedup(unittest.TestCase):
    def setUp(self):
        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.OaAppVersion = OaAppVersion
        self.OaInstance = OaInstance
        self.OaTask = OaTask
        self.now = datetime.now()

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

    def _seed(self, *, flow, advanced, initiator="u0"):
        self.db.add(
            self.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(advanced),
                tenant_no="zkxw",
                create_dept_id="",
                create_time=self.now,
                update_time=self.now,
                del_flag=0,
            )
        )
        self.db.add(
            self.OaInstance(
                id="i1",
                app_id="app1",
                app_version_id="v1",
                title="t",
                initiator_user_id=initiator,
                initiator_dept_id="d1",
                form_data_json="{}",
                status="running",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=self.now,
                update_time=self.now,
                del_flag=0,
            )
        )
        self.db.commit()

    def _load(self):
        inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
        ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
        return inst, ver

    def _tasks(self):
        self.db.flush()
        return self.db.query(self.OaTask).filter_by(instance_id="i1").all()

    def _pending(self, user_id):
        return next(
            t
            for t in self._tasks()
            if t.assignee_user_id == user_id and t.status == "pending"
        )

    def _approve(self, task):
        from app.services.oa_engine import _approve_task_core

        inst, ver = self._load()
        _approve_task_core(
            self.db,
            task=task,
            inst=inst,
            ver=ver,
            user_id=task.assignee_user_id,
            comment="同意",
        )

    def _chain_flow(self):
        """第 1 步 A，第 2 步 B，第 3、4 步再是 A。"""

        def approval(node_id, name, user_ids, child):
            return {
                "id": node_id,
                "type": "approval",
                "name": name,
                "approverConfig": {
                    "assigneeType": "users",
                    "userIds": user_ids,
                    "signMode": "or",
                },
                "child": child,
            }

        return {
            "id": "s",
            "type": "start",
            "child": approval(
                "a1",
                "第一步",
                ["A"],
                approval(
                    "a2",
                    "第二步",
                    ["B"],
                    approval(
                        "a3",
                        "第三步",
                        ["A"],
                        approval("a4", "第四步", ["A"], {"id": "e1", "type": "end"}),
                    ),
                ),
            ),
        }

    def test_later_nodes_auto_pass_after_same_person_approved(self):
        """A 在第 1 步同意后，第 3、4 步不再待办，系统记为自动通过。"""
        from app.services.oa_engine import advance_from

        self._seed(flow=self._chain_flow(), advanced={"approverDedup": True})
        inst, ver = self._load()
        with patch(
            "app.services.oa_engine.resolve_assignees", side_effect=_resolve_from_config
        ), patch("app.services.oa_engine._next_task_order_num", return_value=1):
            advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
            self._approve(self._pending("A"))
            self._approve(self._pending("B"))

        by_node = {t.node_key: t for t in self._tasks()}
        self.assertEqual(by_node["a1"].status, "approved")
        self.assertEqual(by_node["a1"].comment_text, "同意")
        self.assertEqual(by_node["a3"].status, "approved")
        self.assertEqual(by_node["a3"].comment_text, "审批人去重，系统自动通过")
        self.assertEqual(by_node["a4"].status, "approved")
        self.assertEqual(by_node["a4"].comment_text, "审批人去重，系统自动通过")
        self.assertEqual(inst.status, "completed")
        self.assertFalse(any(t.status == "pending" for t in by_node.values()))

    def test_switch_off_keeps_repeat_approver_pending(self):
        from app.services.oa_engine import advance_from

        self._seed(flow=self._chain_flow(), advanced={"approverDedup": False})
        inst, ver = self._load()
        with patch(
            "app.services.oa_engine.resolve_assignees", side_effect=_resolve_from_config
        ), patch("app.services.oa_engine._next_task_order_num", return_value=1):
            advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
            self._approve(self._pending("A"))
            self._approve(self._pending("B"))

        step3 = next(t for t in self._tasks() if t.node_key == "a3")
        self.assertEqual(step3.assignee_user_id, "A")
        self.assertEqual(step3.status, "pending")
        self.assertEqual(inst.status, "running")
        self.assertFalse(any(t.node_key == "a4" for t in self._tasks()))

    def test_or_sign_auto_pass_closes_whole_node(self):
        from app.services.oa_engine import advance_from

        flow = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "第一步",
                "approverConfig": {
                    "assigneeType": "users",
                    "userIds": ["A"],
                    "signMode": "or",
                },
                "child": {
                    "id": "a2",
                    "type": "approval",
                    "name": "或签",
                    "approverConfig": {
                        "assigneeType": "users",
                        "userIds": ["A", "C"],
                        "signMode": "or",
                    },
                    "child": {"id": "e1", "type": "end"},
                },
            },
        }
        self._seed(flow=flow, advanced={"approverDedup": True})
        inst, ver = self._load()
        with patch(
            "app.services.oa_engine.resolve_assignees", side_effect=_resolve_from_config
        ), patch("app.services.oa_engine._next_task_order_num", return_value=1):
            advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
            self._approve(self._pending("A"))

        by_key = {(t.node_key, t.assignee_user_id): t for t in self._tasks()}
        self.assertEqual(by_key[("a2", "A")].status, "approved")
        self.assertEqual(by_key[("a2", "A")].comment_text, "审批人去重，系统自动通过")
        self.assertEqual(by_key[("a2", "C")].status, "done")
        self.assertEqual(inst.status, "completed")

    def test_and_sign_only_auto_passes_repeat_person(self):
        from app.services.oa_engine import advance_from

        flow = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "第一步",
                "approverConfig": {
                    "assigneeType": "users",
                    "userIds": ["A"],
                    "signMode": "or",
                },
                "child": {
                    "id": "a2",
                    "type": "approval",
                    "name": "会签",
                    "approverConfig": {
                        "assigneeType": "users",
                        "userIds": ["A", "C"],
                        "signMode": "and",
                    },
                    "child": {"id": "e1", "type": "end"},
                },
            },
        }
        self._seed(flow=flow, advanced={"approverDedup": True})
        inst, ver = self._load()
        with patch(
            "app.services.oa_engine.resolve_assignees", side_effect=_resolve_from_config
        ), patch("app.services.oa_engine._next_task_order_num", return_value=1):
            advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
            self._approve(self._pending("A"))

        by_user = {
            t.assignee_user_id: t for t in self._tasks() if t.node_key == "a2"
        }
        self.assertEqual(by_user["A"].status, "approved")
        self.assertEqual(by_user["A"].comment_text, "审批人去重，系统自动通过")
        self.assertEqual(by_user["C"].status, "pending")
        self.assertEqual(inst.status, "running")

    def test_non_approved_history_does_not_dedup(self):
        """仅统计已同意；拒绝等状态不触发后续自动通过。"""
        from app.services.oa_engine import advance_from

        flow = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "当前",
                "approverConfig": {
                    "assigneeType": "users",
                    "userIds": ["A"],
                    "signMode": "or",
                },
                "child": {"id": "e1", "type": "end"},
            },
        }
        self._seed(flow=flow, advanced={"approverDedup": True})
        self.db.add(
            self.OaTask(
                id="old-reject",
                instance_id="i1",
                node_key="old",
                node_name="历史",
                task_type="approval",
                assignee_user_id="A",
                status="rejected",
                sign_mode="or",
                round_no=1,
                is_read=0,
                tenant_no="zkxw",
                order_num=0,
                create_dept_id="",
                create_time=self.now,
                update_time=self.now,
                del_flag=0,
            )
        )
        self.db.commit()
        inst, ver = self._load()
        with patch(
            "app.services.oa_engine.resolve_assignees", side_effect=_resolve_from_config
        ), patch("app.services.oa_engine._next_task_order_num", return_value=1):
            advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")

        current = next(t for t in self._tasks() if t.node_key == "a1")
        self.assertEqual(current.status, "pending")
        self.assertEqual(inst.status, "running")