forked from erp-dev/erp
529 lines
22 KiB
Python
529 lines
22 KiB
Python
"""
|
||
测试 BusinessObject 的完整 API 端点
|
||
包括 reset, timeline, add_parameters_to_log, get_log_parameters 等
|
||
"""
|
||
from django.test import TestCase
|
||
from rest_framework.test import APIClient
|
||
from rest_framework import status
|
||
from django.contrib.auth import get_user_model
|
||
from stateflow import models, services
|
||
|
||
User = get_user_model()
|
||
|
||
|
||
class BusinessObjectAPITestCase(TestCase):
|
||
"""测试 BusinessObject 的所有 API 端点"""
|
||
|
||
def setUp(self):
|
||
self.client = APIClient()
|
||
self.user = User.objects.create_user(username='testuser', password='testpass')
|
||
self.client.force_authenticate(user=self.user)
|
||
|
||
# 创建状态
|
||
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.param1 = models.StateParameter.objects.create(
|
||
key='temperature',
|
||
value='',
|
||
is_required=True,
|
||
description='温度'
|
||
)
|
||
self.param2 = models.StateParameter.objects.create(
|
||
key='humidity',
|
||
value='',
|
||
is_required=False,
|
||
description='湿度'
|
||
)
|
||
self.state1.parameters.add(self.param1, self.param2)
|
||
|
||
# 创建流程
|
||
self.process = models.Process.objects.create(name='测试流程')
|
||
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)
|
||
|
||
# 创建业务对象
|
||
self.business_object = models.BusinessObject.objects.create(
|
||
name='测试业务对象',
|
||
process=self.process
|
||
)
|
||
|
||
def test_reset_api(self):
|
||
"""测试重置进度 API"""
|
||
# 先推进两次
|
||
self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{'parameters': {'temperature': '25.5'}},
|
||
format='json'
|
||
)
|
||
self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/')
|
||
|
||
# 验证已有进度
|
||
self.business_object.refresh_from_db()
|
||
self.assertIsNotNone(services.get_business_object_current_state(self.business_object))
|
||
self.assertEqual(self.business_object.state_logs.count(), 2)
|
||
|
||
# 重置进度
|
||
response = self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/reset/')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertTrue(response.data['success'])
|
||
self.assertIn('重置', response.data['message'])
|
||
|
||
# 验证进度已重置（current_state 返回第一个待执行的节点）
|
||
self.business_object.refresh_from_db()
|
||
current_state = services.get_business_object_current_state(self.business_object)
|
||
self.assertEqual(current_state.name, self.state1.name)
|
||
self.assertEqual(self.business_object.state_logs.filter(is_cancelled=False).count(), 0)
|
||
|
||
def test_reset_api_not_started(self):
|
||
"""测试重置未开始的业务对象"""
|
||
# 未开始也可以重置（不会报错）
|
||
response = self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/reset/')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertTrue(response.data['success'])
|
||
|
||
def test_timeline_api(self):
|
||
"""测试状态时间线 API"""
|
||
# 未开始时的时间线
|
||
response = self.client.get(f'/api/v1/stateflow/business-objects/{self.business_object.id}/timeline/')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(len(response.data), 3)
|
||
|
||
# 验证所有状态都是待执行
|
||
for item in response.data:
|
||
self.assertEqual(item['status'], 'not_started')
|
||
self.assertIsNone(item['completed_at'])
|
||
self.assertIsNone(item['completed_by'])
|
||
self.assertFalse(item['is_cancelled'])
|
||
|
||
# 推进第一个状态
|
||
self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{'parameters': {'temperature': '25.5'}},
|
||
format='json'
|
||
)
|
||
|
||
# 再次获取时间线
|
||
response = self.client.get(f'/api/v1/stateflow/business-objects/{self.business_object.id}/timeline/')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(len(response.data), 3)
|
||
|
||
# 第一个状态应该是已完成
|
||
self.assertEqual(response.data[0]['status'], 'completed')
|
||
self.assertIsNotNone(response.data[0]['completed_at'])
|
||
self.assertEqual(response.data[0]['completed_by'], 'testuser')
|
||
self.assertFalse(response.data[0]['is_cancelled'])
|
||
|
||
# 第二个状态是进行中（下一个待执行的）
|
||
self.assertEqual(response.data[1]['status'], 'in_progress')
|
||
self.assertEqual(response.data[2]['status'], 'not_started')
|
||
|
||
def test_timeline_api_with_step_back(self):
|
||
"""测试时间线 API 包含回退后的已撤销状态"""
|
||
# 推进两次
|
||
self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{'parameters': {'temperature': '25.5'}},
|
||
format='json'
|
||
)
|
||
self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/')
|
||
|
||
# 回退一次
|
||
self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/step_back/')
|
||
|
||
# 获取时间线
|
||
response = self.client.get(f'/api/v1/stateflow/business-objects/{self.business_object.id}/timeline/')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
# 第一个状态：已完成且未撤销
|
||
self.assertEqual(response.data[0]['status'], 'completed')
|
||
self.assertFalse(response.data[0]['is_cancelled'])
|
||
|
||
# 第二个状态：已撤销
|
||
self.assertEqual(response.data[1]['status'], 'cancelled')
|
||
self.assertTrue(response.data[1]['is_cancelled'])
|
||
self.assertIsNotNone(response.data[1]['cancelled_at'])
|
||
|
||
# 第三个状态：待执行
|
||
self.assertEqual(response.data[2]['status'], 'not_started')
|
||
|
||
def test_add_parameters_to_log_api(self):
|
||
"""测试为状态流转记录补充参数 API"""
|
||
# 先推进一次
|
||
response = self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{'parameters': {'temperature': '25.5'}},
|
||
format='json'
|
||
)
|
||
|
||
state_log_id = response.data['state_log']['id']
|
||
|
||
# 补充参数
|
||
response = self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/{state_log_id}/add-parameters/',
|
||
{
|
||
'parameters': {
|
||
'humidity': '60%',
|
||
'operator': '张三'
|
||
},
|
||
'remark': '补充测量数据'
|
||
},
|
||
format='json'
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertTrue(response.data['success'])
|
||
self.assertIn('parameter_record', response.data)
|
||
|
||
# 验证参数记录
|
||
param_record = response.data['parameter_record']
|
||
self.assertEqual(param_record['parameters']['humidity'], '60%')
|
||
self.assertEqual(param_record['parameters']['operator'], '张三')
|
||
self.assertEqual(param_record['remark'], '补充测量数据')
|
||
|
||
def test_add_parameters_to_log_api_empty_parameters(self):
|
||
"""测试补充空参数应该失败"""
|
||
# 先推进一次
|
||
response = self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{'parameters': {'temperature': '25.5'}},
|
||
format='json'
|
||
)
|
||
|
||
state_log_id = response.data['state_log']['id']
|
||
|
||
# 尝试补充空参数
|
||
response = self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/{state_log_id}/add-parameters/',
|
||
{
|
||
'parameters': {},
|
||
'remark': '空参数测试'
|
||
},
|
||
format='json'
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||
self.assertFalse(response.data['success'])
|
||
self.assertIn('不能为空', response.data['message'])
|
||
|
||
def test_add_parameters_to_log_api_nonexistent_log(self):
|
||
"""测试为不存在的记录补充参数应该失败"""
|
||
response = self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/99999/add-parameters/',
|
||
{
|
||
'parameters': {'test': 'value'},
|
||
'remark': '测试'
|
||
},
|
||
format='json'
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
|
||
self.assertFalse(response.data['success'])
|
||
self.assertIn('不存在', response.data['message'])
|
||
|
||
def test_get_log_parameters_api_all_records(self):
|
||
"""测试获取状态流转记录的所有参数记录 API"""
|
||
# 推进并提供初始参数
|
||
response = self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{'parameters': {'temperature': '25.5', 'humidity': '60%'}},
|
||
format='json'
|
||
)
|
||
|
||
state_log_id = response.data['state_log']['id']
|
||
|
||
# 补充参数两次
|
||
self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/{state_log_id}/add-parameters/',
|
||
{
|
||
'parameters': {'temperature': '26.0'},
|
||
'remark': '第一次重测'
|
||
},
|
||
format='json'
|
||
)
|
||
self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/{state_log_id}/add-parameters/',
|
||
{
|
||
'parameters': {'temperature': '26.5'},
|
||
'remark': '第二次重测'
|
||
},
|
||
format='json'
|
||
)
|
||
|
||
# 获取所有参数记录
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/{state_log_id}/parameters/'
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['state_log_id'], state_log_id)
|
||
self.assertFalse(response.data['is_cancelled'])
|
||
self.assertEqual(response.data['count'], 3)
|
||
|
||
# 验证汇总数据
|
||
summary = response.data['summary']
|
||
self.assertEqual(summary['temperature'], '26.5') # 最新值
|
||
self.assertEqual(summary['humidity'], '60%')
|
||
|
||
# 验证记录列表
|
||
self.assertEqual(len(response.data['records']), 3)
|
||
|
||
def test_get_log_parameters_api_by_key(self):
|
||
"""测试获取指定参数的历史记录 API"""
|
||
# 推进并提供初始参数
|
||
response = self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{'parameters': {'temperature': '25.5'}},
|
||
format='json'
|
||
)
|
||
|
||
state_log_id = response.data['state_log']['id']
|
||
|
||
# 补充参数
|
||
self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/{state_log_id}/add-parameters/',
|
||
{
|
||
'parameters': {'temperature': '26.0', 'humidity': '65%'},
|
||
'remark': '重测'
|
||
},
|
||
format='json'
|
||
)
|
||
|
||
# 获取 temperature 的历史
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/{state_log_id}/parameters/?key=temperature'
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['key'], 'temperature')
|
||
self.assertEqual(len(response.data['history']), 2)
|
||
self.assertEqual(response.data['history'][0]['value'], '25.5')
|
||
self.assertEqual(response.data['history'][1]['value'], '26.0')
|
||
self.assertEqual(response.data['history'][1]['remark'], '重测')
|
||
|
||
def test_get_log_parameters_api_cancelled_state_excluded(self):
|
||
"""测试已撤销状态的参数默认不可见"""
|
||
# 推进
|
||
response = self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{'parameters': {'temperature': '25.5'}},
|
||
format='json'
|
||
)
|
||
|
||
state_log_id = response.data['state_log']['id']
|
||
|
||
# 回退（撤销状态）
|
||
self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/step_back/')
|
||
|
||
# 默认获取参数（应该返回空）
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/{state_log_id}/parameters/'
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertTrue(response.data['is_cancelled'])
|
||
self.assertEqual(len(response.data['summary']), 0) # 汇总应该为空
|
||
self.assertEqual(len(response.data['records']), 0) # 记录列表应该为空
|
||
|
||
def test_get_log_parameters_api_cancelled_state_with_include_flag(self):
|
||
"""测试 include_cancelled=true 可以查看已撤销状态的参数"""
|
||
# 推进
|
||
response = self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{'parameters': {'temperature': '25.5', 'humidity': '60%'}},
|
||
format='json'
|
||
)
|
||
|
||
state_log_id = response.data['state_log']['id']
|
||
|
||
# 回退（撤销状态）
|
||
self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/step_back/')
|
||
|
||
# 使用 include_cancelled=true 获取参数
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/{state_log_id}/parameters/?include_cancelled=true'
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertTrue(response.data['is_cancelled'])
|
||
self.assertEqual(len(response.data['summary']), 2) # 应该能看到参数
|
||
self.assertEqual(response.data['summary']['temperature'], '25.5')
|
||
self.assertEqual(response.data['summary']['humidity'], '60%')
|
||
|
||
def test_get_log_parameters_api_key_history_cancelled_excluded(self):
|
||
"""测试已撤销状态的参数历史默认不可见"""
|
||
# 推进
|
||
response = self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{'parameters': {'temperature': '25.5'}},
|
||
format='json'
|
||
)
|
||
|
||
state_log_id = response.data['state_log']['id']
|
||
|
||
# 补充参数
|
||
self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/{state_log_id}/add-parameters/',
|
||
{'parameters': {'temperature': '26.0'}, 'remark': '重测'},
|
||
format='json'
|
||
)
|
||
|
||
# 回退
|
||
self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/step_back/')
|
||
|
||
# 获取参数历史（默认应该为空）
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/{state_log_id}/parameters/?key=temperature'
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(len(response.data['history']), 0)
|
||
|
||
# 使用 include_cancelled=true 查看
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/{state_log_id}/parameters/?key=temperature&include_cancelled=true'
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(len(response.data['history']), 2)
|
||
|
||
def test_get_log_parameters_api_nonexistent_log(self):
|
||
"""测试获取不存在的记录的参数应该返回 404"""
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/99999/parameters/'
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
|
||
self.assertFalse(response.data['success'])
|
||
|
||
def test_advance_api_with_parameters(self):
|
||
"""测试推进 API 携带参数"""
|
||
response = self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{
|
||
'parameters': {
|
||
'temperature': '25.5',
|
||
'humidity': '60%',
|
||
'operator': '张三'
|
||
}
|
||
},
|
||
format='json'
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertTrue(response.data['success'])
|
||
self.assertIn('state_log', response.data)
|
||
|
||
# 验证参数摘要（API 返回的是 parameters_summary，不是 parameter_records）
|
||
state_log = response.data['state_log']
|
||
self.assertIn('parameters_summary', state_log)
|
||
|
||
params = state_log['parameters_summary']
|
||
self.assertEqual(params['temperature'], '25.5')
|
||
self.assertEqual(params['humidity'], '60%')
|
||
self.assertEqual(params['operator'], '张三')
|
||
|
||
def test_advance_api_missing_required_parameter(self):
|
||
"""测试推进 API 缺少必填参数应该失败"""
|
||
response = self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{
|
||
'parameters': {
|
||
'humidity': '60%' # 只提供可选参数，缺少必填的 temperature
|
||
}
|
||
},
|
||
format='json'
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||
self.assertFalse(response.data['success'])
|
||
self.assertIn('缺失必填参数', response.data['message'])
|
||
self.assertIn('temperature', response.data['message'])
|
||
|
||
def test_timeline_completed_all_states(self):
|
||
"""测试完成所有状态后的时间线"""
|
||
# 完成所有状态
|
||
self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{'parameters': {'temperature': '25.5'}},
|
||
format='json'
|
||
)
|
||
self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/')
|
||
self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/')
|
||
|
||
# 获取时间线
|
||
response = self.client.get(f'/api/v1/stateflow/business-objects/{self.business_object.id}/timeline/')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
# 所有状态都应该是已完成
|
||
for item in response.data:
|
||
self.assertEqual(item['status'], 'completed')
|
||
self.assertIsNotNone(item['completed_at'])
|
||
self.assertIsNotNone(item['completed_by'])
|
||
self.assertFalse(item['is_cancelled'])
|
||
|
||
def test_advance_when_already_completed(self):
|
||
"""测试流程已完成时尝试推进（预期错误）"""
|
||
# 完成所有状态
|
||
self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{'parameters': {'temperature': '25.5'}},
|
||
format='json'
|
||
)
|
||
self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/')
|
||
self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/')
|
||
|
||
# 验证已完成
|
||
self.assertEqual(services.get_overall_status(self.business_object), 'completed')
|
||
|
||
# 尝试再次推进（应该失败）
|
||
response = self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/'
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||
self.assertFalse(response.data['success'])
|
||
self.assertIn('流程已完成', response.data['message'])
|
||
self.assertIn('无法继续推进', response.data['message'])
|
||
|
||
def test_step_back_when_not_started(self):
|
||
"""测试未开始时尝试回退（预期错误）"""
|
||
# 验证未开始
|
||
self.assertEqual(services.get_overall_status(self.business_object), 'not_started')
|
||
|
||
# 尝试回退（应该失败）
|
||
response = self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/step_back/'
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||
self.assertFalse(response.data['success'])
|
||
self.assertIn('没有任何状态流转记录', response.data['message'])
|
||
|
||
def test_advance_process_without_nodes(self):
|
||
"""测试流程没有节点时尝试推进（预期错误）"""
|
||
# 创建一个没有节点的流程
|
||
empty_process = models.Process.objects.create(name='空流程')
|
||
empty_business_object = models.BusinessObject.objects.create(
|
||
name='空业务对象',
|
||
process=empty_process
|
||
)
|
||
|
||
# 尝试推进（应该失败）
|
||
response = self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{empty_business_object.id}/advance/'
|
||
)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||
self.assertFalse(response.data['success'])
|
||
self.assertIn('流程没有任何节点', response.data['message'])
|