test(oa): cover auto-pass or/and/consecutive/commentRequired cases

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/tests/test_oa_auto_pass_initiator.py

@@ -111,6 +111,198 @@ class TestAutoPassInitiator(unittest.TestCase):
         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()