Skip to content
Snippets Groups Projects
Select Git revision
  • 15fd1747a4ebef815806c96797cd848d97436eae
  • master default protected
  • MAM-106-add-plots
  • MAM-90-executor-with-cwltool
  • MAM-88-change-status-names
  • MAM-50-specification-GUI
  • MAM-44-build-specification-functionality
  • MAM-42-ATDBspec-add-endpoints
  • MAM-41-ATDBspec-database-changes
  • bugfix-ingested-sizes
  • SDC-1663-inputs-validation
  • SDC-1649-compression-pipeline-quality-indicators
  • SDC-1635-monitoring-page
  • split-pipeline-for-gitlab
  • SDC-1590-database-changes
  • SDC-1580-adapt-configuration-to-multi-service
  • multi-edit
  • SDC-1552-datamodel-add-service-host
  • logentry_graph
  • SDC-1549-aggregation-task-ux
  • imaging-compression-pipeline-commissioning
21 results

test_views.py

Blame
  • 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')