Newer
Older
from django.test import TestCase
import json
from taskdatabase.models import Task, Workflow, Activity
from taskdatabase.services.common import State
class TestSummaryTasks(TestCase):
def setUp(self):
"""
initialize test data
"""
self.workflow_requantisation = Workflow(id=22, workflow_uri="psrfits_requantisation")
self.workflow_requantisation.save()
self.no_summary_task_77777 = Task.objects.create(sas_id=77777, new_status=State.DEFINED.value, workflow=self.workflow_requantisation,
outputs={"tar_archive": [{"size": 4885985280, "basename": "L621240_SAP002_B073_P000_bf.tar", "nameroot": "L621240_SAP002_B073_P000_bf"}]})
self.summary_task_defined_77777 = Task.objects.create(sas_id=77777, new_status=State.DEFINED.value, workflow=self.workflow_requantisation,
outputs={"tar_archive": [{"size": 4885985280, "basename": "L185619_summaryCS.tar", "nameroot": "L185619_summaryCS"}]})
self.summary_task_validated_77777 = Task.objects.create(sas_id=77777, new_status=State.VALIDATED.value, workflow=self.workflow_requantisation,
outputs={"tar_archive": [{"size": 4885985280, "basename": "L185619_summaryCS.tar", "nameroot": "L185619_summaryCS"}]})
self.summary_task_processed_77777 = Task.objects.create(sas_id=77777, new_status=State.PROCESSED.value, workflow=self.workflow_requantisation,
outputs={"tar_archive": [{"size": 4885985280, "basename": "L185619_summaryCS.tar", "nameroot": "L185619_summaryCS"}]})
# simulate an Activity that is processed, with 1 task already in STORED, the other one in PROCESSED
self.summary_task_stored_88888 = Task.objects.create(sas_id=88888, new_status=State.STORED.value, workflow=self.workflow_requantisation,
outputs={"tar_archive": [{"size": 4885985280, "basename": "L185619_summaryCS.tar", "nameroot": "L185619_summaryCS"}]})
self.summary_task_processed_88888 = Task.objects.create(sas_id=88888, new_status=State.PROCESSED.value, workflow=self.workflow_requantisation,
outputs={"tar_archive": [{"size": 4885985280, "basename": "L185619_summaryCS.tar", "nameroot": "L185619_summaryCS"}]})
def test_no_summary_task(self):
"""
test task that is not a summary task
"""
def test_task_defined_does_not_know_that_it_is_summary(self):
"""
test summary task, but before it knows that it becomes a summary task (which it only knows when 'processed')
"""
"""
test summary task, at 'stored' it should know that it is a summary task and return True)
"""
self.summary_task_validated_77777.save()
actual = self.summary_task_validated_77777.is_summary
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def test_task_processed_knows_that_it_is_summary(self):
"""
test summary task, at 'stored' it should know that it is a summary task and return True)
"""
self.summary_task_processed_77777.save()
actual = bool(self.summary_task_processed_77777.is_summary)
self.assertEqual(actual, True)
def test_summary_task_processed_goes_on_halt(self):
"""
test summary task, at 'stored' it should know that it is a summary task and return True)
"""
self.summary_task_processed_88888.save()
actual = self.summary_task_processed_88888.resume
self.assertEqual(actual, False)
def test_activity_77777_not_is_processed(self):
"""
activity 77777 should not be fully processed, because some tasks have not reached 'processed' yet
"""
self.summary_task_processed_77777.save()
activity = self.summary_task_processed_77777.activity
actual = bool(activity.is_processed)
self.assertFalse(actual)
def test_activity_88888_is_processed(self):
"""
SAS_ID 88888 should be is_processed, because all its tasks have status 'processed' or 'stored'
"""
self.summary_task_stored_88888.save()
self.summary_task_processed_88888.save()
activity = self.summary_task_processed_88888.activity
actual = activity.is_processed
self.assertTrue(actual)