test(oa): add failing auto-pass-when-approver-is-initiator case

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

backend/app/crud/oa_approval.py

@@ -19,6 +19,7 @@ DEFAULT_ADVANCED_SETTINGS: Dict[str, Any] = {
     "commentRequired": False,
     "allowAddSign": True,
     "allowReturn": True,
+    "autoPassWhenApproverIsInitiator": False,
 }
 
 

backend/tests/test_oa_auto_pass_initiator.py

@@ -0,0 +1,116 @@
+"""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")
+
+
+if __name__ == "__main__":
+    unittest.main()