From 1fafb4a1576775fb02508f137f6e42ab04381ff3 Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Mon, 27 Jul 2026 17:43:18 +0800 Subject: [PATCH] test(oa): cover auto-pass or/and/consecutive/commentRequired cases Co-authored-by: Cursor --- backend/tests/test_oa_auto_pass_initiator.py | 192 +++++++++++++++++++ 1 file changed, 192 insertions(+) diff --git a/backend/tests/test_oa_auto_pass_initiator.py b/backend/tests/test_oa_auto_pass_initiator.py index 67634a8c..0e930cff 100644 --- a/backend/tests/test_oa_auto_pass_initiator.py +++ b/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()