forked from erp-dev/erp
155 lines
5.6 KiB
Python
155 lines
5.6 KiB
Python
"""
|
||
调试脚本：检查 advance_to_next_state 是否会导致"前进2步"的bug
|
||
"""
|
||
import os
|
||
import django
|
||
|
||
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'flower.settings')
|
||
django.setup()
|
||
|
||
from django.contrib.auth import get_user_model
|
||
from stateflow.models import State, Process, BusinessObject, StateFlowRecord
|
||
from stateflow.services import advance_to_next_state, get_business_object_current_state, get_completed_node_ids, get_progress_percentage
|
||
|
||
User = get_user_model()
|
||
|
||
def main():
|
||
# 创建测试用户
|
||
user = User.objects.first()
|
||
if not user:
|
||
user = User.objects.create_user(username='debuguser', password='debug123')
|
||
print(f"创建测试用户: {user.username}")
|
||
|
||
# 创建3个状态节点
|
||
state1 = State.objects.create(name='Debug State 1')
|
||
state2 = State.objects.create(name='Debug State 2')
|
||
state3 = State.objects.create(name='Debug State 3')
|
||
print(f"\n创建状态节点:")
|
||
print(f" - {state1.name} (ID: {state1.id})")
|
||
print(f" - {state2.name} (ID: {state2.id})")
|
||
print(f" - {state3.name} (ID: {state3.id})")
|
||
|
||
# 创建流程
|
||
process = Process.objects.create(name='Debug Process')
|
||
process.replace_nodes([state1, state2, state3])
|
||
print(f"\n创建流程: {process.name}")
|
||
print(f"节点顺序: {state1.name} -> {state2.name} -> {state3.name}")
|
||
|
||
# 创建业务对象
|
||
bo = BusinessObject.objects.create(name='Debug BO', process=process)
|
||
print(f"\n创建业务对象: {bo.name}")
|
||
|
||
# 检查初始状态
|
||
print(f"\n{'='*60}")
|
||
print(f"初始状态检查")
|
||
print(f"{'='*60}")
|
||
current = get_business_object_current_state(bo)
|
||
completed_ids = get_completed_node_ids(bo)
|
||
progress = get_progress_percentage(bo)
|
||
logs_count = StateFlowRecord.objects.filter(business_object=bo, is_cancelled=False).count()
|
||
|
||
print(f"Current State: {current.name if current else 'None'} (ID: {current.id if current else 'N/A'})")
|
||
print(f"Completed Node IDs: {completed_ids}")
|
||
print(f"Progress: {progress:.2f}%")
|
||
print(f"Log Records Count: {logs_count}")
|
||
print(f"\n✓ 期望: Current='{state1.name}', Completed=[], Progress=0%, Logs=0")
|
||
|
||
if current and current.id != state1.id:
|
||
print(f"❌ 错误: Current State 应该是 '{state1.name}' 但实际是 '{current.name}'")
|
||
cleanup(bo, process, [state1, state2, state3])
|
||
return False
|
||
|
||
if len(completed_ids) != 0:
|
||
print(f"❌ 错误: 应该没有完成的节点，但有 {len(completed_ids)} 个")
|
||
cleanup(bo, process, [state1, state2, state3])
|
||
return False
|
||
|
||
print("✅ 初始状态正确")
|
||
|
||
# 第一次推进
|
||
print(f"\n{'='*60}")
|
||
print(f"第一次调用 advance_to_next_state")
|
||
print(f"{'='*60}")
|
||
|
||
success, message, log = advance_to_next_state(bo, user)
|
||
|
||
print(f"Success: {success}")
|
||
print(f"Message: {message}")
|
||
print(f"Created Log ID: {log.id if log else 'None'}")
|
||
if log:
|
||
print(f"Log State: {log.state.name} (ID: {log.state_id})")
|
||
|
||
# 检查推进后的状态
|
||
print(f"\n推进后状态检查:")
|
||
current = get_business_object_current_state(bo)
|
||
completed_ids = get_completed_node_ids(bo)
|
||
progress = get_progress_percentage(bo)
|
||
logs_count = StateFlowRecord.objects.filter(business_object=bo, is_cancelled=False).count()
|
||
all_logs = StateFlowRecord.objects.filter(business_object=bo, is_cancelled=False).values_list('state_id', 'state__name')
|
||
|
||
print(f"Current State: {current.name if current else 'None'} (ID: {current.id if current else 'N/A'})")
|
||
print(f"Completed Node IDs: {completed_ids}")
|
||
print(f"Completed Nodes: {[State.objects.get(id=cid).name for cid in completed_ids]}")
|
||
print(f"Progress: {progress:.2f}%")
|
||
print(f"Log Records Count: {logs_count}")
|
||
print(f"All Logs: {list(all_logs)}")
|
||
|
||
print(f"\n✓ 期望: Current='{state2.name}', Completed=[{state1.name}], Progress=33.33%, Logs=1")
|
||
|
||
# 验证
|
||
has_error = False
|
||
|
||
if current and current.id != state2.id:
|
||
print(f"\n❌ BUG FOUND! Current State 应该是 '{state2.name}' (ID: {state2.id}) 但实际是 '{current.name}' (ID: {current.id})")
|
||
has_error = True
|
||
|
||
if len(completed_ids) != 1:
|
||
print(f"\n❌ BUG FOUND! 应该完成1个节点，但实际完成了 {len(completed_ids)} 个")
|
||
has_error = True
|
||
|
||
if completed_ids and completed_ids[0] != state1.id:
|
||
print(f"\n❌ BUG FOUND! 完成的节点应该是 '{state1.name}' 但实际是 ID {completed_ids[0]}")
|
||
has_error = True
|
||
|
||
if logs_count != 1:
|
||
print(f"\n❌ BUG FOUND! 应该有1条日志记录，但实际有 {logs_count} 条")
|
||
has_error = True
|
||
|
||
if abs(progress - 33.33) > 0.1:
|
||
print(f"\n❌ BUG FOUND! 进度应该是 33.33% 但实际是 {progress:.2f}%")
|
||
has_error = True
|
||
|
||
if not has_error:
|
||
print("\n✅ 所有检查通过！没有发现bug")
|
||
|
||
# 清理
|
||
cleanup(bo, process, [state1, state2, state3])
|
||
|
||
return not has_error
|
||
|
||
def cleanup(bo, process, states):
|
||
"""清理测试数据"""
|
||
print(f"\n{'='*60}")
|
||
print("清理测试数据...")
|
||
bo.delete()
|
||
process.delete()
|
||
for state in states:
|
||
state.delete()
|
||
print("清理完成")
|
||
print(f"{'='*60}")
|
||
|
||
if __name__ == '__main__':
|
||
try:
|
||
result = main()
|
||
if result:
|
||
print("\n🎉 测试通过！")
|
||
exit(0)
|
||
else:
|
||
print("\n💥 测试失败！发现bug")
|
||
exit(1)
|
||
except Exception as e:
|
||
print(f"\n❌ 测试异常: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
exit(1)
|