""" 测试单步回退功能 """ from django.test import TestCase from django.contrib.auth import get_user_model from stateflow import models, services User = get_user_model() class StepBackTestCase(TestCase): """测试单步回退功能""" def setUp(self): # 创建测试用户 self.user = User.objects.create_user(username='testuser', password='testpass') # 创建状态 self.state1 = models.State.objects.create(name='状态1', description='第一个状态') self.state2 = models.State.objects.create(name='状态2', description='第二个状态') self.state3 = models.State.objects.create(name='状态3', description='第三个状态') # 创建流程 self.process = models.Process.objects.create(name='测试流程', description='三个状态的测试流程') # 创建流程节点 models.ProcessNode.objects.create(process=self.process, state=self.state1, order=0) models.ProcessNode.objects.create(process=self.process, state=self.state2, order=1) models.ProcessNode.objects.create(process=self.process, state=self.state3, order=2) # 创建业务对象 self.business_object = models.BusinessObject.objects.create( name='测试业务对象', process=self.process, description='测试回退功能' ) def test_step_back_from_not_started(self): """测试从未开始状态回退（应该返回失败）""" success, message = services.step_back_one_state(self.business_object, self.user) self.assertFalse(success) self.assertIn('没有任何状态流转记录', message) # 验证仍然是未开始状态 self.assertEqual(services.get_overall_status(self.business_object), 'in_progress') def test_step_back_from_first_state(self): """测试从第一个状态回退到未开始""" # 推进（完成第一个状态） success, _, _ = services.advance_to_next_state(self.business_object, self.user) self.assertTrue(success) # 验证 current_state 是 state2（下一个待执行的状态） current_state = services.get_business_object_current_state(self.business_object) self.assertEqual(current_state.name, self.state2.name) # 回退一步 success, message = services.step_back_one_state(self.business_object, self.user) self.assertTrue(success) self.assertIn('状态1', message) # 验证回到未开始状态 self.assertEqual(services.get_overall_status(self.business_object), 'in_progress') current_state = services.get_business_object_current_state(self.business_object) self.assertEqual(current_state.name, self.state1.name) # 验证状态1的记录已被撤销 record = models.StateFlowRecord.objects.filter( business_object=self.business_object, state=self.state1 ).first() self.assertTrue(record.is_cancelled) self.assertIsNotNone(record.cancelled_at) def test_step_back_from_middle_state(self): """测试从中间状态回退""" # 推进两次（完成 state1 和 state2） services.advance_to_next_state(self.business_object, self.user) services.advance_to_next_state(self.business_object, self.user) # 验证 current_state 是 state3（下一个待执行的状态） current_state = services.get_business_object_current_state(self.business_object) self.assertEqual(current_state.name, self.state3.name) # 回退一步 success, message = services.step_back_one_state(self.business_object, self.user) self.assertTrue(success) self.assertIn('状态2', message) # 验证 current_state 变为 state2（下一个待执行的） current_state = services.get_business_object_current_state(self.business_object) self.assertEqual(current_state.name, self.state2.name) # 验证状态2的记录已被撤销 record = models.StateFlowRecord.objects.filter( business_object=self.business_object, state=self.state2 ).first() self.assertTrue(record.is_cancelled) def test_step_back_from_completed(self): """测试从已完成状态回退""" # 推进到完成（完成所有三个状态） services.advance_to_next_state(self.business_object, self.user) services.advance_to_next_state(self.business_object, self.user) services.advance_to_next_state(self.business_object, self.user) # 验证已完成，current_state 是 None（没有待执行的节点） self.assertEqual(services.get_overall_status(self.business_object), 'completed') current_state = services.get_business_object_current_state(self.business_object) self.assertIsNone(current_state) # 回退一步 success, message = services.step_back_one_state(self.business_object, self.user) self.assertTrue(success) self.assertIn('状态3', message) # 验证回到进行中状态，current_state 变为 state3（下一个待执行的） self.assertEqual(services.get_overall_status(self.business_object), 'in_progress') current_state = services.get_business_object_current_state(self.business_object) self.assertEqual(current_state.name, self.state3.name) def test_step_back_multiple_times(self): """测试多次回退""" # 推进到完成 services.advance_to_next_state(self.business_object, self.user) services.advance_to_next_state(self.business_object, self.user) services.advance_to_next_state(self.business_object, self.user) # 第一次回退（撤销 state3） success, _ = services.step_back_one_state(self.business_object, self.user) self.assertTrue(success) current = services.get_business_object_current_state(self.business_object) self.assertEqual(current.name, self.state3.name) # 第二次回退（撤销 state2） success, _ = services.step_back_one_state(self.business_object, self.user) self.assertTrue(success) current = services.get_business_object_current_state(self.business_object) self.assertEqual(current.name, self.state2.name) # 下一个待执行的是 state2 # 第三次回退 success, _ = services.step_back_one_state(self.business_object, self.user) self.assertTrue(success) self.assertEqual(services.get_overall_status(self.business_object), 'in_progress') # 第四次回退（应该失败） success, message = services.step_back_one_state(self.business_object, self.user) self.assertFalse(success) self.assertIn('没有任何状态流转记录', message) def test_advance_after_step_back(self): """测试回退后再前进""" # 推进两步（完成 state1 和 state2） services.advance_to_next_state(self.business_object, self.user) services.advance_to_next_state(self.business_object, self.user) # 回退一步（撤销 state2） services.step_back_one_state(self.business_object, self.user) # 再前进（重新完成 state2） success, _, _ = services.advance_to_next_state(self.business_object, self.user) self.assertTrue(success) # 验证 current_state 是 state3（下一个待执行的） current_state = services.get_business_object_current_state(self.business_object) self.assertEqual(current_state.name, self.state3.name) def test_step_back_preserves_history(self): """测试回退操作保留历史记录""" # 推进两步 services.advance_to_next_state(self.business_object, self.user) services.advance_to_next_state(self.business_object, self.user) # 回退一步 services.step_back_one_state(self.business_object, self.user) # 验证记录仍然存在，只是被标记为已撤销 all_records = models.StateFlowRecord.objects.filter( business_object=self.business_object ).order_by('completed_at') self.assertEqual(all_records.count(), 2) self.assertFalse(all_records[0].is_cancelled) # 状态1未撤销 self.assertTrue(all_records[1].is_cancelled) # 状态2已撤销