""" Stateflow 服务层单元测试 测试新增的服务函数以及 current_state 语义修正 """ from django.test import TestCase from django.contrib.auth import get_user_model from django.contrib.contenttypes.models import ContentType from stateflow import models, services User = get_user_model() class StateFlowServicesTestCase(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) # 创建业务对象 ct = ContentType.objects.get_for_model(models.Process) self.business_object = models.BusinessObject.objects.create( name='测试订单', process=self.process, description='测试订单描述', content_type=ct, object_id=self.process.id ) def test_initial_state(self): """测试初始状态 - 应该是 None（未开始）""" current_state = self.business_object.get_current_state() self.assertIsNone(current_state, '初始状态应该是 None（未开始）') self.assertEqual(self.business_object.get_progress_percentage(), 0.0) # 整体状态应该是 not_started status = services.get_overall_status(self.business_object) self.assertEqual(status, 'not_started') def test_advance_to_next_state(self): """测试推进到下一个状态""" # 第一次推进应该完成第一个状态 success, message, _ = services.advance_to_next_state(self.business_object, self.user) self.assertTrue(success) # current_state 应该是最后完成的状态（state1） current_state = self.business_object.get_current_state() self.assertEqual(current_state.name, self.state1.name) self.assertAlmostEqual(self.business_object.get_progress_percentage(), 33.33, places=1) # 推进到状态2（完成state2） success, message, _ = services.advance_to_next_state(self.business_object, self.user) self.assertTrue(success) # current_state 应该是 state2 current_state = self.business_object.get_current_state() self.assertEqual(current_state.name, self.state2.name) self.assertAlmostEqual(self.business_object.get_progress_percentage(), 66.67, places=1) # 完成最后一个状态（state3） success, message, _ = services.advance_to_next_state(self.business_object, self.user) self.assertTrue(success) self.assertEqual(self.business_object.get_progress_percentage(), 100.0) # current_state 应该是 state3（最后完成的状态） current_state = self.business_object.get_current_state() self.assertEqual(current_state.name, self.state3.name) # 整体状态应该是 completed status = services.get_overall_status(self.business_object) self.assertEqual(status, 'completed') def test_state_status(self): """测试状态的状态""" # 初始状态，所有状态都应该是 not_started self.assertEqual(services.get_business_object_state_status(self.business_object, self.state1), 'not_started') self.assertEqual(services.get_business_object_state_status(self.business_object, self.state2), 'not_started') # 完成第一个状态 services.advance_to_next_state(self.business_object, self.user) self.assertEqual(services.get_business_object_state_status(self.business_object, self.state1), 'completed') self.assertEqual(services.get_business_object_state_status(self.business_object, self.state2), 'not_started') self.assertEqual(services.get_business_object_state_status(self.business_object, self.state3), 'not_started') # 完成第二个状态 services.advance_to_next_state(self.business_object, self.user) self.assertEqual(services.get_business_object_state_status(self.business_object, self.state1), 'completed') self.assertEqual(services.get_business_object_state_status(self.business_object, self.state2), 'completed') self.assertEqual(services.get_business_object_state_status(self.business_object, self.state3), 'not_started') def test_timeline(self): """测试时间线""" timeline = self.business_object.get_timeline() self.assertEqual(len(timeline), 3) # 所有状态都应该是 not_started for item in timeline: self.assertEqual(item['status'], 'not_started') # 完成第一个状态 services.advance_to_next_state(self.business_object, self.user) timeline = self.business_object.get_timeline() self.assertEqual(timeline[0]['status'], 'completed') self.assertEqual(timeline[1]['status'], 'not_started') self.assertEqual(timeline[2]['status'], 'not_started') self.assertIsNotNone(timeline[0]['completed_by']) self.assertEqual(timeline[0]['completed_by'].id, self.user.id) self.assertFalse(timeline[0]['is_cancelled']) def test_reset_progress(self): """测试重置进度 - 不删除记录，而是标记为已撤销""" # 完成两个状态 services.advance_to_next_state(self.business_object, self.user) services.advance_to_next_state(self.business_object, self.user) # 确认有2条日志记录 self.assertEqual(self.business_object.state_logs.count(), 2) # 重置进度 services.reset_business_object_progress(self.business_object) # 日志记录应该还在（不删除） self.assertEqual(self.business_object.state_logs.count(), 2) # 但所有记录都应该标记为已撤销 cancelled_count = self.business_object.state_logs.filter(is_cancelled=True).count() self.assertEqual(cancelled_count, 2) # 应该回到初始状态（未开始） current_state = self.business_object.get_current_state() self.assertIsNone(current_state) # 所有撤销的记录都应该有撤销时间 for log in self.business_object.state_logs.all(): self.assertTrue(log.is_cancelled) self.assertIsNotNone(log.cancelled_at) def test_timeline_with_cancelled(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) # 重置进度 services.reset_business_object_progress(self.business_object) timeline = self.business_object.get_timeline() # 所有状态都应该显示为 cancelled for item in timeline: self.assertEqual(item['status'], 'cancelled') self.assertTrue(item['is_cancelled']) self.assertIsNotNone(item['cancelled_at']) def test_get_next_pending_state(self): """测试获取下一个待执行节点（新增功能）""" # 创建参数 param1 = models.StateParameter.objects.create( key='param1', value='value1', is_required=True ) param2 = models.StateParameter.objects.create( key='param2', value='value2', is_required=False ) self.state1.parameters.add(param1, param2) # 初始状态 - 应该返回第一个节点 next_state = services.get_next_pending_state(self.business_object) self.assertIsNotNone(next_state) self.assertEqual(next_state['state'].name, self.state1.name) self.assertEqual(next_state['order'], 0) self.assertIn('parameters', next_state) self.assertEqual(len(next_state['parameters']), 2) # 测试不包含参数 next_state_no_params = services.get_next_pending_state( self.business_object, include_parameters=False ) self.assertIsNotNone(next_state_no_params) self.assertNotIn('parameters', next_state_no_params) # 完成第一个状态（提供必填参数） success, msg, _ = services.advance_to_next_state( self.business_object, self.user, param1='value1' # 提供必填参数 ) self.assertTrue(success) # 应该返回第二个节点 next_state = services.get_next_pending_state(self.business_object) self.assertIsNotNone(next_state) self.assertEqual(next_state['state'].name, self.state2.name) self.assertEqual(next_state['order'], 1) # 完成所有状态 services.advance_to_next_state(self.business_object, self.user) services.advance_to_next_state(self.business_object, self.user) # 应该返回 None（已完成） next_state = services.get_next_pending_state(self.business_object) self.assertIsNone(next_state) def test_get_all_pending_states(self): """测试获取所有待执行节点列表（新增功能）""" # 初始状态 - 所有节点都待执行 pending_states = services.get_all_pending_states(self.business_object) self.assertEqual(len(pending_states), 3) self.assertEqual(pending_states[0]['state'].name, self.state1.name) self.assertEqual(pending_states[1]['state'].name, self.state2.name) self.assertEqual(pending_states[2]['state'].name, self.state3.name) # 验证不包含 parameters 字段 self.assertNotIn('parameters', pending_states[0]) # 完成第一个状态 services.advance_to_next_state(self.business_object, self.user) # 应该只剩下两个待执行节点 pending_states = services.get_all_pending_states(self.business_object) self.assertEqual(len(pending_states), 2) self.assertEqual(pending_states[0]['state'].name, self.state2.name) self.assertEqual(pending_states[1]['state'].name, self.state3.name) # 完成第二个状态 services.advance_to_next_state(self.business_object, self.user) # 应该只剩下一个待执行节点 pending_states = services.get_all_pending_states(self.business_object) self.assertEqual(len(pending_states), 1) self.assertEqual(pending_states[0]['state'].name, self.state3.name) # 完成所有状态 services.advance_to_next_state(self.business_object, self.user) # 应该没有待执行节点 pending_states = services.get_all_pending_states(self.business_object) self.assertEqual(len(pending_states), 0) def test_get_state_parameters(self): """测试获取节点参数列表（新增功能）""" # 创建必填和可选参数 required_param = models.StateParameter.objects.create( key='required_param', value='required_value', is_required=True ) optional_param1 = models.StateParameter.objects.create( key='optional_param1', value='optional_value1', is_required=False ) optional_param2 = models.StateParameter.objects.create( key='optional_param2', value='optional_value2', is_required=False ) # 关联参数到状态 self.state1.parameters.add(required_param, optional_param1, optional_param2) # 获取所有参数 all_params = services.get_state_parameters(self.state1) self.assertEqual(len(all_params), 3) # 只获取必填参数 required_params = services.get_state_parameters(self.state1, required_only=True) self.assertEqual(len(required_params), 1) self.assertEqual(required_params[0].key, 'required_param') self.assertTrue(required_params[0].is_required) # 测试没有参数的状态 no_params = services.get_state_parameters(self.state2) self.assertEqual(len(no_params), 0) # 测试只有必填参数的状态 self.state3.parameters.add(required_param) params = services.get_state_parameters(self.state3, required_only=True) self.assertEqual(len(params), 1)