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)


# 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


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

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


class SubtaskInputTest(unittest.TestCase):
    def test_SubtaskInput_gets_created_with_correct_creation_timestamp(self):

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

        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())
        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())
        test_data['subtask'] = None

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


class SubtaskTest(unittest.TestCase):
    def test_Subtask_gets_created_with_correct_creation_timestamp(self):

        # setup
        before = datetime.utcnow()
        entry = models.Subtask.objects.create(**Subtask_test_data())

        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())
        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())
        subtask2:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())

        output1 = models.SubtaskOutput.objects.create(subtask=subtask1)
        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())
        subtask3:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())
        subtask4:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())
        subtask5:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())
        subtask6:models.Subtask = models.Subtask.objects.create(**Subtask_test_data())

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

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

        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()))

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)


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)


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