""" 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 ) self.business_object_pending = models.BusinessObject.objects.create( name='待开始订单', process=self.process, description='尚未推进的订单', content_type=ct, object_id=self.process.id ) self.other_process = models.Process.objects.create(name='其他流程', description='另一个流程用于过滤测试') models.ProcessNode.objects.create(process=self.other_process, state=self.state1, order=0) self.other_business_object = models.BusinessObject.objects.create( name='其他流程订单', process=self.other_process, description='不同流程的订单', content_type=ct, object_id=self.other_process.id ) def test_initial_state(self): """测试初始状态 - current_state 是下一个待执行节点（第一个节点）""" current_state = self.business_object.get_current_state() # current_state 返回"下一个待执行的节点"，未开始时是第一个节点 self.assertEqual(current_state.name, self.state1.name) self.assertEqual(self.business_object.get_progress_percentage(), 0.0) # 整体状态应该是 in_progress（废除了 not_started） status = services.get_overall_status(self.business_object) self.assertEqual(status, 'in_progress') 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后是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(), 33.33, places=1) # 推进到状态2（完成state2） success, message, _ = services.advance_to_next_state(self.business_object, self.user) self.assertTrue(success) # current_state 应该是 state3（下一个待执行的） current_state = self.business_object.get_current_state() self.assertEqual(current_state.name, self.state3.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 应该是 None（没有待执行的节点了） current_state = self.business_object.get_current_state() self.assertIsNone(current_state) # 整体状态应该是 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') # 第二个状态是 current_state（下一个待执行的节点），还未开始，所以是 not_started 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 返回下一个待执行的节点（第一个节点） current_state = self.business_object.get_current_state() self.assertEqual(current_state.name, self.state1.name) # 所有撤销的记录都应该有撤销时间 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) def test_get_process_nodes(self): """测试获取业务对象所属流程的所有状态节点""" # 获取流程节点 nodes = services.get_process_nodes(self.business_object) # 验证节点数量 self.assertEqual(len(nodes), 3) # 验证节点顺序和内容 self.assertEqual(nodes[0]['state_id'], self.state1.id) self.assertEqual(nodes[0]['state_name'], self.state1.name) self.assertEqual(nodes[0]['order'], 0) self.assertEqual(nodes[1]['state_id'], self.state2.id) self.assertEqual(nodes[1]['state_name'], self.state2.name) self.assertEqual(nodes[1]['order'], 1) self.assertEqual(nodes[2]['state_id'], self.state3.id) self.assertEqual(nodes[2]['state_name'], self.state3.name) self.assertEqual(nodes[2]['order'], 2) # 验证每个节点都有必要的字段 for node in nodes: self.assertIn('id', node) self.assertIn('state', node) self.assertIn('state_id', node) self.assertIn('state_name', node) self.assertIn('order', node) self.assertIsInstance(node['state'], models.State) def test_query_business_objects_by_state_status_filters_by_completion(self): """验证按节点状态过滤业务对象的服务函数""" # 初始状态：两个业务对象都未开始 state1 qs = services.query_business_objects_by_state_status( state_ids=[self.state1.id], status='not_started', process_id=self.process.id, ) ids = set(qs.values_list('id', flat=True)) self.assertIn(self.business_object.id, ids) self.assertIn(self.business_object_pending.id, ids) self.assertNotIn(self.other_business_object.id, ids) # 推进第一个业务对象，变为已完成 state1 services.advance_to_next_state(self.business_object, self.user) qs = services.query_business_objects_by_state_status( state_ids=[self.state1.id], status='not_started', process_id=self.process.id, ) ids = set(qs.values_list('id', flat=True)) self.assertNotIn(self.business_object.id, ids) self.assertIn(self.business_object_pending.id, ids) qs_completed = services.query_business_objects_by_state_status( state_ids=[self.state1.id], status='completed', ) completed_ids = set(qs_completed.values_list('id', flat=True)) self.assertIn(self.business_object.id, completed_ids) self.assertNotIn(self.business_object_pending.id, completed_ids) def test_query_business_objects_by_state_status_cancelled(self): """撤销记录应该匹配 cancelled 状态""" services.advance_to_next_state(self.business_object_pending, self.user) services.reset_business_object_progress(self.business_object_pending) qs = services.query_business_objects_by_state_status( state_ids=[self.state1.id], status='cancelled', ) ids = set(qs.values_list('id', flat=True)) self.assertIn(self.business_object_pending.id, ids) def test_query_business_objects_by_state_status_invalid_status(self): """非法状态值应抛出异常""" with self.assertRaises(ValueError): services.query_business_objects_by_state_status(status='unknown')