""" 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) def test_create_state(self): """测试创建状态""" data = { 'name': '测试状态', 'description': '这是一个测试状态', 'parameters': [ {'key': 'param1', 'value': 'value1', 'description': '参数1'}, {'key': 'param2', 'value': 'value2', 'description': '参数2'}, ] } 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): """测试获取状态列表""" models.State.objects.create(name='状态1', description='描述1') models.State.objects.create(name='状态2', description='描述2') 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): """测试获取状态详情""" state = models.State.objects.create(name='测试状态', description='描述') models.StateParameter.objects.create(state=state, key='key1', value='value1') response = self.client.get(f'/api/v1/stateflow/states/{state.id}/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['name'], '测试状态') self.assertEqual(len(response.data['parameters']), 1) def test_update_state(self): """测试更新状态""" state = models.State.objects.create(name='旧名称', description='旧描述') data = { 'name': '新名称', 'description': '新描述', 'parameters': [ {'key': 'new_param', 'value': 'new_value', 'description': '新参数'}, ] } response = self.client.put(f'/api/v1/stateflow/states/{state.id}/', data, format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) state.refresh_from_db() self.assertEqual(state.name, '新名称') self.assertEqual(state.parameters.count(), 1) def test_delete_state(self): """测试删除状态""" state = models.State.objects.create(name='待删除状态') response = self.client.delete(f'/api/v1/stateflow/states/{state.id}/') self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) self.assertFalse(models.State.objects.filter(id=state.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" ) # 先创建状态 state = models.State.objects.create(name='测试状态', description='带附件') # 创建带附件的参数 param = models.StateParameter.objects.create( state=state, key='document', value='测试文档', attachment=test_file, description='这是一个测试文档' ) # 验证附件已保存 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): """测试创建不带附件的状态参数""" state = models.State.objects.create(name='测试状态2', description='不带附件') param = models.StateParameter.objects.create( state=state, key='simple_param', value='简单值', description='简单参数' ) # 验证附件字段为空 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): """测试更新状态参数时添加附件""" # 创建初始状态和参数（不带附件） state = models.State.objects.create(name='测试状态3', description='更新附件') param = models.StateParameter.objects.create( state=state, key='updatable_param', value='初始值', description='可更新参数' ) # 验证初始无附件 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" ) state = models.State.objects.create(name='图片状态', description='带图片') param = models.StateParameter.objects.create( state=state, key='image', value='测试图片', attachment=test_file ) # 通过 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() 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)