zkxw/backend/tests/test_oa_auto_pass_initiator.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


class TestAutoPassInitiator(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", tasks=None):
        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,
            )
        )
        for t in tasks or []:
            self.db.add(t)
        self.db.commit()

    def test_or_sign_sole_initiator_auto_passes_to_end(self):
        """开关开：或签唯一审批人=发起人 → 提交推进后自动通过到结束。"""
        from app.services.oa_engine import advance_from

        flow = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "审批人",
                "approverConfig": {
                    "assigneeType": "users",
                    "userIds": ["u0"],
                    "signMode": "or",
                },
                "child": {"id": "e1", "type": "end"},
            },
        }
        self._seed(
            flow=flow,
            advanced={"autoPassWhenApproverIsInitiator": True},
        )
        inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
        ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()

        with patch(
            "app.services.oa_engine.resolve_assignees",
            return_value=(["u0"], {"resolve_source": "users"}),
        ), 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"
            )

        # Session autoflush=False：未 flush 时 query 看不到刚 add 的任务
        self.db.flush()
        tasks = self.db.query(self.OaTask).filter_by(instance_id="i1").all()
        self.assertEqual(len(tasks), 1)
        self.assertEqual(tasks[0].status, "approved")
        self.assertEqual(tasks[0].comment_text, "系统自动通过")
        self.assertEqual(inst.status, "completed")

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

        flow = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "审批人",
                "approverConfig": {
                    "assigneeType": "users",
                    "userIds": ["u0"],
                    "signMode": "or",
                },
                "child": {"id": "e1", "type": "end"},
            },
        }
        self._seed(flow=flow, advanced={"autoPassWhenApproverIsInitiator": False})
        inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
        ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
        with patch(
            "app.services.oa_engine.resolve_assignees",
            return_value=(["u0"], {"resolve_source": "users"}),
        ), 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.db.flush()
        t = self.db.query(self.OaTask).filter_by(instance_id="i1").one()
        self.assertEqual(t.status, "pending")
        self.assertEqual(inst.status, "running")

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

        flow = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "审批人",
                "approverConfig": {
                    "assigneeType": "users",
                    "userIds": ["u0", "u2"],
                    "signMode": "or",
                },
                "child": {"id": "e1", "type": "end"},
            },
        }
        self._seed(flow=flow, advanced={"autoPassWhenApproverIsInitiator": True})
        inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
        ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
        with patch(
            "app.services.oa_engine.resolve_assignees",
            return_value=(["u0", "u2"], {"resolve_source": "users"}),
        ), 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.db.flush()
        by_user = {
            t.assignee_user_id: t
            for t in self.db.query(self.OaTask).filter_by(instance_id="i1").all()
        }
        self.assertEqual(by_user["u0"].status, "approved")
        self.assertEqual(by_user["u0"].comment_text, "系统自动通过")
        self.assertEqual(by_user["u2"].status, "done")
        self.assertEqual(inst.status, "completed")

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

        flow = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "审批人",
                "approverConfig": {
                    "assigneeType": "users",
                    "userIds": ["u0", "u2"],
                    "signMode": "and",
                },
                "child": {"id": "e1", "type": "end"},
            },
        }
        self._seed(flow=flow, advanced={"autoPassWhenApproverIsInitiator": True})
        inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
        ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
        with patch(
            "app.services.oa_engine.resolve_assignees",
            return_value=(["u0", "u2"], {"resolve_source": "users"}),
        ), 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.db.flush()
        by_user = {
            t.assignee_user_id: t
            for t in self.db.query(self.OaTask).filter_by(instance_id="i1").all()
        }
        self.assertEqual(by_user["u0"].status, "approved")
        self.assertEqual(by_user["u2"].status, "pending")
        self.assertEqual(inst.status, "running")

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

        flow = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "一审",
                "approverConfig": {
                    "assigneeType": "users",
                    "userIds": ["u0"],
                    "signMode": "or",
                },
                "child": {
                    "id": "a2",
                    "type": "approval",
                    "name": "二审",
                    "approverConfig": {
                        "assigneeType": "users",
                        "userIds": ["u0"],
                        "signMode": "or",
                    },
                    "child": {"id": "e1", "type": "end"},
                },
            },
        }
        self._seed(flow=flow, advanced={"autoPassWhenApproverIsInitiator": True})
        inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
        ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
        with patch(
            "app.services.oa_engine.resolve_assignees",
            return_value=(["u0"], {"resolve_source": "users"}),
        ), 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.db.flush()
        tasks = self.db.query(self.OaTask).filter_by(instance_id="i1").all()
        self.assertEqual(len(tasks), 2)
        self.assertTrue(all(t.status == "approved" for t in tasks))
        self.assertTrue(all(t.comment_text == "系统自动通过" for t in tasks))
        self.assertEqual(inst.status, "completed")

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

        flow = {
            "id": "s",
            "type": "start",
            "child": {
                "id": "a1",
                "type": "approval",
                "name": "审批人",
                "approverConfig": {
                    "assigneeType": "users",
                    "userIds": ["u0"],
                    "signMode": "or",
                },
                "child": {"id": "e1", "type": "end"},
            },
        }
        self._seed(
            flow=flow,
            advanced={
                "autoPassWhenApproverIsInitiator": True,
                "commentRequired": True,
            },
        )
        inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
        ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
        with patch(
            "app.services.oa_engine.resolve_assignees",
            return_value=(["u0"], {"resolve_source": "users"}),
        ), 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.db.flush()
        t = self.db.query(self.OaTask).filter_by(instance_id="i1").one()
        self.assertEqual(t.status, "approved")
        self.assertEqual(inst.status, "completed")


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