Select Git revision
test_webservice.py
-
Jorrit Schaap authoredTask #8758: merge from branch RA-Task8758 to trunk. svn merge did not work, so I used svn diff on branch and svn patch to trunk. For revision history, see branch.
Jorrit Schaap authoredTask #8758: merge from branch RA-Task8758 to trunk. svn merge did not work, so I used svn diff on branch and svn patch to trunk. For revision history, see branch.
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
t_tmssapp_specification_django_API.py 36.51 KiB
#!/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
import uuid
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
from django.core.exceptions import ValidationError
from lofar.sas.tmss.tmss.exceptions import SchemaValidationException
class GeneratorTemplateTest(unittest.TestCase):
def test_GeneratorTemplate_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.GeneratorTemplate.objects.create(**GeneratorTemplate_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_GeneratorTemplate_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.GeneratorTemplate.objects.create(**GeneratorTemplate_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
class DefaultGeneratorTemplateTest(unittest.TestCase):
def test_DefaultGeneratorTemplate_prevents_same_name(self):
common_forbidden_name = "my_name"
template = models.GeneratorTemplate.objects.create(**GeneratorTemplate_test_data())
test_data_1 = DefaultGeneratorTemplate_test_data(common_forbidden_name, template)
models.DefaultGeneratorTemplate.objects.create(**test_data_1)
test_data_2 = DefaultGeneratorTemplate_test_data(common_forbidden_name, template)
with self.assertRaises(IntegrityError):
models.DefaultGeneratorTemplate.objects.create(**test_data_2)
class SchedulingUnitTemplateTest(unittest.TestCase):
def test_SchedulingUnitTemplate_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.SchedulingUnitTemplate.objects.create(**SchedulingUnitTemplate_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_SchedulingUnitTemplate_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.SchedulingUnitTemplate.objects.create(**SchedulingUnitTemplate_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
class SchedulingConstraintsTemplateTest(unittest.TestCase):
def test_SchedulingConstraintsTemplate_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.SchedulingConstraintsTemplate.objects.create(**SchedulingConstraintsTemplate_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_SchedulingConstraintsTemplate_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.SchedulingConstraintsTemplate.objects.create(**SchedulingConstraintsTemplate_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
class TaskTemplateTest(unittest.TestCase):
def test_TaskTemplate_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.TaskTemplate.objects.create(**TaskTemplate_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_TaskTemplate_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.TaskTemplate.objects.create(**TaskTemplate_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_TaskTemplate_incorrect_schema_raises(self):
with self.assertRaises(SchemaValidationException):
models.TaskTemplate.objects.create(**TaskTemplate_test_data(schema=""))
with self.assertRaises(SchemaValidationException) as context:
models.TaskTemplate.objects.create(**TaskTemplate_test_data(schema={}))
self.assertTrue(True)
with self.assertRaises(SchemaValidationException) as context:
schema = minimal_json_schema()
del schema['$schema']
models.TaskTemplate.objects.create(**TaskTemplate_test_data(schema=schema))
self.assertTrue("Missing required properties" in str(context.exception))
with self.assertRaises(SchemaValidationException) as context:
models.TaskTemplate.objects.create(**TaskTemplate_test_data(schema= minimal_json_schema(id="my id with no url")))
self.assertTrue("should contain a valid URL" in str(context.exception))
def test_TaskTemplate_annotated_schema(self):
schema = minimal_json_schema()
data = TaskTemplate_test_data(schema=schema, name="foo", description="bar")
template = models.TaskTemplate.objects.create(**data)
self.assertEqual("foo", template.name)
self.assertEqual("foo", template.schema['title'])
self.assertEqual("bar", template.description)
self.assertEqual("bar", template.schema['description'])
def test_TaskTemplate_name_version_unique(self):
name = str(uuid.uuid4())
self.assertEqual(0, models.TaskTemplate.objects.filter(name=name).count())
test_data = TaskTemplate_test_data(name=name)
# save data twice
entry1 = models.TaskTemplate.objects.create(**test_data)
entry2 = models.TaskTemplate.objects.create(**test_data)
self.assertEqual(2, models.TaskTemplate.objects.filter(name=name).count())
self.assertEqual(1, entry1.version)
self.assertEqual(2, entry2.version) #version is autoincremented
# try to modify version... should be allowed, cause the template is not used, but should raise IntegrityError (unique constraint)
self.assertFalse(entry2.is_used)
with self.assertRaises(IntegrityError):
entry2.version = 1
entry2.save()
entry2.refresh_from_db()
# versions still the same?
self.assertEqual(1, entry1.version)
self.assertEqual(2, entry2.version)
# let's use the template in a task
models.TaskDraft.objects.create(**TaskDraft_test_data(specifications_template=entry2))
self.assertTrue(entry2.is_used)
# there should still be only 2 templates with this name
self.assertEqual(2, models.TaskTemplate.objects.filter(name=name).count())
# now (try to) modify the template
org_pk = entry2.pk
org_schema = dict(entry2.schema)
new_schema = minimal_json_schema(properties={"new_prop":{"type":"string"}})
entry2.schema = new_schema
entry2.save()
#this should now be a NEW instance
self.assertNotEqual(org_pk, entry2.pk)
self.assertEqual(3, models.TaskTemplate.objects.filter(name=name).count())
# lets request the "old" entry2 via name and version, so we can check if it is unchanged
entry2 = models.TaskTemplate.objects.get(name=name, version=2)
self.assertEqual(org_schema, entry2.schema)
# instead there should be a new version of the template with the new schema
entry3 = models.TaskTemplate.objects.get(name=name, version=3)
self.assertEqual(3, entry3.version)
self.assertEqual(new_schema, entry3.schema)
class TaskRelationSelectionTemplateTest(unittest.TestCase):
def test_TaskRelationSelectionTemplate_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.TaskRelationSelectionTemplate.objects.create(**TaskRelationSelectionTemplate_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_TaskRelationSelectionTemplate_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.TaskRelationSelectionTemplate.objects.create(**TaskRelationSelectionTemplate_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
class TaskConnectorTest(unittest.TestCase):
def test_POST_TaskConnector_prevents_missing_input_of(self):
# setup
test_data_1 = dict(TaskConnectorType_test_data())
test_data_1['input_of'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskConnectorType.objects.create(**test_data_1)
def test_POST_TaskConnector_prevents_missing_output_of(self):
# setup
test_data_1 = dict(TaskConnectorType_test_data())
test_data_1['output_of'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskConnectorType.objects.create(**test_data_1)
class CycleTest(unittest.TestCase):
def test_Cycle_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.Cycle.objects.create(**Cycle_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_Cycle_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.Cycle.objects.create(**Cycle_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
class ProjectTest(unittest.TestCase):
def test_Project_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.Project.objects.create(**Project_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_Project_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.Project.objects.create(**Project_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_Project_raises_ValueError_on_invalid_archive_subdirectory_name(self):
# setup
test_data_1 = Project_test_data(archive_subdirectory="no/trailing/slash")
test_data_2 = Project_test_data(archive_subdirectory="/with/leading/slash/")
# assert
with self.assertRaises(ValueError):
entry = models.Project.objects.create(**test_data_1)
with self.assertRaises(ValueError):
entry = models.Project.objects.create(**test_data_2)
class ProjectQuotaTest(unittest.TestCase):
def test_ProjectQuota_prevents_missing_project(self):
# setup
test_data = dict(ProjectQuota_test_data())
test_data['project'] = None
# assert
with self.assertRaises(IntegrityError):
models.ProjectQuota.objects.create(**test_data)
class SchedulingSetTest(unittest.TestCase):
def test_SchedulingSet_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.SchedulingSet.objects.create(**SchedulingSet_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_SchedulingSet_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.SchedulingSet.objects.create(**SchedulingSet_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_SchedulingSet_prevents_missing_project(self):
# setup
test_data = dict(SchedulingSet_test_data())
test_data['project'] = None
# assert
with self.assertRaises(IntegrityError):
models.SchedulingSet.objects.create(**test_data)
class SchedulingUnitDraftTest(unittest.TestCase):
def test_SchedulingUnitDraft_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_SchedulingUnitDraft_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_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_SchedulingUnitDraft_prevents_missing_template(self):
# setup
test_data = dict(SchedulingUnitDraft_test_data())
test_data['requirements_template'] = None
# assert
with self.assertRaises(IntegrityError):
models.SchedulingUnitDraft.objects.create(**test_data)
def test_SchedulingUnitDraft_prevents_missing_scheduling_set(self):
# setup
test_data = dict(SchedulingUnitDraft_test_data())
test_data['scheduling_set'] = None
# assert
with self.assertRaises(IntegrityError):
models.SchedulingUnitDraft.objects.create(**test_data)
def test_SchedulingUnitDraft_gets_created_with_correct_default_ingest_permission_required(self):
# setup
entry = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data())
#check the auto_ingest on project
self.assertEqual(False, entry.scheduling_set.project.auto_ingest)
#When auto_ingest=False (in project), the scheduling units should be created with ingest_permission_required = True
self.assertEqual(True, entry.ingest_permission_required)
class TaskDraftTest(unittest.TestCase):
def test_TaskDraft_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.TaskDraft.objects.create(**TaskDraft_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_TaskDraft_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.TaskDraft.objects.create(**TaskDraft_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_TaskDraft_gets_created_with_correct_output_pinned_flag(self):
# setup
project_1 = models.Project.objects.create(**Project_test_data(auto_pin=False))
scheduling_set_1 = models.SchedulingSet.objects.create(**SchedulingSet_test_data(project=project_1))
scheduling_unit_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(scheduling_set=scheduling_set_1))
task_draft_1 = models.TaskDraft.objects.create(**TaskDraft_test_data(scheduling_unit_draft=scheduling_unit_1))
project_2 = models.Project.objects.create(**Project_test_data(auto_pin=True))
scheduling_set_2 = models.SchedulingSet.objects.create(**SchedulingSet_test_data(project=project_2))
scheduling_unit_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(scheduling_set=scheduling_set_2))
task_draft_2 = models.TaskDraft.objects.create(**TaskDraft_test_data(scheduling_unit_draft=scheduling_unit_2))
task_draft_1.refresh_from_db()
task_draft_2.refresh_from_db()
# assert
self.assertFalse(task_draft_1.output_pinned)
self.assertTrue(task_draft_2.output_pinned)
def test_TaskDraft_prevents_missing_template(self):
# setup
test_data = dict(TaskDraft_test_data())
test_data['specifications_template'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskDraft.objects.create(**test_data)
def test_TaskDraft_prevents_missing_scheduling_unit_draft(self):
# setup
test_data = dict(TaskDraft_test_data())
test_data['scheduling_unit_draft'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskDraft.objects.create(**test_data)
def test_TaskDraft_predecessors_and_successors_none(self):
task_draft_1:models.TaskDraft = models.TaskDraft.objects.create(**TaskDraft_test_data())
task_draft_2:models.TaskDraft = models.TaskDraft.objects.create(**TaskDraft_test_data())
self.assertEqual(set(), set(task_draft_1.predecessors.all()))
self.assertEqual(set(), set(task_draft_2.predecessors.all()))
self.assertEqual(set(), set(task_draft_1.successors.all()))
self.assertEqual(set(), set(task_draft_2.successors.all()))
def test_TaskDraft_predecessors_and_successors_simple(self):
task_draft_1:models.TaskDraft = models.TaskDraft.objects.create(**TaskDraft_test_data())
task_draft_2:models.TaskDraft = models.TaskDraft.objects.create(**TaskDraft_test_data())
models.TaskRelationDraft.objects.create(**TaskRelationDraft_test_data(producer=task_draft_1,
consumer=task_draft_2))
self.assertEqual(task_draft_1, task_draft_2.predecessors.all()[0])
self.assertEqual(task_draft_2, task_draft_1.successors.all()[0])
def test_TaskDraft_predecessors_and_successors_complex(self):
task_draft_1:models.TaskDraft = models.TaskDraft.objects.create(**TaskDraft_test_data())
task_draft_2:models.TaskDraft = models.TaskDraft.objects.create(**TaskDraft_test_data())
task_draft_3:models.TaskDraft = models.TaskDraft.objects.create(**TaskDraft_test_data())
task_draft_4:models.TaskDraft = models.TaskDraft.objects.create(**TaskDraft_test_data())
task_draft_5:models.TaskDraft = models.TaskDraft.objects.create(**TaskDraft_test_data())
task_draft_6:models.TaskDraft = models.TaskDraft.objects.create(**TaskDraft_test_data())
# ST1 ---> ST3 ---> ST4
# | |
# ST2 - -> ST5 ---> ST6
models.TaskRelationDraft.objects.create(**TaskRelationDraft_test_data(producer=task_draft_1,
consumer=task_draft_3))
models.TaskRelationDraft.objects.create(**TaskRelationDraft_test_data(producer=task_draft_2,
consumer=task_draft_3))
models.TaskRelationDraft.objects.create(**TaskRelationDraft_test_data(producer=task_draft_3,
consumer=task_draft_4))
models.TaskRelationDraft.objects.create(**TaskRelationDraft_test_data(producer=task_draft_3,
consumer=task_draft_5))
models.TaskRelationDraft.objects.create(**TaskRelationDraft_test_data(producer=task_draft_5,
consumer=task_draft_6))
self.assertEqual(set((task_draft_1, task_draft_2)), set(task_draft_3.predecessors.all()))
self.assertEqual(set((task_draft_4, task_draft_5)), set(task_draft_3.successors.all()))
self.assertEqual(set((task_draft_3,)), set(task_draft_4.predecessors.all()))
self.assertEqual(set((task_draft_3,)), set(task_draft_5.predecessors.all()))
self.assertEqual(set((task_draft_3,)), set(task_draft_1.successors.all()))
self.assertEqual(set((task_draft_3,)), set(task_draft_2.successors.all()))
self.assertEqual(set(), set(task_draft_1.predecessors.all()))
self.assertEqual(set(), set(task_draft_2.predecessors.all()))
self.assertEqual(set(), set(task_draft_4.successors.all()))
self.assertEqual(set((task_draft_6,)), set(task_draft_5.successors.all()))
class TaskRelationDraftTest(unittest.TestCase):
def test_TaskRelationDraft_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.TaskRelationDraft.objects.create(**TaskRelationDraft_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_TaskRelationDraft_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.TaskRelationDraft.objects.create(**TaskRelationDraft_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_TaskRelationDraft_prevents_missing_template(self):
# setup
test_data = dict(TaskRelationDraft_test_data())
test_data['selection_template'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskRelationDraft.objects.create(**test_data)
def test_TaskRelationDraft_prevents_missing_consumer(self):
# setup
test_data = dict(TaskRelationDraft_test_data())
test_data['consumer'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskRelationDraft.objects.create(**test_data)
def test_TaskRelationDraft_prevents_missing_producer(self):
# setup
test_data = dict(TaskRelationDraft_test_data())
test_data['producer'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskRelationDraft.objects.create(**test_data)
class SchedulingUnitBlueprintTest(unittest.TestCase):
def test_SchedulingUnitBlueprint_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_SchedulingUnitBlueprint_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_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_SchedulingUnitBlueprint_prevents_missing_template(self):
# setup
test_data = dict(SchedulingUnitBlueprint_test_data())
test_data['requirements_template'] = None
# assert
with self.assertRaises(IntegrityError):
models.SchedulingUnitBlueprint.objects.create(**test_data)
def test_SchedulingUnitBlueprint_prevents_missing_draft(self):
# setup
test_data = dict(SchedulingUnitBlueprint_test_data())
test_data['draft'] = None
# assert
with self.assertRaises(IntegrityError):
models.SchedulingUnitBlueprint.objects.create(**test_data)
def test_SchedulingUnitBlueprint_gets_created_with_correct_default_ingest_permission_required(self):
# setup
entry = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data())
#check the auto_ingest on project
self.assertEqual(False, entry.draft.scheduling_set.project.auto_ingest)
#When auto_ingest=False (in project), the scheduling units should be created with ingest_permission_required = True
self.assertEqual(True, entry.ingest_permission_required)
class TaskBlueprintTest(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.task_draft = models.TaskDraft.objects.create(**TaskDraft_test_data())
cls.scheduling_unit_blueprint = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data())
def test_TaskBlueprint_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(task_draft=self.task_draft, scheduling_unit_blueprint=self.scheduling_unit_blueprint))
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_TaskBlueprint_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(task_draft=self.task_draft, scheduling_unit_blueprint=self.scheduling_unit_blueprint))
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
def test_TaskBlueprint_prevents_missing_template(self):
# setup
test_data = dict(TaskBlueprint_test_data(task_draft=self.task_draft, scheduling_unit_blueprint=self.scheduling_unit_blueprint))
test_data['specifications_template'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskBlueprint.objects.create(**test_data)
def test_TaskBlueprint_prevents_missing_draft(self):
# setup
test_data = dict(TaskBlueprint_test_data(task_draft=self.task_draft, scheduling_unit_blueprint=self.scheduling_unit_blueprint))
test_data['draft'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskBlueprint.objects.create(**test_data)
def test_TaskBlueprint_prevents_missing_scheduling_unit_blueprint(self):
# setup
test_data = dict(TaskBlueprint_test_data(task_draft=self.task_draft, scheduling_unit_blueprint=self.scheduling_unit_blueprint))
test_data['scheduling_unit_blueprint'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskBlueprint.objects.create(**test_data)
def test_TaskBlueprint_predecessors_and_successors_none(self):
task_blueprint_1: models.TaskBlueprint = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(task_draft=self.task_draft, scheduling_unit_blueprint=self.scheduling_unit_blueprint))
task_blueprint_2: models.TaskBlueprint = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(task_draft=self.task_draft, scheduling_unit_blueprint=self.scheduling_unit_blueprint))
self.assertEqual(set(), set(task_blueprint_1.predecessors.all()))
self.assertEqual(set(), set(task_blueprint_2.predecessors.all()))
self.assertEqual(set(), set(task_blueprint_1.successors.all()))
self.assertEqual(set(), set(task_blueprint_2.successors.all()))
def test_TaskBlueprint_predecessors_and_successors_simple(self):
task_blueprint_1: models.TaskBlueprint = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(task_draft=self.task_draft, scheduling_unit_blueprint=self.scheduling_unit_blueprint))
task_blueprint_2: models.TaskBlueprint = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(task_draft=self.task_draft, scheduling_unit_blueprint=self.scheduling_unit_blueprint))
models.TaskRelationBlueprint.objects.create(**TaskRelationBlueprint_test_data(producer=task_blueprint_1,
consumer=task_blueprint_2))
self.assertEqual(task_blueprint_1, task_blueprint_2.predecessors.all()[0])
self.assertEqual(task_blueprint_2, task_blueprint_1.successors.all()[0])
def test_TaskBlueprint_predecessors_and_successors_complex(self):
task_blueprint_1: models.TaskBlueprint = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data())
task_blueprint_2: models.TaskBlueprint = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(task_draft=task_blueprint_1.draft, scheduling_unit_blueprint=task_blueprint_1.scheduling_unit_blueprint))
task_blueprint_3: models.TaskBlueprint = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(task_draft=task_blueprint_1.draft, scheduling_unit_blueprint=task_blueprint_1.scheduling_unit_blueprint))
task_blueprint_4: models.TaskBlueprint = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(task_draft=task_blueprint_1.draft, scheduling_unit_blueprint=task_blueprint_1.scheduling_unit_blueprint))
task_blueprint_5: models.TaskBlueprint = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(task_draft=task_blueprint_1.draft, scheduling_unit_blueprint=task_blueprint_1.scheduling_unit_blueprint))
task_blueprint_6: models.TaskBlueprint = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(task_draft=task_blueprint_1.draft, scheduling_unit_blueprint=task_blueprint_1.scheduling_unit_blueprint))
# ST1 ---> ST3 ---> ST4
# | |
# ST2 - -> ST5 ---> ST6
models.TaskRelationBlueprint.objects.create(**TaskRelationBlueprint_test_data(producer=task_blueprint_1,
consumer=task_blueprint_3))
trb1 = models.TaskRelationBlueprint.objects.create(**TaskRelationBlueprint_test_data(producer=task_blueprint_2,
consumer=task_blueprint_3))
models.TaskRelationBlueprint.objects.create(**TaskRelationBlueprint_test_data(producer=task_blueprint_3,
consumer=task_blueprint_4))
models.TaskRelationBlueprint.objects.create(**TaskRelationBlueprint_test_data(producer=task_blueprint_3,
consumer=task_blueprint_5))
models.TaskRelationBlueprint.objects.create(**TaskRelationBlueprint_test_data(producer=task_blueprint_5,
consumer=task_blueprint_6))
self.assertEqual(set((task_blueprint_1, task_blueprint_2)), set(task_blueprint_3.predecessors.all()))
self.assertEqual(set((task_blueprint_4, task_blueprint_5)), set(task_blueprint_3.successors.all()))
self.assertEqual(set((task_blueprint_3,)), set(task_blueprint_4.predecessors.all()))
self.assertEqual(set((task_blueprint_3,)), set(task_blueprint_5.predecessors.all()))
self.assertEqual(set((task_blueprint_3,)), set(task_blueprint_1.successors.all()))
self.assertEqual(set((task_blueprint_3,)), set(task_blueprint_2.successors.all()))
self.assertEqual(set(), set(task_blueprint_1.predecessors.all()))
self.assertEqual(set(), set(task_blueprint_2.predecessors.all()))
self.assertEqual(set(), set(task_blueprint_4.successors.all()))
self.assertEqual(set((task_blueprint_6,)), set(task_blueprint_5.successors.all()))
class TaskRelationBlueprintTest(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.producer = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data())
cls.consumer = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data())
def test_TaskRelationBlueprint_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.TaskRelationBlueprint.objects.create(**TaskRelationBlueprint_test_data(producer=self.producer, consumer=self.consumer))
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_TaskRelationBlueprint_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.TaskRelationBlueprint.objects.create(**TaskRelationBlueprint_test_data(producer=self.producer, consumer=self.consumer))
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
def test_TaskRelationBlueprint_prevents_missing_selection_template(self):
# setup
test_data = dict(TaskRelationBlueprint_test_data(producer=self.producer, consumer=self.consumer))
test_data['selection_template'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskRelationBlueprint.objects.create(**test_data)
def test_TaskRelationBlueprint_prevents_missing_draft(self):
# setup
test_data = dict(TaskRelationBlueprint_test_data(producer=self.producer, consumer=self.consumer))
test_data['draft'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskRelationBlueprint.objects.create(**test_data)
def test_TaskRelationBlueprint_prevents_missing_producer(self):
# setup
test_data = dict(TaskRelationBlueprint_test_data(producer=self.producer, consumer=self.consumer))
test_data['producer'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskRelationBlueprint.objects.create(**test_data)
def test_TaskRelationBlueprint_prevents_missing_consumer(self):
# setup
test_data = dict(TaskRelationBlueprint_test_data(producer=self.producer, consumer=self.consumer))
test_data['consumer'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskRelationBlueprint.objects.create(**test_data)
def test_TaskRelationBlueprint_prevents_missing_input(self):
# setup
test_data = dict(TaskRelationBlueprint_test_data(producer=self.producer, consumer=self.consumer))
test_data['input_role'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskRelationBlueprint.objects.create(**test_data)
def test_TaskRelationBlueprint_prevents_missing_output(self):
# setup
test_data = dict(TaskRelationBlueprint_test_data(producer=self.producer, consumer=self.consumer))
test_data['output_role'] = None
# assert
with self.assertRaises(IntegrityError):
models.TaskRelationBlueprint.objects.create(**test_data)
if __name__ == "__main__":
os.environ['TZ'] = 'UTC'
unittest.main()