add unittests

atdb/taskdatabase/templates/taskdatabase/index.html

@@ -31,7 +31,7 @@
             {% include 'taskdatabase/pagination.html' %}
         </div>
     </div>
-    <p class="footer"> Version 29 Aug 2024
+    <p class="footer"> Version 30 Aug 2024
 </div>
 
 {% include 'taskdatabase/refresh.html' %}

atdb/taskdatabase/tests/test_services.py

+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 LatestMonitor
+from taskdatabase.views import ServiceHoldResume
+
+from unittest.mock import patch
+
+class TestServices(TestCase):
+    def setUp(self):
+        self.factory = RequestFactory()
+
+        self.user = User.objects.create_user(username='testuser', password='testpass')
+        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.save()
+
+    @patch('taskdatabase.views.LatestMonitor.objects.get')
+    def test_service_hold_resume(self, mock_get):
+        """
+        test ServiceHoldResume
+        """
+
+        # Arrange
+        mock_get.return_value = self.aggregator_service
+
+        request = self.factory.get('/dummy-url')
+        self._set_up_session(request)
+        request.user = self.user
+
+        # Act
+        response = ServiceHoldResume(request, "aggregator", "my_host", "False")
+        self.aggregator_service.refresh_from_db()
+
+        # Assert
+        enabled = self.aggregator_service.metadata['enabled']
+        self.assertEqual(enabled, "False")
+
+        # Check if it redirects to the correct URL
+        self.assertEqual(response.status_code, 302)
+        self.assertEqual(response.url, reverse('monitoring'))
\ No newline at end of file

atdb/taskdatabase/tests/test_views.py

@@ -4,10 +4,11 @@ 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
-from taskdatabase.views import SortTasks, TaskMultiStatus
-from taskdatabase.forms import DiscardAnnotationForm
+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):
@@ -19,8 +20,16 @@ class TestViews(TestCase):
         workflow_requantisation.save()
 
         # Create a test task
-        self.task1 = Task.objects.create(sas_id=456, status='defined', workflow=workflow_requantisation)
-        self.task2 = Task.objects.create(sas_id=123, status='defined', workflow=workflow_requantisation)
+        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')
 
@@ -29,6 +38,9 @@ class TestViews(TestCase):
             '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)
@@ -37,6 +49,9 @@ class TestViews(TestCase):
         request.session.save()
 
     def test_sort(self):
+        """
+        test SortTasks
+        """
         # Set up the URL for the view
 
         # Arrange
@@ -56,9 +71,54 @@ class TestViews(TestCase):
         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_discard(self, mock_is_valid, mock_get):
+    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
@@ -85,7 +145,10 @@ class TestViews(TestCase):
         self.assertEqual(response.url, reverse('query') + '?' + 'status=defined')
 
     @patch('taskdatabase.views.convert_query_params_to_url')
-    def test_task_multi_status_get_discard(self,mock_query_params):
+    def test_task_multi_status_get(self,mock_query_params):
+        """
+        test TaskMultiStatus (get)
+        """
 
         # Arrange
         mock_query_params.return_value = "&status=defined"
@@ -102,3 +165,242 @@ class TestViews(TestCase):
         # 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')