创建中心任务的管理系统

tasks/tests/test_api.py

@@ -0,0 +1,128 @@
+from django.test import TestCase
+from django.urls import reverse
+from rest_framework import status
+from rest_framework.test import APIClient
+from tasks.models import Client, Task, TaskResult
+from tasks.tests.test_factories import ClientFactory, TaskFactory, TaskResultFactory
+
+class TaskAPITest(TestCase):
+    def setUp(self):
+        self.client = APIClient()
+        # Create a test client with token
+        self.test_client = ClientFactory()
+        self.token = self.test_client.token
+        # Authenticate the client
+        self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token}')
+    
+    def test_task_list(self):
+        """Test that authenticated users can list tasks"""
+        # Create some test tasks
+        TaskFactory.create_batch(3)
+        
+        url = reverse('task-list')
+        response = self.client.get(url)
+        
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(response.data), 3)
+    
+    def test_task_create(self):
+        """Test that authenticated users can create tasks"""
+        url = reverse('task-list')
+        data = {
+            'name': 'test_task',
+            'client_name': 'test_client',
+            'script': 'echo "Hello World"',
+            'timeout_seconds': 3600
+        }
+        
+        response = self.client.post(url, data, format='json')
+        
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(Task.objects.count(), 1)
+        self.assertEqual(Task.objects.get().name, 'test_task')
+    
+    def test_task_claim(self):
+        """Test that clients can claim available tasks"""
+        # Create a pending task
+        TaskFactory(client_name='test_client')
+        
+        url = reverse('task-claim')
+        data = {
+            'client_name': 'test_client'
+        }
+        
+        response = self.client.post(url, data, format='json')
+        
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data['status'], 'assigned')
+        self.assertEqual(response.data['assigned_to'], 'test_client')
+    
+    def test_unauthenticated_access(self):
+        """Test that unauthenticated users cannot access API endpoints"""
+        # Create a new client without authentication
+        unauth_client = APIClient()
+        
+        url = reverse('task-list')
+        response = unauth_client.get(url)
+        
+        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
+
+class ClientAPITest(TestCase):
+    def setUp(self):
+        self.client = APIClient()
+        # Create a test client with token
+        self.test_client = ClientFactory()
+        self.token = self.test_client.token
+        # Authenticate the client
+        self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token}')
+    
+    def test_client_list(self):
+        """Test that authenticated users can list clients"""
+        # Create some test clients
+        ClientFactory.create_batch(3)
+        
+        url = reverse('client-list')
+        response = self.client.get(url)
+        
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(response.data), 4)  # Including the test client
+    
+    def test_client_create(self):
+        """Test that authenticated users can create clients"""
+        url = reverse('client-list')
+        data = {
+            'name': 'new_client'
+        }
+        
+        response = self.client.post(url, data, format='json')
+        
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(Client.objects.count(), 2)  # Including the test client
+        self.assertEqual(Client.objects.get(id=response.data['id']).name, 'new_client')
+
+class TaskResultAPITest(TestCase):
+    def setUp(self):
+        self.client = APIClient()
+        # Create a test client with token
+        self.test_client = ClientFactory()
+        self.token = self.test_client.token
+        # Authenticate the client
+        self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token}')
+        # Create a test task
+        self.task = TaskFactory()
+    
+    def test_task_result_create(self):
+        """Test that authenticated users can create task results"""
+        url = reverse('taskresult-list')
+        data = {
+            'task': self.task.id,
+            'client': self.test_client.id,
+            'status': 'success',
+            'message': 'Task completed successfully'
+        }
+        
+        response = self.client.post(url, data, format='json')
+        
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(TaskResult.objects.count(), 1)
+        self.assertEqual(TaskResult.objects.get().status, 'success')