diff --git a/atdb/taskdatabase/tests/test_filters.py b/atdb/taskdatabase/tests/test_filters.py
index 01bb7a7f372f2a58227acc81f71bfe9261ced775..5206de90d5b90d4325460a5e3f71e95e8e7c43db 100644
--- a/atdb/taskdatabase/tests/test_filters.py
+++ b/atdb/taskdatabase/tests/test_filters.py
@@ -1,10 +1,13 @@
from django.test import TestCase
from django.test import RequestFactory
+from django.db.models.query import QuerySet
from django.contrib.sessions.middleware import SessionMiddleware
+from unittest.mock import patch, MagicMock
+
from taskdatabase.models import Task, Workflow, Activity
from taskdatabase.views import get_filtered_tasks
-class FiltersTest(TestCase):
+class TestFilters(TestCase):
def setUp(self):
self.workflow_requantisation = Workflow(id=22, workflow_uri="psrfits_requantisation")
self.workflow_requantisation.save()
@@ -18,6 +21,13 @@ class FiltersTest(TestCase):
self.task4 = Task.objects.create(sas_id=66666, status='scrubbed', workflow=self.workflow_requantisation, activity = self.activity_66666)
self.task5 = Task.objects.create(sas_id=66666, status='archived_failed', workflow=self.workflow_requantisation, activity = self.activity_66666)
+ 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_without_filter(self):
count = 0
for task in Task.objects.all():
@@ -26,12 +36,67 @@ class FiltersTest(TestCase):
self.assertEqual(count,5)
def test_with_ingest_filter(self):
+ # Arrange
# create a request object
request = RequestFactory().get('/atdb/ingest')
# this simulates the 'Queue (scrubbed)' filter on the ingest page
request.session = {'ingest_filter': 'scrubbed'}
+ # Act
# after aggregating per sas_id, 2 objects with status 'scrubbed' remain
tasks = get_filtered_tasks(request, None, "sas_id")
+
+ # Assert
self.assertEqual(tasks.count(), 2)
+
+
+ def test_with_task_status_filter(self):
+ # Arrange
+ # create a request object
+ request = RequestFactory().get('/atdb')
+ request.session = {'task_filter': 'scrubbed'}
+
+ # Act
+ tasks = get_filtered_tasks(request, None, None)
+
+ # Assert
+ self.assertEqual(tasks.count(), 3)
+
+ def test_with_filtered_task_on_session(self):
+ """
+ test with pre_filtered tasks on the session
+ """
+
+ # Arrange
+ # create a request object
+ request = RequestFactory().get('/atdb')
+
+ filtered_tasks_as_list = [self.task1.id,self.task2.id,self.task3.id]
+ request.session = {'filtered_tasks_as_list': filtered_tasks_as_list}
+
+ # Act
+ tasks = get_filtered_tasks(request, None, None)
+
+ # Assert
+ self.assertEqual(tasks.count(), 3)
+
+
+ def test_with_prefiltered_task_as_parameter(self):
+ """
+ test with pre_filtered tasks as function parameter
+ """
+
+ # Arrange
+ # create a request object
+ request = RequestFactory().get('/atdb')
+ self._set_up_session(request)
+ pre_filtered_tasks = Task.objects.all()
+
+ # Act
+ # after aggregating per sas_id, 2 objects with status 'scrubbed' remain
+ tasks = get_filtered_tasks(request, pre_filtered_tasks, None)
+
+ # Assert
+ self.assertEqual(tasks.count(), 5)
+
diff --git a/atdb/taskdatabase/tests/test_views.py b/atdb/taskdatabase/tests/test_views.py
index 33595035b497ae7bb05275c38bfdcf020748597c..f27922bc6fa40db79a3c79c13a9b771f2eec4f90 100644
--- a/atdb/taskdatabase/tests/test_views.py
+++ b/atdb/taskdatabase/tests/test_views.py
@@ -3,10 +3,13 @@ from django.urls import reverse
from django.contrib.auth.models import User
from django.contrib.sessions.middleware import SessionMiddleware
+from django.contrib.messages.storage.fallback import FallbackStorage
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
+ TaskSetStatus, TaskRetry, TaskDiscard, TaskDiscardSasId, ServiceHoldResume, TaskValidateTask,TaskValidateSasId,\
+ convert_query_params_to_url, GetSizeView, ShowTaskQuality, AnnotateQualitySasId, AnnotateQualityTaskId, \
+ ClearAnnotationsSasID
from django.db.models.query import QuerySet
from unittest.mock import patch, MagicMock
@@ -21,13 +24,15 @@ class TestViews(TestCase):
# Create a test task
self.task1 = Task.objects.create(sas_id=123, status='defined', workflow=workflow_requantisation,
- remarks = {"discard_reason" : "no longer needed"})
+ remarks = {"discard_reason" : "no longer needed",
+ "quality_sasid" : "initial remark for sas_id",
+ "quality_taskid" : "initial remark for task_id"})
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"})
+ calculated_qualities = {"per_task" : "moderate","per_sasid" : "good"})
self.task6 = Task.objects.create(sas_id=555, status='stored', workflow=workflow_requantisation,
calculated_qualities = {"per_sasid" : "good"})
@@ -48,6 +53,14 @@ class TestViews(TestCase):
request.session.update(self.session_data)
request.session.save()
+ # Add messages framework to the request
+ request._messages = FallbackStorage(request)
+
+ def test_basic_conversion(self):
+ query_params = "<QueryDict: {'filter': ['nv_123'], 'status': ['stored'], 'sas_id': ['12345']}>"
+ expected_result = "&filter=nv_123&status=stored&sas_id=12345"
+ self.assertEqual(convert_query_params_to_url(query_params), expected_result)
+
def test_sort(self):
"""
test SortTasks
@@ -404,3 +417,193 @@ class TestViews(TestCase):
self.assertEqual(self.task6.quality, 'good')
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, reverse('validation') + '?' + 'page=0')
+
+
+ @patch('taskdatabase.views.Task.objects.get')
+ def test_validate_task(self, mock_get):
+ """
+ test TaskValidateTask
+ """
+
+ # Arrange
+ mock_get.return_value = self.task5
+
+ request = self.factory.get('/dummy-url')
+ self._set_up_session(request)
+ request.user = self.user
+
+ # Act
+ response = TaskValidateTask(request, self.task5.id, quality='calculated', new_status="validated", page="0")
+
+ # Assert
+ self.assertEqual(self.task5.quality, 'moderate')
+ self.assertEqual(response.status_code, 302)
+ self.assertEqual(response.url, reverse('quality') + '?' + 'page=0')
+
+
+ @patch('taskdatabase.views.algorithms.get_size') # Mock the get_size function
+ def test_get_size_view(self, mock_get_size):
+ # Setup
+ mock_get_size.return_value = 100
+
+ request = self.factory.get('/get-size/?status__in=processed,stored&type=to_process')
+ view = GetSizeView.as_view()
+
+ # Act
+ response = view(request)
+
+ # Assert
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.data, {'total_size': 100})
+ mock_get_size.assert_called_once_with(['processed', 'stored'], 'to_process')
+
+
+ @patch('taskdatabase.views.algorithms.convert_quality_to_html')
+ @patch('taskdatabase.views.algorithms.construct_inspectionplots')
+ def test_show_task_quality_with_valid_id(self, mock_construct_inspectionplots, mock_convert_quality_to_html):
+ """
+ test ShowTaskQuality
+ """
+
+ request = self.factory.get('/task_quality/')
+ self._set_up_session(request)
+
+ # Mocking the algorithms
+ mock_convert_quality_to_html.return_value = "<p>Quality HTML</p>"
+ mock_construct_inspectionplots.return_value = "<p>Plots HTML</p>"
+
+ response = ShowTaskQuality(request, id=self.task1.id)
+
+ # Asserts
+ self.assertEqual(response.status_code, 200)
+ self.assertIn('Quality HTML', response.content.decode())
+ self.assertIn('Plots HTML', response.content.decode())
+ self.assertEqual(request.session['task_id'], self.task1.id)
+
+ def test_show_task_quality_with_invalid_id_no_task_in_session(self):
+ """
+ test ShowTaskQuality
+ """
+ request = self.factory.get('/task_quality/')
+ self._set_up_session(request)
+
+ response = ShowTaskQuality(request, id=0)
+
+ # Asserts
+ self.assertEqual(response.status_code, 302)
+ self.assertEqual(response.url, '/atdb/') # Assuming 'index' redirects to '/'
+ self.assertEqual(len(request._messages), 1)
+
+
+ @patch('taskdatabase.views.Task.objects.get')
+ @patch('taskdatabase.views.Task.objects.filter')
+ @patch('taskdatabase.views.QualityAnnotationForm.is_valid')
+ def test_annotate_quality_sasid_post(self, mock_is_valid, mock_filter, mock_get):
+ """
+ test AnnotateQualitySasID (post)
+ """
+
+ # 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
+
+ post_data = {'annotation': 'quality annotation','return_to_page' : 0}
+ request = self.factory.post('/dummy-url', data=post_data)
+ self._set_up_session(request)
+
+ # Act
+ response = AnnotateQualitySasId(request, self.task1.id)
+ self.task1.refresh_from_db()
+
+ # Assert
+ self.assertEqual(self.task1.remarks['quality_sasid'], 'quality annotation')
+ self.assertEqual(response.status_code, 302)
+
+ @patch('taskdatabase.views.Task.objects.get')
+ def test_annotate_quality_sasid_get(self, mock_get):
+ """
+ test AnnotateQualitySasID (get)
+ """
+
+ # Arrange
+ # Mock the Task.objects.get method to return mock tasks
+ mock_get.return_value = self.task1
+
+ request = self.factory.get('/dummy-url')
+ self._set_up_session(request)
+ request.user = self.user
+
+ # Act
+ response = AnnotateQualitySasId(request, self.task1.id)
+
+ # Assert
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(self.task1.remarks['quality_sasid'], 'initial remark for sas_id')
+
+ @patch('taskdatabase.views.Task.objects.get')
+ @patch('taskdatabase.views.QualityAnnotationForm.is_valid')
+ def test_annotate_quality_taskid_post(self, mock_is_valid, mock_get):
+ """
+ test AnnotateQualityTaskId (post)
+ """
+
+ # Arrange
+ # Mock the Task.objects.get method to return mock tasks
+ mock_get.return_value = self.task1
+ mock_is_valid.return_value = True
+
+ post_data = {'annotation': 'quality annotation','return_to_page' : 0}
+ request = self.factory.post('/dummy-url', data=post_data)
+ self._set_up_session(request)
+
+ # Act
+ response = AnnotateQualityTaskId(request, self.task1.id)
+ self.task1.refresh_from_db()
+
+ # Assert
+ self.assertEqual(self.task1.remarks['quality_taskid'], 'quality annotation')
+ self.assertEqual(response.status_code, 302)
+
+ @patch('taskdatabase.views.Task.objects.get')
+ def test_annotate_quality_taskid_get(self, mock_get):
+ """
+ test AnnotateQualityTaskId (get)
+ """
+
+ # Arrange
+ # Mock the Task.objects.get method to return mock tasks
+ mock_get.return_value = self.task1
+
+ request = self.factory.get('/dummy-url')
+ self._set_up_session(request)
+ request.user = self.user
+
+ # Act
+ response = AnnotateQualityTaskId(request, self.task1.id)
+
+ # Assert
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(self.task1.remarks['quality_taskid'], 'initial remark for task_id')
+
+ @patch('taskdatabase.models.Task.objects.get')
+ @patch('taskdatabase.models.Task.objects.filter')
+ def test_clear_annotations_sas_id(self, mock_filter, mock_get):
+ # Arrange
+ # Mock the queryset returned by Task.objects.filter
+ mock_get.return_value = self.task1
+ mock_filter.return_value = [self.task1, self.task2]
+
+ request = MagicMock()
+
+ # Act
+ response = ClearAnnotationsSasID(request, id=self.task1.id)
+
+ # Asserts
+ self.assertEqual(response.status_code, 302) # Redirect status code
+ self.assertEqual(response.url, reverse('validation'))
+
+ # Verify that the remarks were cleared
+ self.assertIsNone(self.task1.remarks['quality_sasid'])
+ self.assertIsNone(self.task2.remarks['quality_sasid'])
diff --git a/atdb/taskdatabase/tests/test_views_details_page.py b/atdb/taskdatabase/tests/test_views_details_page.py
index fec78527b081ac1c9e5fafca954f2fa2f967e84f..ef3145104b328fb787645bf372d1371f41a9042e 100644
--- a/atdb/taskdatabase/tests/test_views_details_page.py
+++ b/atdb/taskdatabase/tests/test_views_details_page.py
@@ -1,27 +1,93 @@
-from django.test import TestCase
+from django.test import TestCase, RequestFactory
from django.urls import reverse
+from unittest.mock import patch, MagicMock
from taskdatabase.models import Task, Workflow
-class DetailsPageViewTest(TestCase):
+from taskdatabase.views import ShowInspectionPlots, ShowInspectionPlotsSasId,ShowSummarySasId
+
+class TestDetailsPage(TestCase):
@classmethod
- def setUp(cls):
+ def setUp(self):
+ self.factory = RequestFactory()
+
# Set up non-modified objects used by all test methods
workflow = Workflow()
workflow.save()
# create a list of Tasks
- Task.objects.get_or_create(sas_id=12345, status='finished', workflow=workflow)
+ self.task1 = Task.objects.create(sas_id=12345, status='finished', workflow=workflow)
def test_url_exists_at_desired_location(self):
# extract the task
- task = Task.objects.all()[0]
- response = self.client.get(f'/atdb/task_details/{task.id}/1')
+ response = self.client.get(f'/atdb/task_details/{self.task1.id}/1')
self.assertEqual(response.status_code, 200)
def test_uses_correct_template(self):
- task = Task.objects.all()[0]
- response = self.client.get(f'/atdb/task_details/{task.id}/1')
+ response = self.client.get(f'/atdb/task_details/{self.task1.id}/1')
+ self.assertEqual(response.status_code, 200)
+ self.assertTemplateUsed(response, 'taskdatabase/tasks/task_details.html')
+
+ def test_show_inputs(self):
+ response = self.client.get(f'/atdb/show-inputs/{self.task1.id}/')
+ self.assertEqual(response.status_code, 200)
+ self.assertTemplateUsed(response, 'taskdatabase/details/inputs.html')
+
+ def test_show_outputs(self):
+ response = self.client.get(f'/atdb/show-outputs/{self.task1.id}/')
+ self.assertEqual(response.status_code, 200)
+ self.assertTemplateUsed(response, 'taskdatabase/details/outputs.html')
+
+ def test_show_metrics(self):
+ response = self.client.get(f'/atdb/show-metrics/{self.task1.id}/')
+ self.assertEqual(response.status_code, 200)
+ self.assertTemplateUsed(response, 'taskdatabase/details/metrics.html')
+
+ @patch('taskdatabase.views.algorithms.construct_inspectionplots')
+ def test_show_inspectionplots(self,mock_plots):
+ # Arrange
+ mock_plots.return_value = "<p>Inspection Plots HTML</p>"
+ # Simulate a GET request to the view
+ request = self.factory.get('/show_inspectionplots/')
+
+ # Act
+ response = ShowInspectionPlots(request, id=self.task1.id)
+
+ # Assert
self.assertEqual(response.status_code, 200)
- self.assertTemplateUsed(response, 'taskdatabase/tasks/task_details.html')
\ No newline at end of file
+
+ # Assert that the my_plots context contains the mocked HTML
+ self.assertIn('Inspection Plots HTML', response.content.decode())
+
+ @patch('taskdatabase.views.algorithms.construct_inspectionplots')
+ def test_show_inspectionplots_sasid(self,mock_plots):
+ # Arrange
+ mock_plots.return_value = "<p>Inspection Plots HTML</p>"
+ # Simulate a GET request to the view
+ request = self.factory.get('/show_inspectionplots_sasid/')
+
+ # Act
+ response = ShowInspectionPlotsSasId(request, id=self.task1.id)
+
+ # Assert
+ self.assertEqual(response.status_code, 200)
+
+ # Assert that the my_plots context contains the mocked HTML
+ self.assertIn('Inspection Plots HTML', response.content.decode())
+
+ @patch('taskdatabase.views.algorithms.construct_summary')
+ def test_show_summary_sasid(self,mock_summary):
+ # Arrange
+ mock_summary.return_value = "<p>Summary HTML</p>"
+ # Simulate a GET request to the view
+ request = self.factory.get('/show_summary/')
+
+ # Act
+ response = ShowSummarySasId(request, id=self.task1.id)
+
+ # Assert
+ self.assertEqual(response.status_code, 200)
+
+ # Assert that the my_plots context contains the mocked HTML
+ self.assertIn('<p>Summary HTML</p>', response.content.decode())
\ No newline at end of file
diff --git a/atdb/taskdatabase/views.py b/atdb/taskdatabase/views.py
index e572fd97cf6f743a2e87ed6e400c1c732c32a0e8..4445d4a913b5593332dd10644ac364fc3bcec271 100644
--- a/atdb/taskdatabase/views.py
+++ b/atdb/taskdatabase/views.py
@@ -1851,7 +1851,7 @@ def UpdateSummaryFlag(request, task_id): # pragma: no cover
-def CreateStatusGraph(request):
+def CreateStatusGraph(request): # pragma: no cover
query_per_hour = """
SELECT
DATE_TRUNC('hour', timestamp) AS hour,