t_tmssapp_scheduling_django_API.py

#!/usr/bin/env python3

# Copyright (C) 2018    ASTRON (Netherlands Institute for Radio Astronomy)
# P.O. Box 2, 7990 AA Dwingeloo, The Netherlands
#
# This file is part of the LOFAR software suite.
# The LOFAR software suite is free software: you can redistribute it and/or
# modify it under the terms of the GNU General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# The LOFAR software suite is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.    See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.

# $Id:  $

import os
import unittest
from datetime import datetime
# use this to create timezone-aware datetime objects: from django.utils import timezone

import logging
logger = logging.getLogger(__name__)
logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO)

from lofar.common.test_utils import exit_with_skipped_code_if_skip_integration_tests
exit_with_skipped_code_if_skip_integration_tests()

# todo: Tags? -> Decide how to deal with them first.
# todo: Immutability of Blueprints on db level?

# Do Mandatory setup:
# use setup/teardown magic for tmss test database
# (ignore pycharm unused import statement, python unittests does use at RunTime the tmss_database_unittest_setup module)
from lofar.sas.tmss.test.tmss_database_unittest_setup import *

from lofar.sas.tmss.test.tmss_test_data_django_models import *

from django.db.utils import IntegrityError
from django.core.exceptions import ValidationError

class SubtaskTemplateTest(unittest.TestCase):
    def test_SubtaskTemplate_gets_created_with_correct_creation_timestamp(self):

        # setup
        before = datetime.utcnow()
        entry = models.SubtaskTemplate.objects.create(**SubtaskTemplate_test_data())

        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.created_at)
        self.assertGreater(after, entry.created_at)

    def test_SubtaskTemplate_update_timestamp_gets_changed_correctly(self):

        # setup
        entry = models.SubtaskTemplate.objects.create(**SubtaskTemplate_test_data())
        before = datetime.utcnow()
        entry.save()
        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.updated_at)
        self.assertGreater(after, entry.updated_at)


class DataproductSpecificationsTemplateTest(unittest.TestCase):

    def test_DataproductSpecificationsTemplate_gets_created_with_correct_creation_timestamp(self):

        # setup
        before = datetime.utcnow()
        entry = models.DataproductSpecificationsTemplate.objects.create(**DataproductSpecificationsTemplate_test_data())

        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.created_at)
        self.assertGreater(after, entry.created_at)

    def test_DataproductSpecificationsTemplate_update_timestamp_gets_changed_correctly(self):

        # setup
        entry = models.DataproductSpecificationsTemplate.objects.create(**DataproductSpecificationsTemplate_test_data())
        before = datetime.utcnow()
        entry.save()
        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.updated_at)
        self.assertGreater(after, entry.updated_at)


class DataproductFeedbackTemplateTest(unittest.TestCase):
    # This currently adds nothing on top of the template base class, so nothing new to test here.
    pass


class SubtaskOutputTest(unittest.TestCase):
    def test_SubtaskOutput_gets_created_with_correct_creation_timestamp(self):

        # setup
        before = datetime.utcnow()
        entry = models.SubtaskOutput.objects.create(**SubtaskOutput_test_data())

        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.created_at)
        self.assertGreater(after, entry.created_at)

    def test_SubtaskOutput_update_timestamp_gets_changed_correctly(self):

        # setup
        entry = models.SubtaskOutput.objects.create(**SubtaskOutput_test_data())
        before = datetime.utcnow()
        entry.save()
        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.updated_at)
        self.assertGreater(after, entry.updated_at)

    def test_SubtaskOutput_prevents_missing_subtask(self):

        # setup
        test_data = dict(SubtaskOutput_test_data())
        test_data['subtask'] = None
        test_data['task_blueprint'] = None

        # assert
        with self.assertRaises(IntegrityError):
            models.SubtaskOutput.objects.create(**test_data)


class SubtaskInputTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls.subtask = models.Subtask.objects.create(**Subtask_test_data())
        cls.producer = models.SubtaskOutput.objects.create(**SubtaskOutput_test_data())

    def test_SubtaskInput_gets_created_with_correct_creation_timestamp(self):

        # setup
        before = datetime.utcnow()
        entry = models.SubtaskInput.objects.create(**SubtaskInput_test_data(subtask=self.subtask, producer=self.producer))

        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.created_at)
        self.assertGreater(after, entry.created_at)

    def test_SubtaskInput_update_timestamp_gets_changed_correctly(self):

        # setup
        entry = models.SubtaskInput.objects.create(**SubtaskInput_test_data(subtask=self.subtask, producer=self.producer))
        before = datetime.utcnow()
        entry.save()
        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.updated_at)
        self.assertGreater(after, entry.updated_at)

    def test_SubtaskInput_prevents_missing_subtask(self):

        # setup
        test_data = dict(SubtaskInput_test_data(subtask=self.subtask, producer=self.producer))
        test_data['subtask'] = None

        # assert
        with self.assertRaises(IntegrityError):
            models.SubtaskInput.objects.create(**test_data)


class SubtaskTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls.task_blueprint = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data())

    def test_Subtask_gets_created_with_correct_creation_timestamp(self):

        # setup
        before = datetime.utcnow()
        entry = models.Subtask.objects.create(**Subtask_test_data())
        entry.task_blueprints.set([self.task_blueprint])
        entry.save()

        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.created_at)
        self.assertGreater(after, entry.created_at)

    def test_Subtask_update_timestamp_gets_changed_correctly(self):

        # setup
        entry = models.Subtask.objects.create(**Subtask_test_data())
        entry.task_blueprints.set([self.task_blueprint])
        before = datetime.utcnow()
        entry.save()
        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.updated_at)
        self.assertGreater(after, entry.updated_at)

    def test_Subtask_prevents_missing_template(self):

        # setup
        test_data = dict(Subtask_test_data())
        test_data['specifications_template'] = None

        # assert
        with self.assertRaises(IntegrityError):
            models.Subtask.objects.create(**test_data)

    def test_Subtask_predecessors_and_successors_none(self):
        subtask1:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())
        subtask2:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())


        self.assertEqual(set(), set(subtask1.predecessors.all()))
        self.assertEqual(set(), set(subtask2.predecessors.all()))
        self.assertEqual(set(), set(subtask1.successors.all()))
        self.assertEqual(set(), set(subtask2.successors.all()))

    def test_Subtask_predecessors_and_successors_simple(self):
        subtask1:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())
        subtask1.task_blueprints.set([self.task_blueprint])
        subtask1.save()
        subtask2:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())
        subtask2.task_blueprints.set([self.task_blueprint])
        subtask2.save()

        output1 = models.SubtaskOutput.objects.create(subtask=subtask1, task_blueprint=self.task_blueprint)
        models.SubtaskInput.objects.create(**SubtaskInput_test_data(subtask=subtask2, producer=output1))

        self.assertEqual(subtask1, subtask2.predecessors.all()[0])
        self.assertEqual(subtask2, subtask1.successors.all()[0])

    def test_Subtask_predecessors_and_successors_complex(self):
        subtask1:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())
        subtask2:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())
        subtask2.task_blueprints.set(subtask1.task_blueprints.all())
        subtask2.save()
        subtask3:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())
        subtask3.task_blueprints.set(subtask1.task_blueprints.all())
        subtask3.save()
        subtask4:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())
        subtask4.task_blueprints.set(subtask1.task_blueprints.all())
        subtask4.save()
        subtask5:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())
        subtask5.task_blueprints.set(subtask1.task_blueprints.all())
        subtask5.save()
        subtask6:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())
        subtask6.task_blueprints.set(subtask1.task_blueprints.all())
        subtask6.save()

        # ST1 ---> ST3 ---> ST4
        #      |        |
        # ST2 -          -> ST5 ---> ST6

        output1 = models.SubtaskOutput.objects.create(subtask=subtask1, task_blueprint=self.task_blueprint)
        output2 = models.SubtaskOutput.objects.create(subtask=subtask2, task_blueprint=self.task_blueprint)
        output3 = models.SubtaskOutput.objects.create(subtask=subtask3, task_blueprint=self.task_blueprint)
        output4 = models.SubtaskOutput.objects.create(subtask=subtask4, task_blueprint=self.task_blueprint)
        output5 = models.SubtaskOutput.objects.create(subtask=subtask5, task_blueprint=self.task_blueprint)
        output6 = models.SubtaskOutput.objects.create(subtask=subtask6, task_blueprint=self.task_blueprint)

        models.SubtaskInput.objects.create(**SubtaskInput_test_data(subtask=subtask3, producer=output1))
        models.SubtaskInput.objects.create(**SubtaskInput_test_data(subtask=subtask3, producer=output2))
        models.SubtaskInput.objects.create(**SubtaskInput_test_data(subtask=subtask4, producer=output3))
        models.SubtaskInput.objects.create(**SubtaskInput_test_data(subtask=subtask5, producer=output3))
        models.SubtaskInput.objects.create(**SubtaskInput_test_data(subtask=subtask6, producer=output5))

        self.assertEqual(set((subtask1, subtask2)), set(subtask3.predecessors.all()))
        self.assertEqual(set((subtask4, subtask5)), set(subtask3.successors.all()))
        self.assertEqual(set((subtask3,)), set(subtask4.predecessors.all()))
        self.assertEqual(set((subtask3,)), set(subtask5.predecessors.all()))
        self.assertEqual(set((subtask3,)), set(subtask1.successors.all()))
        self.assertEqual(set((subtask3,)), set(subtask2.successors.all()))
        self.assertEqual(set(), set(subtask1.predecessors.all()))
        self.assertEqual(set(), set(subtask2.predecessors.all()))
        self.assertEqual(set(), set(subtask4.successors.all()))
        self.assertEqual(set((subtask6,)), set(subtask5.successors.all()))

    def test_Subtask_transformed_dataproducts(self):
        # setup
        subtask1:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())
        output1:models.SubtaskOutput = models.SubtaskOutput.objects.create(subtask=subtask1,
                                                                           task_blueprint=self.task_blueprint)
        output1_dp:models.Dataproduct = models.Dataproduct.objects.create(**Dataproduct_test_data(producer=output1))

        subtask2:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())
        input2:models.SubtaskInput = models.SubtaskInput.objects.create(**SubtaskInput_test_data(subtask=subtask2, producer=output1))
        input2_dp = output1_dp
        input2.dataproducts.set([input2_dp])
        input2.save()
        output2:models.SubtaskOutput = models.SubtaskOutput.objects.create(subtask=subtask2,
                                                                           task_blueprint=self.task_blueprint)
        output2_dp:models.Dataproduct = models.Dataproduct.objects.create(**Dataproduct_test_data(producer=output2))

        models.DataproductTransform.objects.create(input=input2_dp, output=output2_dp, identity=True)

        subtask1.refresh_from_db()
        subtask2.refresh_from_db()

        # make sure the subtask and input/output dp's are setup correct...
        self.assertEqual(1, subtask2.output_dataproducts.count())
        self.assertEqual(1, subtask2.input_dataproducts.count())
        self.assertEqual(output2_dp, subtask2.output_dataproducts.first())
        self.assertEqual(input2_dp, subtask2.input_dataproducts.first())

        # now test the get_transformed_input_dataproduct/get_transformed_ouput_dataproduct methods
        self.assertEqual(input2_dp, subtask2.get_transformed_input_dataproduct(output2_dp.id))
        self.assertEqual(output2_dp, subtask2.get_transformed_output_dataproduct(input2_dp.id))

        # add some extra data
        output1_dp2:models.Dataproduct = models.Dataproduct.objects.create(**Dataproduct_test_data(producer=output1))
        input2_dp2 = output1_dp2
        input2.dataproducts.set([input2_dp, input2_dp2])
        input2.save()

        output2_dp2:models.Dataproduct = models.Dataproduct.objects.create(**Dataproduct_test_data(producer=output2))
        models.DataproductTransform.objects.create(input=input2_dp2, output=output2_dp2, identity=True)

        # make sure the subtask and input/output dp's are setup correct...
        self.assertEqual(2, subtask2.output_dataproducts.count())
        self.assertEqual(2, subtask2.input_dataproducts.count())

        # test the get_transformed_input_dataproduct/get_transformed_ouput_dataproduct methods again, wit multiple dp's
        self.assertEqual(input2_dp, subtask2.get_transformed_input_dataproduct(output2_dp.id))
        self.assertEqual(input2_dp2, subtask2.get_transformed_input_dataproduct(output2_dp2.id))
        self.assertEqual(output2_dp, subtask2.get_transformed_output_dataproduct(input2_dp.id))
        self.assertEqual(output2_dp2, subtask2.get_transformed_output_dataproduct(input2_dp2.id))


    def test_Subtask_raises_ValidationError_on_duplicate_pointing_names(self):
        # setup
        test_data = Subtask_test_data()
        test_data['specifications_doc']['stations'] = {'digital_pointings': [{'name': 'popular_name'},
                                                                          {'name': 'popular_name'},
                                                                          {'name': 'not_so_popular'}]}
        # assert
        with self.assertRaises(ValidationError) as context:
            models.Subtask = models.Subtask.objects.create(**test_data)
        self.assertIn('popular_name', str(context.exception))


class DataproductTest(unittest.TestCase):
    def test_Dataproduct_gets_created_with_correct_creation_timestamp(self):

        # setup
        before = datetime.utcnow()
        entry = models.Dataproduct.objects.create(**Dataproduct_test_data())

        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.created_at)
        self.assertGreater(after, entry.created_at)

    def test_Dataproduct_update_timestamp_gets_changed_correctly(self):

        # setup
        entry = models.Dataproduct.objects.create(**Dataproduct_test_data())
        before = datetime.utcnow()
        entry.save()
        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.updated_at)
        self.assertGreater(after, entry.updated_at)


    def test_Dataproduct_prevents_missing_specifications_template(self):

        # setup
        test_data = dict(Dataproduct_test_data())
        test_data['specifications_template'] = None

        # assert
        with self.assertRaises(IntegrityError):
            models.Dataproduct.objects.create(**test_data)


class AntennaSetTest(unittest.TestCase):
    def test_AntennaSet_gets_created_with_correct_creation_timestamp(self):

        # setup
        before = datetime.utcnow()
        entry = models.AntennaSet.objects.create(**AntennaSet_test_data())

        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.created_at)
        self.assertGreater(after, entry.created_at)

    def test_AntennaSet_update_timestamp_gets_changed_correctly(self):

        # setup
        entry = models.AntennaSet.objects.create(**AntennaSet_test_data())
        before = datetime.utcnow()
        entry.save()
        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.updated_at)
        self.assertGreater(after, entry.updated_at)


class DataproductTransformTest(unittest.TestCase):
    def test_DataproductTransform_gets_created_with_correct_creation_timestamp(self):

        # setup
        before = datetime.utcnow()
        entry = models.DataproductTransform.objects.create(**DataproductTransform_test_data())

        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.created_at)
        self.assertGreater(after, entry.created_at)

    def test_DataproductTransform_update_timestamp_gets_changed_correctly(self):

        # setup
        entry = models.DataproductTransform.objects.create(**DataproductTransform_test_data())
        before = datetime.utcnow()
        entry.save()
        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.updated_at)
        self.assertGreater(after, entry.updated_at)


class FilesystemTest(unittest.TestCase):
    def test_Filesystem_gets_created_with_correct_creation_timestamp(self):

        # setup
        before = datetime.utcnow()
        entry = models.Filesystem.objects.create(**Filesystem_test_data())

        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.created_at)
        self.assertGreater(after, entry.created_at)

    def test_Filesystem_update_timestamp_gets_changed_correctly(self):

        # setup
        entry = models.Filesystem.objects.create(**Filesystem_test_data())
        before = datetime.utcnow()
        entry.save()
        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.updated_at)
        self.assertGreater(after, entry.updated_at)

    def test_Filesystem_appends_trailing_slash_to_dirname(self):
        fs = models.Filesystem.objects.create(**Filesystem_test_data(directory="/no/trailing/slash"))
        self.assertTrue(fs.directory.endswith('/'))


class ClusterTest(unittest.TestCase):
    def test_Cluster_gets_created_with_correct_creation_timestamp(self):
        # setup
        before = datetime.utcnow()
        entry = models.Cluster.objects.create(**Cluster_test_data())

        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.created_at)
        self.assertGreater(after, entry.created_at)

    def test_Cluster_update_timestamp_gets_changed_correctly(self):

        # setup
        entry = models.Cluster.objects.create(**Cluster_test_data())
        before = datetime.utcnow()
        entry.save()
        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.updated_at)
        self.assertGreater(after, entry.updated_at)


class DataproductArchiveInfoTest(unittest.TestCase):
    def test_DataproductArchiveInfo_gets_created_with_correct_creation_timestamp(self):

        # setup
        before = datetime.utcnow()
        entry = models.DataproductArchiveInfo.objects.create(**DataproductArchiveInfo_test_data())

        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.created_at)
        self.assertGreater(after, entry.created_at)

    def test_DataproductArchiveInfo_update_timestamp_gets_changed_correctly(self):

        # setup
        entry = models.DataproductArchiveInfo.objects.create(**DataproductArchiveInfo_test_data())
        before = datetime.utcnow()
        entry.save()
        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.updated_at)
        self.assertGreater(after, entry.updated_at)


class DataproductHashTest(unittest.TestCase):
    def test_DataproductHash_gets_created_with_correct_creation_timestamp(self):

        # setup
        before = datetime.utcnow()
        entry = models.DataproductHash.objects.create(**DataproductHash_test_data())

        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.created_at)
        self.assertGreater(after, entry.created_at)

    def test_DataproductHash_update_timestamp_gets_changed_correctly(self):

        # setup
        entry = models.DataproductHash.objects.create(**DataproductHash_test_data())
        before = datetime.utcnow()
        entry.save()
        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.updated_at)
        self.assertGreater(after, entry.updated_at)


class SAPTemplateTest(unittest.TestCase):
    def test_SAPTemplate_gets_created_with_correct_creation_timestamp(self):

        # setup
        before = datetime.utcnow()
        entry = models.SAPTemplate.objects.create(**SAPTemplate_test_data())

        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.created_at)
        self.assertGreater(after, entry.created_at)

    def test_SAPTemplate_update_timestamp_gets_changed_correctly(self):

        # setup
        entry = models.SAPTemplate.objects.create(**SAPTemplate_test_data())
        before = datetime.utcnow()
        entry.save()
        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.updated_at)
        self.assertGreater(after, entry.updated_at)


class SAPTest(unittest.TestCase):
    def test_SAP_gets_created_with_correct_creation_timestamp(self):

        # setup
        before = datetime.utcnow()
        entry = models.SAP.objects.create(**SAP_test_data())

        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.created_at)
        self.assertGreater(after, entry.created_at)

    def test_SAP_update_timestamp_gets_changed_correctly(self):

        # setup
        entry = models.SAP.objects.create(**SAP_test_data())
        before = datetime.utcnow()
        entry.save()
        after = datetime.utcnow()

        # assert
        self.assertLess(before, entry.updated_at)
        self.assertGreater(after, entry.updated_at)


    def test_SAP_prevents_missing_specifications_template(self):

        # setup
        test_data = dict(SAP_test_data())
        test_data['specifications_template'] = None

        # assert
        with self.assertRaises(IntegrityError):
            models.SAP.objects.create(**test_data)


if __name__ == "__main__":
    os.environ['TZ'] = 'UTC'
    unittest.main()