""" 测试 PlateOrder advance_to_next_state 是否会跳步 """ from django.test import TestCase from django.contrib.auth import get_user_model from printing import models as printing_models from stateflow import models as stateflow_models from basic_info import models as basic_info_models from stateflow import services as stateflow_services from django.conf import settings User = get_user_model() class PlateOrderAdvanceTestCase(TestCase): """测试 PlateOrder 推进到下一步是否会跳步""" def setUp(self): """设置测试数据""" # 创建用户 self.user = User.objects.create_user(username='testuser', password='testpass123') # 创建商户 self.merchant = basic_info_models.Merchant.objects.create( name='测试商户', type=basic_info_models.MerchantTypeEnum.FACTORY ) # 创建客户 self.customer = basic_info_models.Customer.objects.create( name='测试客户', merchant=self.merchant ) # 创建流程 - 至少3个状态节点 self.process = stateflow_models.Process.objects.create( name='开版流程测试', description='用于测试的开版流程' ) # 创建3个状态 self.state1 = stateflow_models.State.objects.create( name='设计', description='设计阶段' ) self.state2 = stateflow_models.State.objects.create( name='制版', description='制版阶段' ) self.state3 = stateflow_models.State.objects.create( name='验收', description='验收阶段' ) # 创建流程节点 self.node1 = stateflow_models.ProcessNode.objects.create( process=self.process, state=self.state1, order=1 ) self.node2 = stateflow_models.ProcessNode.objects.create( process=self.process, state=self.state2, order=2 ) self.node3 = stateflow_models.ProcessNode.objects.create( process=self.process, state=self.state3, order=3 ) def test_plate_order_advance_step_by_step(self): """测试 PlateOrder 一步一步推进，不应该跳步""" # 创建 PlateOrder，使用测试流程 plate_order = printing_models.PlateOrder.objects.create( design_code='TEST001', customer=self.customer, style_name='测试款式', process=self.process.id # 使用测试流程ID ) # 验证 business_object 已创建 self.assertIsNotNone(plate_order.business_object) bo = plate_order.business_object # 初始状态：应该是第一个节点 current = stateflow_services.get_business_object_current_state(bo) self.assertIsNotNone(current) self.assertEqual(current.id, self.state1.id, "初始状态应该是 state1") # 检查进度 progress = stateflow_services.get_progress_percentage(bo) self.assertEqual(progress, 0.0, "初始进度应该是 0%") print(f"\n初始状态: {current.name}") print(f"初始进度: {progress}%") # 第一次推进：完成 state1 success, message, state_log = stateflow_services.advance_to_next_state(bo, self.user) self.assertTrue(success, f"第一次推进应该成功: {message}") self.assertEqual(state_log.state.id, self.state1.id, "第一次推进应该完成 state1") # 检查完成后的状态 current_after_1 = stateflow_services.get_business_object_current_state(bo) progress_after_1 = stateflow_services.get_progress_percentage(bo) print(f"\n第一次推进后:") print(f" 完成的状态: {state_log.state.name}") print(f" 当前状态 (下一个待执行): {current_after_1.name if current_after_1 else 'None'}") print(f" 进度: {progress_after_1}%") self.assertIsNotNone(current_after_1, "第一次推进后，current_state 应该是 state2") self.assertEqual(current_after_1.id, self.state2.id, "第一次推进后，下一个待执行应该是 state2") self.assertAlmostEqual(progress_after_1, 33.33, places=1, msg="完成1/3，进度应该约为 33.33%") # 第二次推进：完成 state2 success, message, state_log = stateflow_services.advance_to_next_state(bo, self.user) self.assertTrue(success, f"第二次推进应该成功: {message}") self.assertEqual(state_log.state.id, self.state2.id, "第二次推进应该完成 state2") # 检查完成后的状态 current_after_2 = stateflow_services.get_business_object_current_state(bo) progress_after_2 = stateflow_services.get_progress_percentage(bo) print(f"\n第二次推进后:") print(f" 完成的状态: {state_log.state.name}") print(f" 当前状态 (下一个待执行): {current_after_2.name if current_after_2 else 'None'}") print(f" 进度: {progress_after_2}%") self.assertIsNotNone(current_after_2, "第二次推进后，current_state 应该是 state3") self.assertEqual(current_after_2.id, self.state3.id, "第二次推进后，下一个待执行应该是 state3") self.assertAlmostEqual(progress_after_2, 66.67, places=1, msg="完成2/3，进度应该约为 66.67%") # 第三次推进：完成 state3 success, message, state_log = stateflow_services.advance_to_next_state(bo, self.user) self.assertTrue(success, f"第三次推进应该成功: {message}") self.assertEqual(state_log.state.id, self.state3.id, "第三次推进应该完成 state3") # 检查完成后的状态 current_after_3 = stateflow_services.get_business_object_current_state(bo) progress_after_3 = stateflow_services.get_progress_percentage(bo) print(f"\n第三次推进后:") print(f" 完成的状态: {state_log.state.name}") print(f" 当前状态 (下一个待执行): {current_after_3}") print(f" 进度: {progress_after_3}%") self.assertIsNone(current_after_3, "第三次推进后，所有状态都已完成，current_state 应该是 None") self.assertEqual(progress_after_3, 100.0, "完成3/3，进度应该是 100%") # 验证不能再推进了 success, message, _ = stateflow_services.advance_to_next_state(bo, self.user) self.assertFalse(success, "流程已完成，不应该能继续推进") self.assertIn("已完成", message, "应该提示流程已完成") print(f"\n第四次推进（应该失败）: {message}") # 验证 StateFlowRecord 的数量 completed_count = bo.state_logs.filter(is_cancelled=False).count() self.assertEqual(completed_count, 3, "应该有3条完成记录") print(f"\n总结：") print(f" - 共推进3次，完成3个状态") print(f" - 没有跳步") print(f" - 完成记录数: {completed_count}") def test_plate_order_with_default_process(self): """测试使用默认流程ID的 PlateOrder""" # 确保默认流程存在 default_process_id = getattr(settings, 'PLATE_ORDER_DEFAULT_PROCESS_ID', None) if default_process_id: try: default_process = stateflow_models.Process.objects.get(id=default_process_id) print(f"\n默认流程: {default_process.name} (ID: {default_process_id})") # 创建 PlateOrder（会使用默认流程） plate_order = printing_models.PlateOrder.objects.create( design_code='TEST002', customer=self.customer, style_name='测试款式2' ) self.assertIsNotNone(plate_order.business_object) self.assertEqual(plate_order.process, default_process_id) # 获取默认流程的节点数 node_count = default_process.process_nodes.count() print(f"默认流程节点数: {node_count}") if node_count > 0: # 测试第一次推进 bo = plate_order.business_object first_state = stateflow_services.get_business_object_current_state(bo) print(f"初始状态: {first_state.name if first_state else 'None'}") success, message, state_log = stateflow_services.advance_to_next_state(bo, self.user) if success: print(f"第一次推进成功: {message}") current = stateflow_services.get_business_object_current_state(bo) print(f"推进后的当前状态: {current.name if current else 'None (已完成)'}") else: print(f"第一次推进失败: {message}") except stateflow_models.Process.DoesNotExist: print(f"\n警告：默认流程ID {default_process_id} 不存在") self.skipTest(f"默认流程ID {default_process_id} 不存在") else: print("\n警告：未配置 PLATE_ORDER_DEFAULT_PROCESS_ID") self.skipTest("未配置 PLATE_ORDER_DEFAULT_PROCESS_ID") def test_status_display_consistency(self): """测试 status 显示的一致性""" # 创建 PlateOrder plate_order = printing_models.PlateOrder.objects.create( design_code='TEST003', customer=self.customer, style_name='测试款式3', process=self.process.id ) bo = plate_order.business_object print(f"\n=== 测试 PlateOrder status 显示 ===") print(f"初始状态:") print(f" plate_order.status = '{plate_order.status}'") print(f" plate_order.status_id = {plate_order.status_id}") print(f" current_state (services) = {stateflow_services.get_business_object_current_state(bo).name if stateflow_services.get_business_object_current_state(bo) else 'None'}") print(f" 进度 = {plate_order.progress_percentage}%") # 第一次推进 success, message, state_log = stateflow_services.advance_to_next_state(bo, self.user) self.assertTrue(success) plate_order.refresh_from_db() print(f"\n第一次推进后（完成: {state_log.state.name}）:") print(f" plate_order.status = '{plate_order.status}'") print(f" plate_order.status_id = {plate_order.status_id}") print(f" current_state (services) = {stateflow_services.get_business_object_current_state(bo).name if stateflow_services.get_business_object_current_state(bo) else 'None'}") print(f" 进度 = {plate_order.progress_percentage}%") print(f" 已完成的状态: {[log.state.name for log in bo.state_logs.filter(is_cancelled=False)]}") # 验证：第一次推进后，status 应该显示下一个待执行的状态（state2） self.assertEqual(plate_order.status, self.state2.name, f"第一次推进后，status 应该是 '{self.state2.name}'，但实际是 '{plate_order.status}'") self.assertEqual(plate_order.status_id, self.state2.id, f"第一次推进后，status_id 应该是 {self.state2.id}，但实际是 {plate_order.status_id}") # 第二次推进 success, message, state_log = stateflow_services.advance_to_next_state(bo, self.user) self.assertTrue(success) plate_order.refresh_from_db() print(f"\n第二次推进后（完成: {state_log.state.name}）:") print(f" plate_order.status = '{plate_order.status}'") print(f" plate_order.status_id = {plate_order.status_id}") print(f" current_state (services) = {stateflow_services.get_business_object_current_state(bo).name if stateflow_services.get_business_object_current_state(bo) else 'None'}") print(f" 进度 = {plate_order.progress_percentage}%") print(f" 已完成的状态: {[log.state.name for log in bo.state_logs.filter(is_cancelled=False)]}") # 验证：第二次推进后，status 应该显示下一个待执行的状态（state3） self.assertEqual(plate_order.status, self.state3.name, f"第二次推进后，status 应该是 '{self.state3.name}'，但实际是 '{plate_order.status}'") self.assertEqual(plate_order.status_id, self.state3.id, f"第二次推进后，status_id 应该是 {self.state3.id}，但实际是 {plate_order.status_id}")