forked from erp-dev/erp
321 lines
15 KiB
Python
321 lines
15 KiB
Python
"""
|
||
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):
|
||
"""测试初始状态 - 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)
|