""" Stateflow API 测试 """ 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 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) 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)