forked from erp-dev/erp
617 lines
25 KiB
Python
617 lines
25 KiB
Python
"""
|
||
Stateflow API 测试
|
||
"""
|
||
from django.test import TestCase
|
||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||
from rest_framework.test import APIClient
|
||
from rest_framework import status
|
||
from django.contrib.auth import get_user_model
|
||
from stateflow import models
|
||
import tempfile
|
||
import os
|
||
|
||
User = get_user_model()
|
||
|
||
|
||
class StateAPITestCase(TestCase):
|
||
"""测试 State 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.param1 = models.StateParameter.objects.create(
|
||
key='param1', value='value1', description='参数1'
|
||
)
|
||
self.param2 = models.StateParameter.objects.create(
|
||
key='param2', value='value2', description='参数2'
|
||
)
|
||
|
||
# 创建基础状态供测试使用
|
||
self.state1 = models.State.objects.create(name='状态1', description='描述1')
|
||
self.state2 = models.State.objects.create(name='状态2', description='描述2')
|
||
|
||
def test_create_state(self):
|
||
"""测试创建状态"""
|
||
data = {
|
||
'name': '测试状态',
|
||
'description': '这是一个测试状态',
|
||
'parameter_ids': [self.param1.id, self.param2.id]
|
||
}
|
||
|
||
response = self.client.post('/api/v1/stateflow/states/', data, format='json')
|
||
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
||
self.assertEqual(response.data['name'], '测试状态')
|
||
|
||
# 验证参数已关联
|
||
state = models.State.objects.get(name='测试状态')
|
||
self.assertEqual(state.parameters.count(), 2)
|
||
|
||
def test_list_states(self):
|
||
"""测试获取状态列表"""
|
||
response = self.client.get('/api/v1/stateflow/states/?limit=10&offset=0')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['count'], 2)
|
||
self.assertEqual(len(response.data['results']), 2)
|
||
|
||
def test_retrieve_state(self):
|
||
"""测试获取状态详情"""
|
||
self.state1.parameters.add(self.param1)
|
||
|
||
response = self.client.get(f'/api/v1/stateflow/states/{self.state1.id}/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['name'], '状态1')
|
||
self.assertEqual(len(response.data['parameters']), 1)
|
||
|
||
def test_update_state(self):
|
||
"""测试更新状态"""
|
||
data = {
|
||
'name': '新名称',
|
||
'description': '新描述',
|
||
'parameter_ids': [self.param1.id]
|
||
}
|
||
|
||
response = self.client.put(f'/api/v1/stateflow/states/{self.state1.id}/', data, format='json')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
self.state1.refresh_from_db()
|
||
self.assertEqual(self.state1.name, '新名称')
|
||
self.assertEqual(self.state1.parameters.count(), 1)
|
||
|
||
def test_delete_state(self):
|
||
"""测试删除状态"""
|
||
response = self.client.delete(f'/api/v1/stateflow/states/{self.state2.id}/')
|
||
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
|
||
self.assertFalse(models.State.objects.filter(id=self.state2.id).exists())
|
||
|
||
def test_search_states(self):
|
||
"""测试搜索状态"""
|
||
# 创建特定名称的状态用于搜索测试
|
||
models.State.objects.create(name='审核状态', description='需要审核')
|
||
models.State.objects.create(name='发货状态', description='已经发货')
|
||
|
||
response = self.client.get('/api/v1/stateflow/states/?search=审核&limit=10&offset=0')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['count'], 1)
|
||
|
||
def test_create_state_parameter_with_attachment(self):
|
||
"""测试创建带附件的状态参数"""
|
||
# 创建一个测试文件
|
||
test_file = SimpleUploadedFile(
|
||
"test_doc.txt",
|
||
b"This is a test document content",
|
||
content_type="text/plain"
|
||
)
|
||
|
||
# 创建带附件的参数
|
||
param = models.StateParameter.objects.create(
|
||
key='document',
|
||
value='测试文档',
|
||
attachment=test_file,
|
||
description='这是一个测试文档'
|
||
)
|
||
|
||
# 创建状态并关联参数
|
||
state = models.State.objects.create(name='测试状态', description='带附件')
|
||
state.parameters.add(param)
|
||
|
||
# 验证附件已保存
|
||
self.assertIsNotNone(param.attachment)
|
||
self.assertIn('test_doc', param.attachment.name) # 文件名可能有哈希前缀
|
||
|
||
# 通过 API 获取状态详情，验证附件字段
|
||
response = self.client.get(f'/api/v1/stateflow/states/{state.id}/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
# 验证参数中包含附件信息
|
||
parameters = response.data['parameters']
|
||
self.assertEqual(len(parameters), 1)
|
||
self.assertIn('attachment', parameters[0])
|
||
self.assertIn('attachment_url', parameters[0])
|
||
self.assertIsNotNone(parameters[0]['attachment_url'])
|
||
|
||
# 清理测试文件
|
||
if param.attachment:
|
||
param.attachment.delete()
|
||
|
||
def test_state_parameter_without_attachment(self):
|
||
"""测试创建不带附件的状态参数"""
|
||
param = models.StateParameter.objects.create(
|
||
key='simple_param',
|
||
value='简单值',
|
||
description='简单参数'
|
||
)
|
||
state = models.State.objects.create(name='测试状态2', description='不带附件')
|
||
state.parameters.add(param)
|
||
|
||
# 验证附件字段为空
|
||
self.assertFalse(param.attachment)
|
||
|
||
# 通过 API 获取状态详情
|
||
response = self.client.get(f'/api/v1/stateflow/states/{state.id}/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
# 验证附件字段为 null
|
||
parameters = response.data['parameters']
|
||
self.assertEqual(len(parameters), 1)
|
||
self.assertIsNone(parameters[0]['attachment'])
|
||
self.assertIsNone(parameters[0]['attachment_url'])
|
||
|
||
def test_update_state_parameter_with_attachment(self):
|
||
"""测试更新状态参数时添加附件"""
|
||
# 创建初始参数（不带附件）
|
||
param = models.StateParameter.objects.create(
|
||
key='updatable_param',
|
||
value='初始值',
|
||
description='可更新参数'
|
||
)
|
||
state = models.State.objects.create(name='测试状态3', description='更新附件')
|
||
state.parameters.add(param)
|
||
|
||
# 验证初始无附件
|
||
self.assertFalse(param.attachment)
|
||
|
||
# 添加附件
|
||
test_file = SimpleUploadedFile(
|
||
"updated_doc.pdf",
|
||
b"Updated document content",
|
||
content_type="application/pdf"
|
||
)
|
||
param.attachment = test_file
|
||
param.save()
|
||
|
||
# 验证附件已添加
|
||
param.refresh_from_db()
|
||
self.assertIsNotNone(param.attachment)
|
||
self.assertIn('updated_doc', param.attachment.name) # 文件名可能有哈希前缀
|
||
|
||
# 通过 API 验证
|
||
response = self.client.get(f'/api/v1/stateflow/states/{state.id}/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
parameters = response.data['parameters']
|
||
self.assertIsNotNone(parameters[0]['attachment'])
|
||
self.assertIsNotNone(parameters[0]['attachment_url'])
|
||
|
||
# 清理
|
||
if param.attachment:
|
||
param.attachment.delete()
|
||
|
||
def test_state_parameter_attachment_url_format(self):
|
||
"""测试附件 URL 格式正确"""
|
||
test_file = SimpleUploadedFile(
|
||
"test_image.jpg",
|
||
b"fake image content",
|
||
content_type="image/jpeg"
|
||
)
|
||
|
||
param = models.StateParameter.objects.create(
|
||
key='image',
|
||
value='测试图片',
|
||
attachment=test_file
|
||
)
|
||
state = models.State.objects.create(name='图片状态', description='带图片')
|
||
state.parameters.add(param)
|
||
|
||
# 通过 API 获取
|
||
response = self.client.get(f'/api/v1/stateflow/states/{state.id}/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
parameters = response.data['parameters']
|
||
attachment_url = parameters[0]['attachment_url']
|
||
|
||
# 验证 URL 格式
|
||
self.assertIsNotNone(attachment_url)
|
||
self.assertIn('http', attachment_url) # 应该是完整 URL
|
||
self.assertIn('state_parameters/', attachment_url) # 包含上传路径
|
||
|
||
# 清理
|
||
if param.attachment:
|
||
param.attachment.delete()
|
||
|
||
def test_state_parameter_is_required(self):
|
||
"""测试参数必填字段"""
|
||
# 创建必填参数
|
||
required_param = models.StateParameter.objects.create(
|
||
key='required_param',
|
||
value='必填参数',
|
||
is_required=True
|
||
)
|
||
|
||
# 创建可选参数
|
||
optional_param = models.StateParameter.objects.create(
|
||
key='optional_param',
|
||
value='可选参数',
|
||
is_required=False
|
||
)
|
||
|
||
# 创建状态并关联参数
|
||
state = models.State.objects.create(name='测试状态', description='测试必填参数')
|
||
state.parameters.add(required_param, optional_param)
|
||
|
||
# 通过 API 验证
|
||
response = self.client.get(f'/api/v1/stateflow/states/{state.id}/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
parameters = response.data['parameters']
|
||
self.assertEqual(len(parameters), 2)
|
||
|
||
# 验证必填字段正确返回
|
||
required_params = [p for p in parameters if p['is_required']]
|
||
optional_params = [p for p in parameters if not p['is_required']]
|
||
|
||
self.assertEqual(len(required_params), 1)
|
||
self.assertEqual(len(optional_params), 1)
|
||
self.assertEqual(required_params[0]['key'], 'required_param')
|
||
|
||
def test_state_get_parameters_method(self):
|
||
"""测试 State.get_parameters() 方法"""
|
||
# 创建必填和可选参数
|
||
required_param = models.StateParameter.objects.create(
|
||
key='required1',
|
||
value='必填',
|
||
is_required=True
|
||
)
|
||
optional_param = models.StateParameter.objects.create(
|
||
key='optional1',
|
||
value='可选',
|
||
is_required=False
|
||
)
|
||
|
||
state = models.State.objects.create(name='测试状态')
|
||
state.parameters.add(required_param, optional_param)
|
||
|
||
# 测试获取所有参数
|
||
all_params = state.get_parameters(required_only=False)
|
||
self.assertEqual(len(all_params), 2)
|
||
|
||
# 测试只获取必填参数
|
||
required_params = state.get_parameters(required_only=True)
|
||
self.assertEqual(len(required_params), 1)
|
||
self.assertEqual(required_params[0].key, 'required1')
|
||
self.assertTrue(required_params[0].is_required)
|
||
|
||
def test_state_parameter_is_image_path(self):
|
||
"""测试参数图片路径字段"""
|
||
# 创建普通参数
|
||
normal_param = models.StateParameter.objects.create(
|
||
key='normal_param',
|
||
value='普通文本值',
|
||
is_image_path=False
|
||
)
|
||
|
||
# 创建图片路径参数
|
||
image_param = models.StateParameter.objects.create(
|
||
key='image_param',
|
||
value='https://example.com/image.jpg',
|
||
is_image_path=True
|
||
)
|
||
|
||
# 创建状态并关联参数
|
||
state = models.State.objects.create(name='测试图片参数', description='测试is_image_path字段')
|
||
state.parameters.add(normal_param, image_param)
|
||
|
||
# 通过 API 验证
|
||
response = self.client.get(f'/api/v1/stateflow/states/{state.id}/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
parameters = response.data['parameters']
|
||
self.assertEqual(len(parameters), 2)
|
||
|
||
# 验证 is_image_path 字段正确返回
|
||
image_params = [p for p in parameters if p['is_image_path']]
|
||
normal_params = [p for p in parameters if not p['is_image_path']]
|
||
|
||
self.assertEqual(len(image_params), 1)
|
||
self.assertEqual(len(normal_params), 1)
|
||
self.assertEqual(image_params[0]['key'], 'image_param')
|
||
self.assertEqual(image_params[0]['value'], 'https://example.com/image.jpg')
|
||
|
||
def test_create_parameter_with_image_path_flag(self):
|
||
"""测试创建带图片路径标记的参数"""
|
||
param = models.StateParameter.objects.create(
|
||
key='product_image',
|
||
value='https://cdn.example.com/products/abc123.png',
|
||
is_image_path=True,
|
||
description='产品图片URL'
|
||
)
|
||
|
||
# 验证字段已保存
|
||
self.assertTrue(param.is_image_path)
|
||
self.assertIn('https://', param.value)
|
||
|
||
# 创建状态并关联
|
||
state = models.State.objects.create(name='产品展示')
|
||
state.parameters.add(param)
|
||
|
||
# API 验证
|
||
response = self.client.get(f'/api/v1/stateflow/states/{state.id}/')
|
||
params = response.data['parameters']
|
||
|
||
self.assertEqual(len(params), 1)
|
||
self.assertTrue(params[0]['is_image_path'])
|
||
self.assertEqual(params[0]['key'], 'product_image')
|
||
|
||
|
||
class ProcessAPITestCase(TestCase):
|
||
"""测试 Process 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')
|
||
self.state2 = models.State.objects.create(name='状态2')
|
||
self.state3 = models.State.objects.create(name='状态3')
|
||
|
||
def test_create_process(self):
|
||
"""测试创建流程"""
|
||
data = {
|
||
'name': '测试流程',
|
||
'description': '这是一个测试流程',
|
||
'nodes': [
|
||
{'state_id': self.state1.id, 'order': 0},
|
||
{'state_id': self.state2.id, 'order': 1},
|
||
{'state_id': self.state3.id, 'order': 2},
|
||
]
|
||
}
|
||
|
||
response = self.client.post('/api/v1/stateflow/processes/', data, format='json')
|
||
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
||
self.assertEqual(response.data['name'], '测试流程')
|
||
|
||
# 验证节点已创建
|
||
process = models.Process.objects.get(name='测试流程')
|
||
self.assertEqual(process.process_nodes.count(), 3)
|
||
|
||
def test_list_processes(self):
|
||
"""测试获取流程列表"""
|
||
models.Process.objects.create(name='流程1', description='描述1')
|
||
models.Process.objects.create(name='流程2', description='描述2')
|
||
|
||
response = self.client.get('/api/v1/stateflow/processes/?limit=10&offset=0')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['count'], 2)
|
||
self.assertEqual(len(response.data['results']), 2)
|
||
|
||
def test_retrieve_process(self):
|
||
"""测试获取流程详情"""
|
||
process = models.Process.objects.create(name='测试流程', description='描述')
|
||
models.ProcessNode.objects.create(process=process, state=self.state1, order=0)
|
||
models.ProcessNode.objects.create(process=process, state=self.state2, order=1)
|
||
|
||
response = self.client.get(f'/api/v1/stateflow/processes/{process.id}/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['name'], '测试流程')
|
||
self.assertEqual(len(response.data['nodes']), 2)
|
||
|
||
def test_update_process(self):
|
||
"""测试更新流程"""
|
||
process = models.Process.objects.create(name='旧名称', description='旧描述')
|
||
models.ProcessNode.objects.create(process=process, state=self.state1, order=0)
|
||
|
||
data = {
|
||
'name': '新名称',
|
||
'description': '新描述',
|
||
'nodes': [
|
||
{'state_id': self.state2.id, 'order': 0},
|
||
{'state_id': self.state3.id, 'order': 1},
|
||
]
|
||
}
|
||
|
||
response = self.client.put(f'/api/v1/stateflow/processes/{process.id}/', data, format='json')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
process.refresh_from_db()
|
||
self.assertEqual(process.name, '新名称')
|
||
self.assertEqual(process.process_nodes.count(), 2)
|
||
|
||
def test_delete_process(self):
|
||
"""测试删除流程"""
|
||
process = models.Process.objects.create(name='待删除流程')
|
||
|
||
response = self.client.delete(f'/api/v1/stateflow/processes/{process.id}/')
|
||
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
|
||
self.assertFalse(models.Process.objects.filter(id=process.id).exists())
|
||
|
||
def test_search_processes(self):
|
||
"""测试搜索流程"""
|
||
models.Process.objects.create(name='订单流程', description='处理订单')
|
||
models.Process.objects.create(name='退货流程', description='处理退货')
|
||
|
||
response = self.client.get('/api/v1/stateflow/processes/?search=订单&limit=10&offset=0')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['count'], 1)
|
||
|
||
|
||
class BusinessObjectNewAPITestCase(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='第三个状态')
|
||
|
||
# 为状态1添加参数
|
||
self.param1 = models.StateParameter.objects.create(
|
||
key='param1', value='value1', is_required=True, description='必填参数'
|
||
)
|
||
self.param2 = models.StateParameter.objects.create(
|
||
key='param2', value='value2', 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_next_pending_state_api(self):
|
||
"""测试获取下一个待执行节点 API"""
|
||
# 未开始时应返回第一个节点
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/next_pending_state/'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['state']['id'], self.state1.id)
|
||
self.assertEqual(response.data['order'], 0)
|
||
self.assertIn('parameters', response.data)
|
||
self.assertEqual(len(response.data['parameters']), 2)
|
||
|
||
# 测试不包含参数
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/next_pending_state/?include_parameters=false'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertNotIn('parameters', response.data)
|
||
|
||
# 推进一步（提供必填参数）
|
||
self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{'parameters': {'param1': 'test_value'}},
|
||
format='json'
|
||
)
|
||
|
||
# 应返回第二个节点
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/next_pending_state/'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['state']['id'], self.state2.id)
|
||
|
||
# 完成所有节点
|
||
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/')
|
||
|
||
# 应返回 None
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/next_pending_state/'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertIsNone(response.data['next_state'])
|
||
|
||
def test_pending_states_api(self):
|
||
"""测试获取所有待执行节点 API"""
|
||
# 初始状态应返回所有节点
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/pending_states/'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['count'], 3)
|
||
self.assertEqual(len(response.data['pending_states']), 3)
|
||
|
||
# 验证不包含 parameters
|
||
self.assertNotIn('parameters', response.data['pending_states'][0])
|
||
|
||
# 推进一步（提供必填参数）
|
||
self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{'parameters': {'param1': 'test_value'}},
|
||
format='json'
|
||
)
|
||
|
||
# 应只剩两个待执行节点
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/pending_states/'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['count'], 2)
|
||
|
||
# 完成所有节点
|
||
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}/pending_states/'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['count'], 0)
|
||
|
||
def test_current_state_parameters_api(self):
|
||
"""测试获取当前状态（下一个待执行状态）参数 API"""
|
||
# 未开始时，current_state 为 state1（第一个待执行的），应返回 state1 的参数
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/current_state_parameters/'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['state']['id'], self.state1.id)
|
||
self.assertEqual(len(response.data['parameters']), 2)
|
||
|
||
# 推进（完成第一个状态），current_state 变为 state2（下一个待执行的状态）
|
||
self.client.post(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
|
||
{'parameters': {'param1': 'test_value'}},
|
||
format='json'
|
||
)
|
||
|
||
# 现在 current_state 是 state2（无参数）
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/business-objects/{self.business_object.id}/current_state_parameters/'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['state']['id'], self.state2.id)
|
||
self.assertEqual(response.data['count'], 0)
|
||
self.assertEqual(len(response.data['parameters']), 0)
|
||
|
||
def test_state_parameters_api(self):
|
||
"""测试 State 的参数 API"""
|
||
# 获取状态1的所有参数
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/states/{self.state1.id}/parameters/'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['state_id'], self.state1.id)
|
||
self.assertEqual(response.data['count'], 2)
|
||
self.assertEqual(len(response.data['parameters']), 2)
|
||
|
||
# 只获取必填参数
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/states/{self.state1.id}/parameters/?required_only=true'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['count'], 1)
|
||
|
||
# 测试没有参数的状态
|
||
response = self.client.get(
|
||
f'/api/v1/stateflow/states/{self.state2.id}/parameters/'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['count'], 0)
|