add unit tests

atdb/taskdatabase/tests/test_update_activity.py

@@ -9,17 +9,39 @@ class TestUpdateActivity(TestCase):
         self.workflow_requantisation.save()
 
         self.task1 = Task.objects.create(sas_id=12345,
+                                         filter="nv_19feb_2024",
                                          status='stored',
                                          workflow=self.workflow_requantisation,
                                          calculated_qualities={"per_task": "good", "per_sasid": "good"})
         self.task2 = Task.objects.create(sas_id=12345,
+                                         status='scrubbed',
+                                         workflow=self.workflow_requantisation,
+                                         calculated_qualities={"per_task": "good", "per_sasid": "good"})
+        self.task3 = Task.objects.create(sas_id=12345,
                                          status='archived',
                                          workflow=self.workflow_requantisation,
                                          archive={
-        "path_to_lta": "https://lta.lofar.eu//Lofar?project=ALL&mode=show_dataproducts_pipe&product=PulsarPipeline&pipeline_object_id=1101AB934B386BD5E063164A17AC38B9",
-        "lta_object_id": "1101AB934B386BD5E063164A17AC38B9",
-        "sas_id_archived": "1219995"
-    })
+                                            "path_to_lta": "https://lta.lofar.eu//Lofar?project=ALL&mode=show_dataproducts_pipe&product=PulsarPipeline&pipeline_object_id=1101AB934B386BD5E063164A17AC38B9",
+                                            "lta_object_id": "1101AB934B386BD5E063164A17AC38B9",
+                                            "sas_id_archived": "1219995"
+                                         },
+                                         size_to_process=1000,
+                                         size_processed=500)
+        self.task4 = Task.objects.create(sas_id=12345,
+                                         status='finished',
+                                         workflow=self.workflow_requantisation,
+                                         archive={
+                                             "path_to_lta": "https://lta.lofar.eu//Lofar?project=ALL&mode=show_dataproducts_pipe&product=PulsarPipeline&pipeline_object_id=1101AB934B386BD5E063164A17AC38B9",
+                                             "lta_object_id": "1101AB934B386BD5E063164A17AC38B9",
+                                             "sas_id_archived": "1219995"
+                                         },
+                                         size_to_process=1000,
+                                         size_processed=500)
+        self.task5 = Task.objects.create(sas_id=12345,
+                                         status='archived_failed',
+                                         workflow=self.workflow_requantisation,
+                                         size_to_process=1000,
+                                         size_processed=500)
 
     def test_created_activity(self):
 
@@ -29,6 +51,7 @@ class TestUpdateActivity(TestCase):
         actual = activity.sas_id
         self.assertEqual(actual, 12345)
 
+
     def test_stored(self):
 
         # test if the activity gets the calculated quality of the sas_id of the stored task
@@ -37,12 +60,62 @@ class TestUpdateActivity(TestCase):
         actual = activity.calculated_quality
         self.assertEqual(actual, "good")
 
-    def test_archived(self):
+
+    def test_scrubbed(self):
         # test if the activity gets the ingested_fraction of an archived task
         activity = self.task2.activity
 
         actual = activity.ingestq_status
-        self.assertEqual(actual, {'archived': 1})
+        self.assertEqual(actual, {'scrubbed': 1})
+
+        actual = activity.ingested_fraction
+        self.assertEqual(actual, 0)
+
+
+    def test_archived(self):
+        # test if the activity gets the ingested_fraction of an archived task
+        activity = self.task3.activity
+
+        actual = activity.ingestq_status
+        self.assertEqual(actual,  {'scrubbed': 1, 'archived': 1})
 
         actual = activity.archive['sas_id_archived']
         self.assertEqual(actual, "1219995")
+
+
+    def test_finished(self):
+        # test if the activity gets the ingested_fraction of an archived task
+        activity = self.task4.activity
+
+        actual = activity.ingestq_status
+        self.assertEqual(actual,  {'scrubbed': 1, 'archived': 1})
+
+        actual = activity.archive['sas_id_archived']
+        self.assertEqual(actual, "1219995")
+
+
+    def test_failed(self):
+        # test if the activity gets the ingested_fraction of an archived task
+        activity = self.task5.activity
+
+        actual = activity.finished_fraction
+        self.assertEqual(actual,  33)
+
+        actual = activity.total_size
+        self.assertEqual(actual, 3000)
+
+        actual = activity.remaining
+        self.assertEqual(actual, 2000)
+
+
+    def test_filter_and_workflow(self):
+
+        # test if the activity gets the calculated quality of the sas_id of the stored task
+        activity = self.task1.activity
+
+        actual = activity.filter
+        self.assertEqual(actual, "nv_19feb_2024")
+
+        actual = activity.workflow_id
+        self.assertEqual(actual, 22)
+