forked from erp-dev/erp
266 lines
13 KiB
Python
266 lines
13 KiB
Python
"""
|
||
测试 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}")
|