129 lines
4.7 KiB
Python
129 lines
4.7 KiB
Python
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')
|