Select Git revision
_hdf_file_writer.py
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
test_views.py 14.11 KiB
from django.test import TestCase, RequestFactory
from django.urls import reverse
from django.contrib.auth.models import User
from django.contrib.sessions.middleware import SessionMiddleware
from taskdatabase.models import Task, Workflow, LatestMonitor
from taskdatabase.views import SortTasks, TaskMultiStatus, TaskMultiHold, TaskMultiPurge, Hold, HoldQuery, PurgeQuery, \
TaskSetStatus, TaskRetry, TaskDiscard, TaskDiscardSasId, ServiceHoldResume, TaskValidateTask,TaskValidateSasId
from django.db.models.query import QuerySet
from unittest.mock import patch, MagicMock
class TestViews(TestCase):
def setUp(self):
self.factory = RequestFactory()
# Create a test workflow
workflow_requantisation = Workflow(workflow_uri="psrfits_requantisation")
workflow_requantisation.save()
# Create a test task
self.task1 = Task.objects.create(sas_id=123, status='defined', workflow=workflow_requantisation,
remarks = {"discard_reason" : "no longer needed"})
self.task2 = Task.objects.create(sas_id=456, status='defined', workflow=workflow_requantisation)
self.task3 = Task.objects.create(sas_id=789, status='processed', workflow=workflow_requantisation)
self.task4 = Task.objects.create(sas_id=123, status='stored', workflow=workflow_requantisation,
remarks = {"discard_reason" : "no longer needed"})
self.task5 = Task.objects.create(sas_id=555, status='stored', workflow=workflow_requantisation,
calculated_qualities = {"per_sasid" : "good"})
self.task6 = Task.objects.create(sas_id=555, status='stored', workflow=workflow_requantisation,
calculated_qualities = {"per_sasid" : "good"})
self.user = User.objects.create_user(username='testuser', password='testpass')
self.session_data = {
'filtered_tasks_as_list': [self.task1.id, self.task2.id],
'current_query_params': 'status=defined'
}
self.aggregator_service = LatestMonitor.objects.create(name="aggregator",hostname="my_host",metadata={"enabled":"True"})
def _set_up_session(self, request):
"""Helper function to set up session for the request"""
middleware = SessionMiddleware(get_response=lambda r: None)
middleware.process_request(request)
request.session.update(self.session_data)
request.session.save()
def test_sort(self):
"""
test SortTasks
"""
# Set up the URL for the view
# Arrange
# Create a request object
request = self.factory.get('/dummy-url')
self._set_up_session(request)
# Act
# Call the function with sort='priority' and redirect_to_page='tasks_list'
response = SortTasks(request, sort='sas_id', redirect_to_page='atdb')
# Assert
# Check if the sort field is correctly stored in the session
self.assertEqual(request.session['sort'], 'sas_id')
# Check if it redirects to the 'index' page
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse('index'))
def test_task_set_status(self):
"""
test TaskSetStatus
"""
# Arrange
request = self.factory.post('/dummy-url')
self._set_up_session(request)
request.user = self.user
# Act
response = TaskSetStatus(request, self.task3.id, new_status='finished', page = 0)
self.task3.refresh_from_db()
# Assert
# Check that the tasks were updated correctly
self.assertEqual(self.task3.status,'finished')
# check redirection
self.assertEqual(response.status_code, 302)
def test_task_retry(self):
"""
test TaskRetry
"""
# Arrange
request = self.factory.post('/dummy-url')
self._set_up_session(request)
request.user = self.user
# Act
response = TaskRetry(request, self.task3.id, new_status='processed', page = 0)
# Assert
# Check that the tasks were updated correctly
self.assertEqual(self.task3.status,'processed')
# check redirection
self.assertEqual(response.status_code, 302)
@patch('taskdatabase.views.Task.objects.get')
@patch('taskdatabase.views.DiscardAnnotationForm.is_valid')
def test_task_multi_status_post(self, mock_is_valid, mock_get):
"""
test TaskMultiStatus (post)
"""
# Arrange
# Mock the Task.objects.get method to return mock tasks
mock_get.side_effect = lambda id: self.task1 if id == self.task1.id else self.task2
mock_is_valid.return_value = True
request = self.factory.post('/dummy-url', data={'annotation': 'test annotation'})
self._set_up_session(request)
request.user = self.user
# Act
# Call the function with new_status='discarded'
response = TaskMultiStatus(request, new_status='discarded', query_params='status=defined')
# Assert
# Check that the tasks were updated correctly
self.assertEqual(self.task1.status, 'discarded')
self.assertEqual(self.task2.status, 'discarded')
self.assertEqual(self.task1.remarks['discard_reason'], 'test annotation')
self.assertEqual(self.task2.remarks['discard_reason'], 'test annotation')
# Check if it redirects to the correct URL
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse('query') + '?' + 'status=defined')
@patch('taskdatabase.views.convert_query_params_to_url')
def test_task_multi_status_get(self,mock_query_params):
"""
test TaskMultiStatus (get)
"""
# Arrange
mock_query_params.return_value = "&status=defined"
request = self.factory.get('/dummy-url')
self._set_up_session(request)
request.user = self.user
expected_params_on_session = "&status=defined"
# Act
# Call the function with new_status='discarded'
response = TaskMultiStatus(request, new_status='discarded', query_params='status=defined')
# Assert
self.assertEqual(response.status_code, 200)
self.assertEqual(request.session['current_query_params'], expected_params_on_session)
def test_task_hold(self):
"""
test Hold
"""
# Arrange
request = self.factory.post('/dummy-url')
self._set_up_session(request)
request.user = self.user
# Act
response = Hold(request, self.task1.id, hold_it='resume')
# Assert
# Check that the tasks were updated correctly
self.assertTrue(self.task1.resume)
# Check if it redirects to the correct URL
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse('query'))
@patch('taskdatabase.views.convert_query_params_to_url')
def test_task_hold_query(self, mock_query_params):
"""
test HoldQuery
"""
# Arrange
mock_query_params.return_value = "&status=defined"
request = self.factory.post('/dummy-url')
self._set_up_session(request)
request.user = self.user
# Act
response = HoldQuery(request, self.task1.id, hold_it='resume', query_params='status=defined')
# Assert
# Check that the tasks were updated correctly
self.assertTrue(self.task1.resume)
# check redirection
self.assertEqual(response.status_code, 302)
@patch('taskdatabase.views.Task.objects.get')
def test_task_multi_hold_post(self, mock_get):
"""
test TaskMultiHold
"""
# Arrange
# Mock the Task.objects.get method to return mock tasks
mock_get.side_effect = lambda id: self.task1 if id == self.task1.id else self.task2
request = self.factory.post('/dummy-url')
self._set_up_session(request)
request.user = self.user
# Act
response = TaskMultiHold(request, onhold='resume', query_params='status=defined')
# Assert
# Check that the tasks were updated correctly
self.assertTrue(self.task1.resume)
self.assertTrue(self.task2.resume)
self.assertEqual(response.status_code, 302)
@patch('taskdatabase.views.convert_query_params_to_url')
def test_task_purge_query(self, mock_query_params):
"""
test PurgeQuery
"""
# Arrange
mock_query_params.return_value = "&status=defined"
request = self.factory.post('/dummy-url')
self._set_up_session(request)
request.user = self.user
# Act
response = PurgeQuery(request, self.task1.id, purge_policy='no', query_params='status=defined')
# Assert
# Check that the tasks were updated correctly
self.assertEqual(self.task1.purge_policy,"no")
# check redirection
self.assertEqual(response.status_code, 302)
@patch('taskdatabase.views.Task.objects.get')
def test_task_multi_purge_post(self, mock_get):
"""
test TaskMultiPurge
"""
# Arrange
# Mock the Task.objects.get method to return mock tasks
mock_get.side_effect = lambda id: self.task1 if id == self.task1.id else self.task2
request = self.factory.post('/dummy-url')
self._set_up_session(request)
request.user = self.user
# Act
response = TaskMultiPurge(request, purge_policy='yes', query_params='status=defined')
# Assert
# Check that the tasks were updated correctly
self.assertEqual(self.task1.purge_policy,"yes")
self.assertEqual(self.task2.purge_policy,"yes")
self.assertEqual(response.status_code, 302)
@patch('taskdatabase.views.Task.objects.get')
@patch('taskdatabase.views.DiscardAnnotationForm.is_valid')
def test_task_discard_post(self, mock_is_valid, mock_get):
"""
test TaskDiscard
"""
# Arrange
# Mock the Task.objects.get method to return mock tasks
mock_get.return_value = self.task1
mock_is_valid.return_value = True
request = self.factory.post('/dummy-url', data={'annotation': 'test annotation'})
self._set_up_session(request)
request.user = self.user
# Act
# Call the function with new_status='discarded'
response = TaskDiscard(request, self.task1.id, new_status='discard', page=0)
# Assert
# Check that the tasks were updated correctly
self.assertEqual(self.task1.status, 'discard')
self.assertEqual(self.task1.remarks['discard_reason'], 'test annotation')
# Check if it redirects to the correct URL
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse('task-details'))
@patch('taskdatabase.views.Task.objects.get')
def test_task_discard_get(self, mock_get):
"""
test TaskDiscard
"""
# Arrange
mock_get.return_value = self.task1
request = self.factory.get('/dummy-url')
self._set_up_session(request)
request.user = self.user
# Act
# Call the function with new_status='discarded'
response = TaskDiscard(request, self.task1.id, new_status='discard', page=0)
# Assert
self.assertEqual(response.status_code, 200)
self.assertEqual(self.task1.remarks['discard_reason'], 'no longer needed')
@patch('taskdatabase.views.Task.objects.get')
@patch('taskdatabase.views.Task.objects.filter')
@patch('taskdatabase.views.DiscardAnnotationForm.is_valid')
def test_discard_sasid_post(self, mock_is_valid, mock_filter, mock_get):
"""
test TaskDiscardSasId
"""
# Arrange
# Mock the Task.objects.get method to return mock tasks
mock_get.return_value = self.task1
mock_filter.return_value = [self.task1,self.task4]
mock_is_valid.return_value = True
request = self.factory.post('/dummy-url')
self._set_up_session(request)
request.user = self.user
# Act
response = TaskDiscardSasId(request, self.task1.id, new_status='discard', page="0")
# Assert
self.assertEqual(self.task1.status, 'discard')
self.assertEqual(response.status_code, 302)
@patch('taskdatabase.views.Task.objects.get')
@patch('taskdatabase.views.Task.objects.filter')
def test_discard_sasid_get(self, mock_filter, mock_get):
"""
test TaskDiscardSasId
"""
# Arrange
# Mock the Task.objects.get method to return mock tasks
mock_get.return_value = self.task1
# mock the queryset, because tasks.count() is used in TaskDiscardSasId
mock_queryset = MagicMock(spec=QuerySet)
mock_filter.return_value = mock_queryset
mock_queryset.count.return_value = 2
mock_queryset.__iter__.return_value = iter([self.task1, self.task4])
request = self.factory.get('/dummy-url')
self._set_up_session(request)
request.user = self.user
# Act
response = TaskDiscardSasId(request, self.task1.id, new_status='discard', page="0")
# Assert
self.assertEqual(response.status_code, 200)
self.assertEqual(self.task1.remarks['discard_reason'], 'no longer needed')
@patch('taskdatabase.views.Task.objects.get')
@patch('taskdatabase.views.Task.objects.filter')
def test_validate_sasid(self, mock_filter, mock_get):
"""
test TaskValidatedSasId
"""
# Arrange
mock_get.return_value = self.task5
mock_filter.return_value = [self.task5, self.task6]
request = self.factory.get('/dummy-url')
self._set_up_session(request)
request.user = self.user
# Act
response = TaskValidateSasId(request, self.task5.id, quality='calculated', new_status="validated", page="0")
# Assert
self.assertEqual(self.task5.quality, 'good')
self.assertEqual(self.task6.quality, 'good')
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse('validation') + '?' + 'page=0')