diff --git a/backend/app/crud/oa_approval.py b/backend/app/crud/oa_approval.py index caf92e1d..d18abb75 100644 --- a/backend/app/crud/oa_approval.py +++ b/backend/app/crud/oa_approval.py @@ -19,6 +19,7 @@ DEFAULT_ADVANCED_SETTINGS: Dict[str, Any] = { "commentRequired": False, "allowAddSign": True, "allowReturn": True, + "autoPassWhenApproverIsInitiator": False, } diff --git a/backend/tests/test_oa_auto_pass_initiator.py b/backend/tests/test_oa_auto_pass_initiator.py new file mode 100644 index 00000000..67634a8c --- /dev/null +++ b/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()