Select Git revision
queuetypes.py
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
t_tmssapp_specification_REST_API.py 223.87 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: $
# This functional test talks to the API like a regular user would.
# It is supposed to cover all REST http methods for all ViewSets.
# I am still a bit under the impression that we re-test Django functionality that we can expect to just work
# with some of these tests. On the other hand a lot of these provide us a nice basis for differentiating out
# behavior in a controlled way.
# We should probably also fully test behavior wrt mandatory and nullable fields.
from dateutil import parser
from datetime import datetime, timedelta
import unittest
import logging
logger = logging.getLogger('lofar.'+__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()
# Do Mandatory setup step:
# use setup/teardown magic for tmss test database, ldap server and django server
# (ignore pycharm unused import statement, python unittests does use at RunTime the tmss_test_environment_unittest_setup module)
from lofar.sas.tmss.test.tmss_test_environment_unittest_setup import *
from lofar.sas.tmss.test.tmss_test_data_django_models import *
from lofar.sas.tmss.tmss.tmssapp import models
from lofar.sas.tmss.test.test_utils import assertUrlList
from django.contrib.auth.models import Group, Permission
from django.contrib.auth import get_user_model
User = get_user_model()
# import and setup test data creator
from lofar.sas.tmss.test.tmss_test_data_rest import TMSSRESTTestDataCreator
test_data_creator = TMSSRESTTestDataCreator(BASE_URL, AUTH)
tmss_test_env.populate_schemas()
from lofar.common.json_utils import add_defaults_to_json_object_for_schema
# todo: for overall speed improvements, but also for clarity, it would be nice to switch to django.test.TestCase
# in order to separate the db content between them. Investigated why that currently yields a ton of 404 errors.
# Note that mixing unittest.TestCase and django.test.TestCase does not seem to isolate tests properly.
class BasicFunctionTestCase(unittest.TestCase):
# todo: test_welcome_page (once we have one :))
pass
class GeneratorTemplateTestCase(unittest.TestCase):
def test_generator_template_list_apiformat(self):
r = requests.get(BASE_URL + '/generator_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Generator Template List" in r.content.decode('utf8'))
def test_generator_template_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/generator_template/1234321/', 404)
def test_generator_template_POST_and_GET(self):
# POST and GET a new item and assert correctness
test_data = test_data_creator.GeneratorTemplate()
expected_data = test_data_creator.update_schema_from_template("generatortemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_generator_template_PUT_invalid_raises_error(self):
test_data = test_data_creator.GeneratorTemplate()
PUT_and_assert_expected_response(self, BASE_URL + '/generator_template/9876789876/', test_data, 404, {})
def test_generator_template_PUT(self):
# POST new item, verify
test_data = test_data_creator.GeneratorTemplate()
expected_data = test_data_creator.update_schema_from_template("generatortemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# PUT new values, verify
test_data2 = test_data_creator.GeneratorTemplate("generatortemplate2")
expected_data2 = test_data_creator.update_schema_from_template("generatortemplate", test_data2)
PUT_and_assert_expected_response(self, url, test_data2, 200, expected_data2)
GET_OK_and_assert_equal_expected_response(self, url, expected_data2)
def test_generator_template_PATCH(self):
# POST new item, verify
test_data = test_data_creator.GeneratorTemplate()
expected_data = test_data_creator.update_schema_from_template("generatortemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
test_patch = {"name": "new_name",
"description": "better description",
"schema": minimal_json_schema(properties={"mykey": {"type":"string", "default":"my better value"}})}
# PATCH item and verify
expected_patch_data = test_data_creator.update_schema_from_template("generatortemplate", test_patch)
PATCH_and_assert_expected_response(self, url, test_patch, 200, expected_patch_data)
expected_data = dict(test_data)
expected_data.update(expected_patch_data)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_generator_template_DELETE(self):
# POST new item, verify
test_data = test_data_creator.GeneratorTemplate()
expected_data = test_data_creator.update_schema_from_template("generatortemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
class SchedulingUnitTemplateTestCase(unittest.TestCase):
def test_scheduling_unit_template_list_apiformat(self):
r = requests.get(BASE_URL + '/scheduling_unit_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Scheduling Unit Template List" in r.content.decode('utf8'))
def test_scheduling_unit_template_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_template/1234321/', 404)
def test_scheduling_unit_template_POST_and_GET(self):
# POST and GET a new item and assert correctness
test_data = test_data_creator.SchedulingUnitTemplate()
expected_data = test_data_creator.update_schema_from_template("schedulingunittemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url+'?format=json', expected_data)
def test_scheduling_unit_template_PUT_invalid_raises_error(self):
test_data = test_data_creator.SchedulingUnitTemplate()
PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/9876789876/', test_data, 404, {})
def test_scheduling_unit_template_PUT(self):
# POST new item, verify
test_data = test_data_creator.SchedulingUnitTemplate()
expected_data = test_data_creator.update_schema_from_template("schedulingunittemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# PUT new values, verify
test_data2 = test_data_creator.SchedulingUnitTemplate("schedulingunittemplate2")
expected_data2 = test_data_creator.update_schema_from_template("schedulingunittemplate", test_data2)
PUT_and_assert_expected_response(self, url, test_data2, 200, expected_data2)
GET_OK_and_assert_equal_expected_response(self, url, expected_data2)
def test_scheduling_unit_template_PATCH(self):
# POST new item, verify
test_data = test_data_creator.SchedulingUnitTemplate()
expected_data = test_data_creator.update_schema_from_template("schedulingunittemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
test_patch = {"name": "new_name",
"description": "better description",
"schema": minimal_json_schema(properties={"mykey": {"type":"string", "default":"my better value"}})}
# PATCH item and verify
expected_patch_data = test_data_creator.update_schema_from_template("schedulingunittemplate", test_patch)
PATCH_and_assert_expected_response(self, url, test_patch, 200, expected_patch_data)
expected_data = dict(test_data)
expected_data.update(expected_patch_data)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_scheduling_unit_template_DELETE(self):
# POST new item, verify
test_data = test_data_creator.SchedulingUnitTemplate()
expected_data = test_data_creator.update_schema_from_template("schedulingunittemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
class SchedulingConstraintsTemplateTestCase(unittest.TestCase):
def test_scheduling_constraints_template_list_apiformat(self):
r = requests.get(BASE_URL + '/scheduling_constraints_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Scheduling Constraints Template List" in r.content.decode('utf8'))
def test_scheduling_constraints_template_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_constraints_template/1234321/', 404)
def test_scheduling_constraints_template_POST_and_GET(self):
# POST and GET a new item and assert correctness
test_data = test_data_creator.SchedulingConstraintsTemplate()
expected_data = test_data_creator.update_schema_from_template("schedulingconstraintstemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_constraints_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url+'?format=json', expected_data)
def test_scheduling_constraints_template_PUT_invalid_raises_error(self):
test_data = test_data_creator.SchedulingConstraintsTemplate()
PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_constraints_template/9876789876/', test_data, 404, {})
def test_scheduling_constraints_template_PUT(self):
# POST new item, verify
test_data = test_data_creator.SchedulingConstraintsTemplate()
expected_data = test_data_creator.update_schema_from_template("schedulingconstraintstemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_constraints_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# PUT new values, verify
test_data2 = test_data_creator.SchedulingConstraintsTemplate("schedulingconstraintstemplate2")
expected_data2 = test_data_creator.update_schema_from_template("schedulingconstraintstemplate", test_data2)
PUT_and_assert_expected_response(self, url, test_data2, 200, expected_data2)
GET_OK_and_assert_equal_expected_response(self, url, expected_data2)
def test_scheduling_constraints_template_PATCH(self):
# POST new item, verify
test_data = test_data_creator.SchedulingConstraintsTemplate()
expected_data = test_data_creator.update_schema_from_template("schedulingconstraintstemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_constraints_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
test_patch = {"name": "new_name",
"description": "better description",
"schema": minimal_json_schema(properties={"mykey": {"type":"string", "default":"my better value"}})}
# PATCH item and verify
expected_patch_data = test_data_creator.update_schema_from_template("schedulingconstraintstemplate", test_patch)
PATCH_and_assert_expected_response(self, url, test_patch, 200, expected_patch_data)
expected_data = dict(test_data)
expected_data.update(expected_patch_data)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_scheduling_constraints_template_DELETE(self):
# POST new item, verify
test_data = test_data_creator.SchedulingConstraintsTemplate()
expected_data = test_data_creator.update_schema_from_template("schedulingconstraintstemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_constraints_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
class ReservationTemplateTestCase(unittest.TestCase):
def test_reservation_template_list_apiformat(self):
r = requests.get(BASE_URL + '/reservation_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Reservation Template List" in r.content.decode('utf8'))
def test_reservation_template_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/reservation_template/1234321/', 404)
def test_reservation_template_POST_and_GET(self):
# POST and GET a new item and assert correctness
test_data = test_data_creator.ReservationTemplate()
expected_data = test_data_creator.update_schema_from_template("reservationtemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/reservation_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url+'?format=json', expected_data)
def test_reservation_template_PUT_invalid_raises_error(self):
test_data = test_data_creator.ReservationTemplate()
PUT_and_assert_expected_response(self, BASE_URL + '/reservation_template/9876789876/', test_data, 404, {})
def test_reservation_template_PUT(self):
# POST new item, verify
test_data = test_data_creator.ReservationTemplate()
expected_data = test_data_creator.update_schema_from_template("reservationtemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/reservation_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# PUT new values, verify
test_data2 = test_data_creator.ReservationTemplate("reservationtemplate2")
expected_data2 = test_data_creator.update_schema_from_template("reservationtemplate", test_data2)
PUT_and_assert_expected_response(self, url, test_data2, 200, expected_data2)
GET_OK_and_assert_equal_expected_response(self, url, expected_data2)
def test_reservation_template_PATCH(self):
# POST new item, verify
test_data = test_data_creator.ReservationTemplate()
expected_data = test_data_creator.update_schema_from_template("reservationtemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/reservation_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
test_patch = {"name": "new_name",
"description": "better description",
"schema": minimal_json_schema(properties={"mykey": {"type":"string", "default":"my better value"}})}
# PATCH item and verify
expected_patch_data = test_data_creator.update_schema_from_template("reservationtemplate", test_patch)
PATCH_and_assert_expected_response(self, url, test_patch, 200, expected_patch_data)
expected_data = dict(test_data)
expected_data.update(expected_patch_data)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_reservation_template_DELETE(self):
# POST new item, verify
test_data = test_data_creator.ReservationTemplate()
expected_data = test_data_creator.update_schema_from_template("reservationtemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/reservation_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
class ReservationStrategyTemplateTestCase(unittest.TestCase):
def test_reservation_strategy_template_list_apiformat(self):
r = requests.get(BASE_URL + '/reservation_strategy_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Reservation Strategy Template List" in r.content.decode('utf8'))
def test_reservation_strategy_template_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/reservation_strategy_template/1234321/', 404)
def test_reservation_strategy_template_POST_and_GET(self):
# POST and GET a new item and assert correctness
test_data = test_data_creator.ReservationStrategyTemplate()
expected_data = test_data_creator.update_schema_from_template("reservationtemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/reservation_strategy_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_reservation_strategy_template_PUT_invalid_raises_error(self):
test_data = test_data_creator.ReservationStrategyTemplate()
PUT_and_assert_expected_response(self, BASE_URL + '/reservation_strategy_template/9876789876/', test_data, 404, {})
def test_reservation_strategy_template_PUT(self):
# POST new item, verify
test_data = test_data_creator.ReservationStrategyTemplate()
expected_data = test_data_creator.update_schema_from_template("reservationtemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/reservation_strategy_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# PUT new values, verify
test_data2 = test_data_creator.ReservationStrategyTemplate("reservationtemplate2")
expected_data2 = test_data_creator.update_schema_from_template("reservationtemplate", test_data2)
PUT_and_assert_expected_response(self, url, test_data2, 200, expected_data2)
GET_OK_and_assert_equal_expected_response(self, url, expected_data2)
def test_reservation_strategy_template_PATCH(self):
# POST new item, verify
test_data = test_data_creator.ReservationStrategyTemplate()
expected_data = test_data_creator.update_schema_from_template("reservationtemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/reservation_strategy_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
test_patch = {"name": "new_name",
"description": "better description"}
# PATCH item and verify
expected_patch_data = test_data_creator.update_schema_from_template("reservationtemplate", test_patch)
PATCH_and_assert_expected_response(self, url, test_patch, 200, expected_patch_data)
expected_data = dict(test_data)
expected_data.update(expected_patch_data)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_reservation_strategy_template_DELETE(self):
# POST new item, verify
test_data = test_data_creator.ReservationStrategyTemplate()
expected_data = test_data_creator.update_schema_from_template("reservationtemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/reservation_strategy_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
class TaskTemplateTestCase(unittest.TestCase):
def test_task_template_list_apiformat(self):
r = requests.get(BASE_URL + '/task_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Template List" in r.content.decode('utf8'))
def test_task_template_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_template/1234321/', 404)
def test_task_template_POST_and_GET(self):
# POST and GET a new item and assert correctness
test_data = test_data_creator.TaskTemplate()
expected_data = test_data_creator.update_schema_from_template("tasktemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_task_template_PUT_invalid_raises_error(self):
test_data = test_data_creator.TaskTemplate()
PUT_and_assert_expected_response(self, BASE_URL + '/task_template/9876789876/', test_data, 404, {})
def test_task_template_PUT(self):
# POST new item, verify
test_data = test_data_creator.TaskTemplate()
expected_data = test_data_creator.update_schema_from_template("tasktemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# PUT new values, verify
test_data2 = test_data_creator.TaskTemplate("tasktemplate2")
expected_data2 = test_data_creator.update_schema_from_template("tasktemplate", test_data2)
PUT_and_assert_expected_response(self, url, test_data2, 200, expected_data2)
GET_OK_and_assert_equal_expected_response(self, url, expected_data2)
def test_task_template_PATCH(self):
# POST new item, verify
test_data = test_data_creator.TaskTemplate()
expected_data = test_data_creator.update_schema_from_template("tasktemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
test_patch = {"name": "new_name",
"description": "better description",
"schema": minimal_json_schema(properties={"mykey": {"type":"string", "default":"my better value"}})}
# PATCH item and verify
expected_patch_data = test_data_creator.update_schema_from_template("tasktemplate", test_patch)
PATCH_and_assert_expected_response(self, url, test_patch, 200, expected_patch_data)
expected_data = dict(test_data)
expected_data.update(expected_patch_data)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_task_template_DELETE(self):
# POST new item, verify
test_data = test_data_creator.TaskTemplate()
expected_data = test_data_creator.update_schema_from_template("tasktemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_GET_task_template_view_returns_correct_entry(self):
test_data_1 = TaskTemplate_test_data("task_template_1")
test_data_2 = TaskTemplate_test_data("task_template_2")
id1 = models.TaskTemplate.objects.create(**test_data_1).id
id2 = models.TaskTemplate.objects.create(**test_data_2).id
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_template/' + str(id1) + '/', test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_template/' + str(id2) + '/', test_data_2)
def test_task_template_PROTECT_behavior_on_type_choice_deleted(self):
# create dependency that is safe to delete (enums are not populated / re-established between tests)
type_data = {'value': 'kickme'}
POST_and_assert_expected_response(self, BASE_URL + '/task_type/', type_data, 201, type_data)
type_url = BASE_URL + '/task_type/kickme'
# POST new item and verify
test_data = test_data_creator.TaskTemplate()
test_data['type'] = type_url
expected_data = test_data_creator.update_schema_from_template("tasktemplate", test_data)
url = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', test_data, 201, expected_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(type_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, type_url, type_data)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
class TaskRelationSelectionTemplateTestCase(unittest.TestCase):
def test_task_relation_selection_template_list_apiformat(self):
r = requests.get(BASE_URL + '/task_relation_selection_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Relation Selection Template List" in r.content.decode('utf8'))
def test_task_relation_selection_template_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_relation_selection_template/1234321/', 404)
def test_task_relation_selection_template_POST_and_GET(self):
# POST and GET a new item and assert correctness
test_data = test_data_creator.TaskRelationSelectionTemplate()
expected_data = test_data_creator.update_schema_from_template("taskrelationselectiontemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_selection_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url+'?format=json', expected_data)
def test_task_relation_selection_template_PUT_invalid_raises_error(self):
test_data = test_data_creator.TaskRelationSelectionTemplate()
PUT_and_assert_expected_response(self, BASE_URL + '/task_relation_selection_template/9876789876/', test_data, 404, {})
def test_task_relation_selection_template_PUT(self):
# POST new item, verify
test_data = test_data_creator.TaskRelationSelectionTemplate()
expected_data = test_data_creator.update_schema_from_template("taskrelationselectiontemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_selection_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# PUT new values, verify
test_data2 = test_data_creator.TaskRelationSelectionTemplate("taskrelationselectiontemplate2")
expected_data2 = test_data_creator.update_schema_from_template("taskrelationselectiontemplate", test_data2)
PUT_and_assert_expected_response(self, url, test_data2, 200, expected_data2)
GET_OK_and_assert_equal_expected_response(self, url, expected_data2)
def test_task_relation_selection_template_PATCH(self):
# POST new item, verify
test_data = test_data_creator.TaskRelationSelectionTemplate()
expected_data = test_data_creator.update_schema_from_template("taskrelationselectiontemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_selection_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
test_patch = {"name": "new_name",
"description": "better description",
"schema": minimal_json_schema(properties={"mykey": {"type":"string", "default":"my better value"}})}
# PATCH item and verify
expected_patch_data = test_data_creator.update_schema_from_template("taskrelationselectiontemplate", test_patch)
PATCH_and_assert_expected_response(self, url, test_patch, 200, expected_patch_data)
expected_data = dict(test_data)
expected_data.update(expected_patch_data)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_task_relation_selection_template_DELETE(self):
# POST new item, verify
test_data = test_data_creator.TaskRelationSelectionTemplate()
expected_data = test_data_creator.update_schema_from_template("taskrelationselectiontemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_selection_template/', test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_GET_task_relation_selection_template_view_returns_correct_entry(self):
test_data_1 = TaskRelationSelectionTemplate_test_data("task_relation_selection_template_1")
test_data_2 = TaskRelationSelectionTemplate_test_data("task_relation_selection_template_2")
id1 = models.TaskRelationSelectionTemplate.objects.create(**test_data_1).id
id2 = models.TaskRelationSelectionTemplate.objects.create(**test_data_2).id
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_relation_selection_template/' + str(id1) + '/', test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_relation_selection_template/' + str(id2) + '/', test_data_2)
class TaskConnectorTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.task_template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
def test_task_connector_list_apiformat(self):
r = requests.get(BASE_URL + '/task_connector_type/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Connector Type List" in r.content.decode('utf8'))
def test_task_connector_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_connector_type/1234321/', 404)
def test_task_connector_POST_and_GET(self):
tc_test_data = test_data_creator.TaskConnectorType(task_template_url=self.task_template_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connector_type/', tc_test_data, 201, tc_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tc_test_data)
def test_task_connector_POST_invalid_role_raises_error(self):
# POST a new item with invalid choice
test_data_invalid_role = dict(test_data_creator.TaskConnectorType(task_template_url=self.task_template_url))
test_data_invalid_role['role'] = BASE_URL + '/role/forbidden/'
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connector_type/', test_data_invalid_role, 400, {})
self.assertTrue('Invalid hyperlink' in str(r_dict['role']))
def test_task_connector_POST_invalid_datatype_raises_error(self):
# POST a new item with invalid choice
test_data_invalid = dict(test_data_creator.TaskConnectorType(task_template_url=self.task_template_url))
test_data_invalid['datatype'] = BASE_URL + '/datatype/forbidden/'
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connector_type/', test_data_invalid, 400, {})
self.assertTrue('Invalid hyperlink' in str(r_dict['datatype']))
def test_task_connector_POST_invalid_dataformats_raises_error(self):
# POST a new item with invalid choice
test_data_invalid = dict(test_data_creator.TaskConnectorType(task_template_url=self.task_template_url))
test_data_invalid['dataformat'] = BASE_URL + '/dataformat/forbidden/'
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connector_type/', test_data_invalid, 400, {})
self.assertTrue('Invalid hyperlink' in str(r_dict['dataformat']))
def test_task_connector_POST_nonexistant_task_template_raises_error(self):
# POST a new item with wrong reference
test_data_invalid = dict(test_data_creator.TaskConnectorType(task_template_url=self.task_template_url))
test_data_invalid['task_template'] = BASE_URL + "/task_template/6353748/"
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connector_type/', test_data_invalid, 400, {})
self.assertTrue('Invalid hyperlink' in str(r_dict['task_template']))
def test_task_connector_POST_existing_outputs_works(self):
# First POST a new item to reference
test_data = test_data_creator.TaskTemplate()
expected_data = test_data_creator.update_schema_from_template("tasktemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', test_data, 201, expected_data)
url = r_dict['url']
# POST a new item with correct reference
test_data_valid = dict(test_data_creator.TaskConnectorType(task_template_url=self.task_template_url))
test_data_valid['task_template'] = url
POST_and_assert_expected_response(self, BASE_URL + '/task_connector_type/', test_data_valid, 201, test_data_valid)
def test_task_connector_PUT_nonexistant_raises_error(self):
PUT_and_assert_expected_response(self, BASE_URL + '/task_connector_type/9876789876/', test_data_creator.TaskConnectorType(task_template_url=self.task_template_url), 404, {})
def test_task_connector_PUT(self):
task_template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
tc_test_data1 = test_data_creator.TaskConnectorType(role="correlator", task_template_url=task_template_url)
tc_test_data2 = test_data_creator.TaskConnectorType(role="beamformer", task_template_url=task_template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connector_type/', tc_test_data1, 201, tc_test_data1)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tc_test_data1)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, tc_test_data2, 200, tc_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, tc_test_data2)
def test_task_connector_PATCH(self):
tc_test_data = test_data_creator.TaskConnectorType(task_template_url=self.task_template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connector_type/', tc_test_data, 201, tc_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tc_test_data)
test_patch = {"role": BASE_URL + '/role/calibrator',
"dataformat": BASE_URL + '/dataformat/Beamformed'}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(tc_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_task_connector_DELETE(self):
tc_test_data = test_data_creator.TaskConnectorType(task_template_url=self.task_template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connector_type/', tc_test_data, 201, tc_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tc_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_task_relation_blueprint_CASCADE_behavior_on_template_deleted(self):
task_template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
tc_test_data = test_data_creator.TaskConnectorType(task_template_url=task_template_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_connector_type/', tc_test_data, 201, tc_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, tc_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, task_template_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_GET_task_connector_view_returns_correct_entry(self):
test_data_1 = TaskConnectorType_test_data()
test_data_2 = TaskConnectorType_test_data()
id1 = models.TaskConnectorType.objects.create(**test_data_1).id
id2 = models.TaskConnectorType.objects.create(**test_data_2).id
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_connector_type/' + str(id1) + '/', test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_connector_type/' + str(id2) + '/', test_data_2)
class DefaultTemplates(unittest.TestCase):
def test_default_generator_template_POST(self):
test_data = test_data_creator.GeneratorTemplate()
expected_data = test_data_creator.update_schema_from_template("generatortemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', test_data, 201, expected_data)
url = r_dict['url']
test_data_1 = dict(test_data_creator.DefaultTemplates())
test_data_1['template'] = url
expected_data_1 = test_data_creator.update_schema_from_template("generatortemplate", test_data_1)
POST_and_assert_expected_response(self, BASE_URL + '/default_generator_template/', test_data_1, 201, expected_data_1)
def test_default_scheduling_unit_template_POST(self):
test_data = test_data_creator.SchedulingUnitTemplate()
expected_data = test_data_creator.update_schema_from_template("schedulingunittemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/',
test_data, 201,
expected_data)
url = r_dict['url']
test_data_1 = dict(test_data_creator.DefaultTemplates())
test_data_1['template'] = url
expected_data_1 = test_data_creator.update_schema_from_template("schedulingunittemplate", test_data_1)
POST_and_assert_expected_response(self, BASE_URL + '/default_scheduling_unit_template/', test_data_1, 201, expected_data_1)
def test_default_scheduling_constraints_template_POST(self):
test_data = test_data_creator.SchedulingConstraintsTemplate()
expected_data = test_data_creator.update_schema_from_template("schedulingconstraintstemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_constraints_template/',
test_data, 201,
expected_data)
url = r_dict['url']
test_data_1 = dict(test_data_creator.DefaultTemplates())
test_data_1['template'] = url
expected_data_1 = test_data_creator.update_schema_from_template("schedulingconstraintstemplate", test_data_1)
POST_and_assert_expected_response(self, BASE_URL + '/default_scheduling_constraints_template/', test_data_1, 201, expected_data_1)
def test_default_task_template_POST(self):
test_data = test_data_creator.TaskTemplate()
expected_data = test_data_creator.update_schema_from_template("tasktemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/',
test_data, 201,
expected_data)
url = r_dict['url']
test_data_1 = dict(test_data_creator.DefaultTemplates())
test_data_1['template'] = url
expected_data_1 = test_data_creator.update_schema_from_template("tasktemplate", test_data_1)
POST_and_assert_expected_response(self, BASE_URL + '/default_task_template/', test_data_1, 201, expected_data_1)
def test_default_task_relation_selection_template_POST(self):
test_data = test_data_creator.TaskRelationSelectionTemplate()
expected_data = test_data_creator.update_schema_from_template("taskrelationselectiontemplate", test_data)
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_selection_template/',
test_data, 201,
expected_data)
url = r_dict['url']
test_data_1 = dict(test_data_creator.DefaultTemplates())
test_data_1['template'] = url
expected_data_1 = test_data_creator.update_schema_from_template("taskrelationselectiontemplate", test_data_1)
POST_and_assert_expected_response(self, BASE_URL + '/default_task_relation_selection_template/', test_data_1, 201, expected_data_1)
def test_default_generator_template_PROTECT_behavior_on_template_deleted(self):
# POST with dependency
test_data = test_data_creator.GeneratorTemplate()
expected_data = test_data_creator.update_schema_from_template("generatortemplate", test_data)
template_url = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/',
test_data, 201,
expected_data)['url']
test_data2 = dict(test_data_creator.DefaultTemplates("defaulttemplate2"))
test_data2['template'] = template_url
expected_data2 = test_data_creator.update_schema_from_template("defaultgeneratortemplate", test_data2)
POST_and_assert_expected_response(self, BASE_URL + '/default_generator_template/',
test_data2, 201, expected_data2)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, template_url, expected_data)
def test_default_scheduling_unit_template_PROTECT_behavior_on_template_deleted(self):
# POST with dependency
test_data = test_data_creator.SchedulingUnitTemplate()
expected_data = test_data_creator.update_schema_from_template("schedulingunittemplate", test_data)
template_url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/',
test_data, 201,
expected_data)['url']
test_data2 = dict(test_data_creator.DefaultTemplates("defaulttemplate2"))
test_data2['template'] = template_url
expected_data2 = test_data_creator.update_schema_from_template("defaultschedulingunittemplate", test_data2)
POST_and_assert_expected_response(self, BASE_URL + '/default_scheduling_unit_template/',
test_data2, 201, expected_data2)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, template_url, expected_data)
def test_default_scheduling_constraints_template_PROTECT_behavior_on_template_deleted(self):
# POST with dependency
test_data = test_data_creator.SchedulingConstraintsTemplate()
expected_data = test_data_creator.update_schema_from_template("schedulingconstraintstemplate", test_data)
template_url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_constraints_template/',
test_data, 201,
expected_data)['url']
test_data2 = dict(test_data_creator.DefaultTemplates("defaulttemplate2"))
test_data2['template'] = template_url
expected_data2 = test_data_creator.update_schema_from_template("defaultschedulingconstraintstemplate", test_data2)
POST_and_assert_expected_response(self, BASE_URL + '/default_scheduling_constraints_template/',
test_data2, 201, expected_data2)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, template_url, expected_data)
def test_default_task_template_PROTECT_behavior_on_template_deleted(self):
# POST with dependency
test_data = test_data_creator.TaskTemplate()
expected_data = test_data_creator.update_schema_from_template("tasktemplate", test_data)
template_url = POST_and_assert_expected_response(self, BASE_URL + '/task_template/',
test_data, 201,
expected_data)['url']
test_data2 = dict(test_data_creator.DefaultTemplates("defaulttemplate2"))
test_data2['template'] = template_url
expected_data2 = test_data_creator.update_schema_from_template("defaulttasktemplate", test_data2)
POST_and_assert_expected_response(self, BASE_URL + '/default_task_template/',
test_data2, 201, expected_data2)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, template_url, expected_data)
def test_default_task_relation_selection_template_PROTECT_behavior_on_template_deleted(self):
# POST with dependency
test_data = test_data_creator.TaskRelationSelectionTemplate()
expected_data = test_data_creator.update_schema_from_template("taskrelationselectiontemplate", test_data)
template_url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_selection_template/',
test_data, 201,
expected_data)['url']
test_data2 = dict(test_data_creator.DefaultTemplates("defaulttemplate2"))
test_data2['template'] = template_url
expected_data2 = test_data_creator.update_schema_from_template("defaulttaskrelationselectiontemplate", test_data2)
POST_and_assert_expected_response(self, BASE_URL + '/default_task_relation_selection_template/',
test_data2, 201, expected_data2)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, template_url, expected_data)
class CycleTestCase(unittest.TestCase):
def test_cycle_list_apiformat(self):
r = requests.get(BASE_URL + '/cycle/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Cycle List" in r.content.decode('utf8'))
def test_cycle_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/cycle/1234321/', 404)
def test_cycle_POST_and_GET(self):
# POST and GET a new item and assert correctness
cycle_test_data = test_data_creator.Cycle()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', cycle_test_data, 201, cycle_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, cycle_test_data)
def test_cycle_PUT_invalid_raises_error(self):
PUT_and_assert_expected_response(self, BASE_URL + '/cycle/9876789876/', test_data_creator.Cycle(), 404, {})
def test_cycle_PUT(self):
cycle_test_data = test_data_creator.Cycle()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', cycle_test_data, 201, cycle_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, cycle_test_data)
# PUT new values, verify
test_data = dict(test_data_creator.Cycle("other description"))
test_data['name'] = cycle_test_data['name'] # since name is PK, need to keep that unchanged
PUT_and_assert_expected_response(self, url, test_data, 200, test_data)
GET_OK_and_assert_equal_expected_response(self, url, test_data)
def test_cycle_PATCH(self):
cycle_test_data = test_data_creator.Cycle()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', cycle_test_data, 201, cycle_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, cycle_test_data)
test_patch = {"start": datetime(year=2015, month=10, day=21).isoformat()}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(cycle_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_cycle_DELETE(self):
cycle_test_data = test_data_creator.Cycle()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', cycle_test_data, 201, cycle_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, cycle_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_GET_cycle_list_shows_entry(self):
test_data_1 = Cycle_test_data() # uuid makes name unique
test_data_1["start"] = datetime(2999, 1, 1) # cycles are ordered by start, so make this the latest date and hence the latest cycle
models.Cycle.objects.create(**test_data_1)
nbr_results = models.Cycle.objects.count()
test_data_1["start"] = test_data_1["start"].isoformat()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/cycle/', test_data_1, nbr_results)
def test_GET_cycle_view_returns_correct_entry(self):
test_data_1 = Cycle_test_data() # uuid makes name unique
test_data_2 = Cycle_test_data()
id1 = models.Cycle.objects.create(**test_data_1).name # name is pk
id2 = models.Cycle.objects.create(**test_data_2).name # name is pk
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/cycle/' + str(id1) + '/', test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/cycle/' + str(id2) + '/', test_data_2)
def test_cycle_contains_list_of_releated_projects(self):
cycle_test_data_1 = Cycle_test_data()
project_test_data_1 = Project_test_data() # uuid makes name unique
project_test_data_2 = Project_test_data() # uuid makes name unique
cycle = models.Cycle.objects.create(**cycle_test_data_1)
project1 = models.Project.objects.create(**project_test_data_1)
project1.cycles.set([cycle])
project1.save()
project2 = models.Project.objects.create(**project_test_data_2)
project2.cycles.set([cycle])
project2.save()
response_data = GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/cycle/' + cycle.name + '/', cycle_test_data_1)
assertUrlList(self, response_data['projects'], [project1, project2])
class ProjectTestCase(unittest.TestCase):
def test_project_list_apiformat(self):
r = requests.get(BASE_URL + '/project/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Project List" in r.content.decode('utf8'))
def test_project_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/project/1234321/', 404)
def test_project_POST_and_GET(self):
project_test_data = test_data_creator.Project()
# POST and GET a new item and assert correctness
expected = project_test_data.copy()
expected.pop('quota') # exclude quota from comparison, because these get auto-generated
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', project_test_data, 201, expected)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected)
def test_project_PUT_invalid_raises_error(self):
PUT_and_assert_expected_response(self, BASE_URL + '/project/9876789876/', test_data_creator.Project(), 404, {})
def test_project_PUT(self):
project_test_data = test_data_creator.Project()
# POST new item, verify
expected = project_test_data.copy()
expected.pop('quota') # exclude quota from comparison, because these get auto-generated
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', project_test_data, 201, expected)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected)
# PUT new values, verify
test_data = dict(test_data_creator.Project("other description"))
test_data['name'] = project_test_data['name'] # since name is PK, need to keep that unchanged
expected = test_data.copy()
expected.pop('quota') # exclude quota from comparison, because these get auto-generated
PUT_and_assert_expected_response(self, url, test_data, 200, expected)
GET_OK_and_assert_equal_expected_response(self, url, expected)
def test_project_PATCH(self):
project_test_data = test_data_creator.Project()
# POST new item, verify
expected = project_test_data.copy()
expected.pop('quota') # exclude quota from comparison, because these get auto-generated
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', project_test_data, 201, expected)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected)
test_patch = {"priority_rank": 1.0,
"tags": ["SUPERIMPORTANT"]}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected)
def test_project_DELETE(self):
project_test_data = test_data_creator.Project()
# POST new item, verify
expected = project_test_data.copy()
expected.pop('quota') # exclude quota from comparison, because these get auto-generated
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', project_test_data, 201, expected)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected)
# DELETE related auto-generated quota first
quotas = r_dict['quota']
for quota in quotas:
DELETE_and_assert_gone(self, quota)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_project_PROTECT_behavior_on_cycle_deleted(self):
# POST new item with dependencies
cycle_test_data = test_data_creator.Cycle()
cycle_url = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', cycle_test_data, 201, cycle_test_data)['url']
test_data = dict(test_data_creator.Project())
test_data['cycles'] = [cycle_url]
expected = test_data.copy()
expected.pop('quota') # exclude quota from comparison, because these get auto-generated
url = POST_and_assert_expected_response(self, BASE_URL + '/project/', test_data, 201, expected)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, expected)
# add project reference to cycle test data (we make Django add that to the cycle in serializer)
cycle_test_data['projects'] = [url]
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(cycle_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, cycle_url, cycle_test_data)
def test_GET_project_list_shows_entry(self):
test_data_1 = Project_test_data() # uuid makes name unique
test_data_1["name"] = "ZZZZZZZZZZZZZZZ" # projects are ordered by name, so make this the latest project (in sorted alphabetical order)
models.Project.objects.create(**test_data_1)
nbr_results = models.Project.objects.count()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/project/', test_data_1, nbr_results)
def test_GET_project_view_returns_correct_entry(self):
test_data_1 = Project_test_data() # uuid makes name unique
test_data_2 = Project_test_data()
id1 = models.Project.objects.create(**test_data_1).name # name is pk
id2 = models.Project.objects.create(**test_data_2).name
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/project/' + str(id1) + '/', test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/project/' + str(id2) + '/', test_data_2)
def test_nested_projects_are_filtered_according_to_cycle(self):
test_data_1 = dict(Project_test_data()) # uuid makes project unique
project_1 = models.Project.objects.create(**test_data_1)
cycle_1 = models.Cycle.objects.create(**Cycle_test_data())
project_1.cycles.set([cycle_1])
# assert the returned list contains related items, A list of length 1 is retrieved
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/cycle/%s/project/' % cycle_1.name, test_data_1, 1)
class ResourceTypeTestCase(unittest.TestCase):
def test_resource_type_list_apiformat(self):
r = requests.get(BASE_URL + '/resource_type/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Resource Type List" in r.content.decode('utf8'))
def test_resource_type_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/resource_type/1234321/', 404)
def test_resource_type_POST_and_GET(self):
resource_type_test_data = test_data_creator.ResourceType()
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/resource_type/', resource_type_test_data, 201, resource_type_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, resource_type_test_data)
class ProjectQuotaTestCase(unittest.TestCase):
def test_project_quota_list_apiformat(self):
r = requests.get(BASE_URL + '/project_quota/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Project Quota List" in r.content.decode('utf8'))
def test_project_quota_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/project_quota/1234321/', 404)
def test_project_quota_POST_and_GET(self):
project_quota_test_data = test_data_creator.ProjectQuota()
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project_quota/', project_quota_test_data, 201, project_quota_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, project_quota_test_data)
def test_project_quota_POST_and_GET(self):
project_quota_test_data = test_data_creator.ProjectQuota()
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project_quota/', project_quota_test_data, 201, project_quota_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, project_quota_test_data)
def test_project_quota_PUT_invalid_raises_error(self):
PUT_and_assert_expected_response(self, BASE_URL + '/project_quota/9876789876/', test_data_creator.Project(), 404, {})
def test_project_quota_PUT(self):
project_quota_test_data = test_data_creator.ProjectQuota()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project_quota/', project_quota_test_data, 201, project_quota_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, project_quota_test_data)
# PUT new values, verify
test_data = dict(test_data_creator.ProjectQuota("other description"))
PUT_and_assert_expected_response(self, url, test_data, 200, test_data)
GET_OK_and_assert_equal_expected_response(self, url, test_data)
def test_project_quota_PATCH(self):
project_quota_test_data = test_data_creator.ProjectQuota()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project_quota/', project_quota_test_data, 201, project_quota_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, project_quota_test_data)
test_patch = {"value": 500}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(project_quota_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_project_quota_DELETE(self):
project_quota_test_data = test_data_creator.ProjectQuota()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project_quota/', project_quota_test_data, 201, project_quota_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, project_quota_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_project_quota_PROTECT_behavior_on_project_deleted(self):
# POST new item with dependencies
project_test_data = test_data_creator.Project()
expected = project_test_data.copy()
expected.pop('quota') # exclude quota from comparison, because these get auto-generated
project_url = POST_and_assert_expected_response(self, BASE_URL + '/project/', project_test_data, 201, expected)['url']
project_quota_test_data = dict(test_data_creator.ProjectQuota(project_url=project_url))
project_quota_url = POST_and_assert_expected_response(self, BASE_URL + '/project_quota/', project_quota_test_data, 201, project_quota_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, project_quota_url, project_quota_test_data)
project_test_data['project'] = [project_quota_url] # add the
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(project_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, project_quota_url, project_quota_test_data)
class SchedulingSetTestCase(unittest.TestCase):
def test_scheduling_set_list_apiformat(self):
r = requests.get(BASE_URL + '/scheduling_set/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Scheduling Set List" in r.content.decode('utf8'))
def test_scheduling_set_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_set/1234321/', 404)
def test_scheduling_set_POST_and_GET(self):
schedulingset_test_data = test_data_creator.SchedulingSet()
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', schedulingset_test_data, 201, schedulingset_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingset_test_data)
def test_scheduling_set_PUT_invalid_raises_error(self):
schedulingset_test_data = test_data_creator.SchedulingSet()
PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_set/9876789876/', schedulingset_test_data, 404, {})
def test_scheduling_set_PUT(self):
project_url = test_data_creator.post_data_and_get_url(test_data_creator.Project(), '/project/')
schedulingset_test_data = test_data_creator.SchedulingSet(project_url=project_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', schedulingset_test_data, 201, schedulingset_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingset_test_data)
schedulingset_test_data2 = test_data_creator.SchedulingSet("schedulingset2", project_url=project_url)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, schedulingset_test_data2, 200, schedulingset_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, schedulingset_test_data2)
def test_scheduling_set_PATCH(self):
schedulingset_test_data = test_data_creator.SchedulingSet()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', schedulingset_test_data, 201, schedulingset_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingset_test_data)
test_patch = {"description": "This is a new and improved description",
"generator_doc": {"foo": "xyz"}}
# PATCH item and verify
expected_patch_data = test_data_creator.update_schema_from_template("generatortemplate", test_patch)
PATCH_and_assert_expected_response(self, url, test_patch, 200, expected_patch_data)
expected_data = dict(schedulingset_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_patch_data)
def test_scheduling_set_DELETE(self):
schedulingset_test_data = test_data_creator.SchedulingSet()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', schedulingset_test_data, 201, schedulingset_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingset_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_scheduling_set_PROTECT_behavior_on_project_deleted(self):
project_url = test_data_creator.post_data_and_get_url(test_data_creator.Project(), '/project/')
schedulingset_test_data = test_data_creator.SchedulingSet(project_url=project_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', schedulingset_test_data, 201, schedulingset_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, schedulingset_test_data)
# fetch project data before we delete it (for later comparison)
project_test_data = GET_and_assert_equal_expected_code(self, project_url, 200)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(project_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, project_url, project_test_data)
def test_scheduling_set_SET_NULL_behavior_on_generator_template_deleted(self):
generator_template_url = test_data_creator.post_data_and_get_url(test_data_creator.GeneratorTemplate(), '/generator_template/')
schedulingset_test_data = test_data_creator.SchedulingSet(generator_template_url=generator_template_url)
# POST new item
test_data = dict(schedulingset_test_data)
url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', test_data, 201, test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, test_data)
# DELETE dependency
DELETE_and_assert_gone(self, generator_template_url)
# assert
test_data['generator_template'] = None
GET_OK_and_assert_equal_expected_response(self, url, test_data)
def test_GET_SchedulingSet_list_shows_entry(self):
test_data_1 = SchedulingSet_test_data()
models.SchedulingSet.objects.create(**test_data_1)
nbr_results = models.SchedulingSet.objects.count()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/scheduling_set/', test_data_1, nbr_results)
def test_GET_SchedulingSet_view_returns_correct_entry(self):
test_data_1 = SchedulingSet_test_data() # uuid makes name unique
test_data_2 = SchedulingSet_test_data()
id1 = models.SchedulingSet.objects.create(**test_data_1).id
id2 = models.SchedulingSet.objects.create(**test_data_2).id
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/scheduling_set/' + str(id1) + '/', test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/scheduling_set/' + str(id2) + '/', test_data_2)
def test_SchedulingSet_contains_list_of_releated_SchedulingUnitDraft(self):
test_data_1 = SchedulingSet_test_data()
scheduling_set = models.SchedulingSet.objects.create(**test_data_1)
scheduling_unit_draft_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data("scheduler draft one"))
scheduling_unit_draft_1.scheduling_set = scheduling_set
scheduling_unit_draft_1.save()
scheduling_unit_draft_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data("scheduler draft one"))
scheduling_unit_draft_2.scheduling_set = scheduling_set
scheduling_unit_draft_2.save()
response_data = GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/scheduling_set/%d/' % scheduling_set.id, test_data_1)
assertUrlList(self, response_data['scheduling_unit_drafts'], [scheduling_unit_draft_1, scheduling_unit_draft_2])
class SchedulingUnitDraftTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.scheduling_set_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingSet(), '/scheduling_set/')
cls.template_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitTemplate(), '/scheduling_unit_template/')
def test_scheduling_unit_draft_list_apiformat(self):
r = requests.get(BASE_URL + '/scheduling_unit_draft/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Scheduling Unit Draft List" in r.content.decode('utf8'))
def test_scheduling_unit_draft_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/1234321/', 404)
def test_scheduling_unit_draft_POST_and_GET(self):
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(scheduling_set_url=self.scheduling_set_url, template_url=self.template_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingunitdraft_test_data)
def test_scheduling_unit_draft_PUT_invalid_raises_error(self):
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(scheduling_set_url=self.scheduling_set_url, template_url=self.template_url)
PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/9876789876/', schedulingunitdraft_test_data, 404, {})
def test_scheduling_unit_draft_PUT(self):
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(scheduling_set_url=self.scheduling_set_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingunitdraft_test_data)
schedulingunitdraft_test_data2 = test_data_creator.SchedulingUnitDraft("my_scheduling_unit_draft2", scheduling_set_url=self.scheduling_set_url, template_url=self.template_url)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, schedulingunitdraft_test_data2, 200, schedulingunitdraft_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, schedulingunitdraft_test_data2)
def test_scheduling_unit_draft_PATCH(self):
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(scheduling_set_url=self.scheduling_set_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingunitdraft_test_data)
test_patch = {"description": "This is a new and improved description",
"requirements_doc": {"foo": "barbar"}}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(schedulingunitdraft_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_scheduling_unit_draft_DELETE(self):
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(scheduling_set_url=self.scheduling_set_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingunitdraft_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_scheduling_unit_draft_CASCADE_behavior_on_scheduling_unit_template_deleted(self):
template_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitTemplate(), '/scheduling_unit_template/')
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(template_url=template_url, scheduling_set_url=self.scheduling_set_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, schedulingunitdraft_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, template_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_scheduling_unit_draft_CASCADE_behavior_on_scheduling_set_deleted(self):
scheduling_set_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingSet(), '/scheduling_set/')
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(scheduling_set_url=scheduling_set_url, template_url=self.template_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, schedulingunitdraft_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, scheduling_set_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_scheduling_unit_draft_SET_NULL_behavior_on_copies_deleted(self):
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(scheduling_set_url=self.scheduling_set_url, template_url=self.template_url)
# POST new item with dependency
copy_url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)['url']
test_data = dict(schedulingunitdraft_test_data)
test_data['copies'] = copy_url
url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', test_data, 201, test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, test_data)
# DELETE dependency
DELETE_and_assert_gone(self, copy_url)
# assert
test_data['copies'] = None
GET_OK_and_assert_equal_expected_response(self, url, test_data)
def test_GET_SchedulingUnitDraft_list_view_shows_entry(self):
test_data_1 = SchedulingUnitDraft_test_data("scheduler unit draft one")
obj = models.SchedulingUnitDraft.objects.create(**test_data_1)
nbr_results = models.SchedulingUnitDraft.objects.count()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/scheduling_unit_draft/', test_data_1, nbr_results, obj.id)
def test_GET_SchedulingUnitDraft_view_returns_correct_entry(self):
test_data_1 = SchedulingUnitDraft_test_data("scheduler unit draft one one")
test_data_2 = SchedulingUnitDraft_test_data("scheduler unit draft one two")
id1 = models.SchedulingUnitDraft.objects.create(**test_data_1).id
id2 = models.SchedulingUnitDraft.objects.create(**test_data_2).id
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/scheduling_unit_draft/%s/' % id1, test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/scheduling_unit_draft/%s/' % id2, test_data_2)
def test_nested_SchedulingUnitDraft_are_filtered_according_to_SchedulingSet(self):
# setup
test_data_1 = SchedulingUnitDraft_test_data("scheduler unit draft two one")
sst_test_data_1 = SchedulingSet_test_data("scheduler set one")
scheduling_set_1 = models.SchedulingSet.objects.create(**sst_test_data_1)
test_data_1 = dict(test_data_1)
test_data_1['scheduling_set'] = scheduling_set_1
scheduling_unit_draft_1 = models.SchedulingUnitDraft.objects.create(**test_data_1)
# assert the returned list contains related items, A list of length 1 is retrieved
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/scheduling_set/%s/scheduling_unit_draft/'
% scheduling_set_1.id, test_data_1, 1)
def test_SchedulingUnitDraft_contains_list_of_related_SchedulingUnitBlueprint(self):
# setup
test_data_1 = SchedulingUnitDraft_test_data("scheduler unit draft one")
subt_test_data_1 = SchedulingUnitBlueprint_test_data("scheduler unit blue print one")
subt_test_data_2 = SchedulingUnitBlueprint_test_data("scheduler unit blue print two")
scheduling_unit_draft = models.SchedulingUnitDraft.objects.create(**test_data_1)
scheduling_unit_blueprint_1 = models.SchedulingUnitBlueprint.objects.create(**subt_test_data_1)
scheduling_unit_blueprint_1.draft = scheduling_unit_draft
scheduling_unit_blueprint_1.save()
scheduling_unit_blueprint_2 = models.SchedulingUnitBlueprint.objects.create(**subt_test_data_2)
scheduling_unit_blueprint_2.draft = scheduling_unit_draft
scheduling_unit_blueprint_2.save()
# assert
response_data = GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/scheduling_unit_draft/%s/' % scheduling_unit_draft.id, test_data_1)
assertUrlList(self, response_data['scheduling_unit_blueprints'], [scheduling_unit_blueprint_1, scheduling_unit_blueprint_2])
def test_SchedulingUnitDraft_contains_list_of_related_TaskDraft(self):
# setup
test_data_1 = SchedulingUnitDraft_test_data("scheduler unit draft one")
tdt_test_data_1 = TaskDraft_test_data("task draft one of su1")
tdt_test_data_2 = TaskDraft_test_data("task draft two of su2 ")
scheduling_unit_draft = models.SchedulingUnitDraft.objects.create(**test_data_1)
task_draft_1 = models.TaskDraft.objects.create(**tdt_test_data_1)
task_draft_1.scheduling_unit_draft = scheduling_unit_draft
task_draft_1.save()
task_draft_2 = models.TaskDraft.objects.create(**tdt_test_data_2)
task_draft_2.scheduling_unit_draft = scheduling_unit_draft
task_draft_2.save()
# assert
response_data = GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/scheduling_unit_draft/%s/' %
scheduling_unit_draft.id, test_data_1)
assertUrlList(self, response_data['task_drafts'], [task_draft_1, task_draft_2])
def test_GET_SchedulingUnitDraft_view_filters_for_project(self):
"""
Test we can filter on this property, which is explicitly named on the model-specific property filter
"""
# setup
project_1 = models.Project.objects.create(**Project_test_data(name='myproject_draft_partial_1_%s' % uuid.uuid4()))
project_2 = models.Project.objects.create(**Project_test_data(name='myproject_draft_partial_2_%s' % uuid.uuid4()))
scheduling_set_1 = models.SchedulingSet.objects.create(**SchedulingSet_test_data(project=project_1))
scheduling_set_2 = models.SchedulingSet.objects.create(**SchedulingSet_test_data(project=project_2))
su_draft_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud1_%s' % uuid.uuid4(), scheduling_set=scheduling_set_1))
su_draft_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud2_%s' % uuid.uuid4(), scheduling_set=scheduling_set_2))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?project=%s' % project_1.name, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?project=%s' % project_2.name, 200)
response_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?project=foo', 200)
response_4 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?project=myproject_draft_partia', 200)
self.assertEqual(response_1['count'], 1)
self.assertEqual(response_1['results'][0]['name'], su_draft_1.name)
self.assertEqual(response_2['count'], 1)
self.assertEqual(response_2['results'][0]['name'], su_draft_2.name)
self.assertEqual(response_3['count'], 0)
self.assertEqual(response_4['count'], 2)
self.assertEqual({response_4['results'][0]['name'], response_4['results'][1]['name']},
{su_draft_1.name, su_draft_2.name})
def test_GET_SchedulingUnitDraft_view_filters_for_id(self):
"""
Test we can filter on this field, which is explicitly named on the model-specific property filter
"""
# setup
su_draft_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud1_%s' % uuid.uuid4()))
su_draft_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud2_%s' % uuid.uuid4()))
su_draft_3 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud3_%s' % uuid.uuid4()))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?id=%s' % su_draft_1.id, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?id=%s,%s' % (su_draft_1.id, su_draft_3.id), 200)
response_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?id=%s' % 999999, 200)
response_4 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?id_min=%s&id_max=%s' % (su_draft_2.id, su_draft_3.id), 200)
self.assertEqual(response_1['count'], 1)
self.assertEqual(response_1['results'][0]['name'], su_draft_1.name)
self.assertEqual(response_2['count'], 2)
self.assertEqual(response_2['results'][0]['name'], su_draft_1.name)
self.assertEqual(response_2['results'][1]['name'], su_draft_3.name)
self.assertEqual(response_3['count'], 0)
self.assertEqual(response_4['count'], 2)
self.assertEqual(response_4['results'][0]['name'], su_draft_2.name)
self.assertEqual(response_4['results'][1]['name'], su_draft_3.name)
def test_GET_SchedulingUnitDraft_view_filters_for_blueprints(self):
"""
Test we can filter on this related field, which is explicitly named on the model-specific property filter
"""
# setup
models.SchedulingUnitBlueprint.objects.all().delete()
models.SchedulingUnitDraft.objects.all().delete()
su_draft_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud1_%s' % uuid.uuid4()))
su_draft_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud2_%s' % uuid.uuid4()))
su_draft_3 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud3_%s' % uuid.uuid4()))
su_blueprint_1 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(name='sub1_%s' % uuid.uuid4(), draft=su_draft_1))
su_blueprint_2 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(name='sub2_%s' % uuid.uuid4(), draft=su_draft_2))
su_blueprint_3 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(name='sub3_%s' % uuid.uuid4(), draft=su_draft_3))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?scheduling_unit_blueprints=%s' % su_blueprint_1.id, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?scheduling_unit_blueprints=%s&scheduling_unit_blueprints=%s' % (su_blueprint_1.id, su_blueprint_3.id), 200)
response_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?scheduling_unit_blueprints=%s' % 999999, 400)
self.assertEqual(response_1['count'], 1)
self.assertEqual(response_1['results'][0]['name'], su_draft_1.name)
self.assertEqual(response_2['count'], 2)
self.assertEqual(response_2['results'][0]['name'], su_draft_1.name)
self.assertEqual(response_2['results'][1]['name'], su_draft_3.name)
self.assertIn('Select a valid choice', str(response_3))
def test_GET_SchedulingUnitDraft_view_filters_for_priority_rank(self):
"""
Test we can filter on this property, which is explicitly named on the model-specific property filter
"""
# setup
su_draft_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud1_%s' % uuid.uuid4()), priority_rank=0.111)
su_draft_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud2_%s' % uuid.uuid4()), priority_rank=0.222)
su_draft_3 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud3_%s' % uuid.uuid4()), priority_rank=0.333)
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?priority_rank_min=%s&priority_rank_max=%s' % (0.110, 0.112), 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?priority_rank_min=%s&priority_rank_max=%s' % (0.2, 0.4), 200)
self.assertEqual(response_1['count'], 1)
self.assertEqual(response_1['results'][0]['name'], su_draft_1.name)
self.assertEqual(response_2['count'], 2)
self.assertEqual(response_2['results'][0]['name'], su_draft_2.name)
self.assertEqual(response_2['results'][1]['name'], su_draft_3.name)
def test_GET_SchedulingUnitDraft_view_filters_for_duration(self):
"""
Test we can filter on this property, which is explicitly named on the model-specific property filter
"""
# setup
models.SchedulingUnitBlueprint.objects.all().delete()
models.SchedulingUnitDraft.objects.all().delete()
su_draft_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud1_%s' % uuid.uuid4()))
su_draft_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud2_%s' % uuid.uuid4()))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?duration_min=%s&duration_max=%s' % ('PT1H', 'PT2H'), 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?duration_min=%s&duration_max=%s' % ('PT0S', 'PT1H'), 200)
self.assertEqual(response_1['count'], 0)
self.assertEqual(response_2['count'], 2)
self.assertEqual({result['name'] for result in response_2['results']},
{su_draft_1.name, su_draft_2.name})
def test_GET_SchedulingUnitDraft_view_filters_for_pointing(self):
"""
Test we can filter on this property, which is explicitly named on the model-specific property filter
"""
# setup
strategy_template = models.SchedulingUnitObservingStrategyTemplate.objects.get(name="UC1 CTC+pipelines")
specs_1 = add_defaults_to_json_object_for_schema(strategy_template.template,
strategy_template.scheduling_unit_template.schema)
specs_2 = add_defaults_to_json_object_for_schema(strategy_template.template,
strategy_template.scheduling_unit_template.schema)
specs_1['tasks']['Target Observation']['specifications_doc']['SAPs'][0]['digital_pointing'] = {'angle1': 0.333, 'angle2': 0.999, 'direction_type': 'J2000'}
specs_2['tasks']['Target Observation']['specifications_doc']['SAPs'][0]['digital_pointing'] = {'angle1': 0.111, 'angle2': 0.777, 'direction_type': 'J2000'}
su_draft_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud1_%s' % uuid.uuid4(), template=strategy_template.scheduling_unit_template, observation_strategy_template=strategy_template, requirements_doc=specs_1))
su_draft_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud2_%s' % uuid.uuid4(), template=strategy_template.scheduling_unit_template, observation_strategy_template=strategy_template, requirements_doc=specs_2))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?angle1_min=%s' % 0.222, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?angle2_max=%s' % 0.888, 200)
self.assertEqual(response_1['count'], 1)
self.assertEqual(response_1['results'][0]['name'], su_draft_1.name)
self.assertEqual(response_2['count'], 1)
self.assertEqual(response_2['results'][0]['name'], su_draft_2.name)
def test_GET_SchedulingUnitBlueprint_view_filters_for_strategy_template(self):
"""
Test we can filter on this field, which is explicitly named on the model-specific property filter
"""
# setup
template_1 = models.SchedulingUnitObservingStrategyTemplate.objects.create(**SchedulingUnitObservingStrategyTemplate_test_data(name='suost1_%s' % uuid.uuid4()))
template_2 = models.SchedulingUnitObservingStrategyTemplate.objects.create(**SchedulingUnitObservingStrategyTemplate_test_data(name='suost1_%s' % uuid.uuid4()))
su_draft_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud1_%s' % uuid.uuid4(), observation_strategy_template=template_1))
su_draft_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud2_%s' % uuid.uuid4(), observation_strategy_template=template_2))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?observation_strategy_template=%s' % template_1.id, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?observation_strategy_template_name=%s' % template_2.name[:-3], 200)
response_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?observation_strategy_template_name=%s' % 'gibberish', 200)
GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?observation_strategy_template=%s' % 999999, 400)
self.assertEqual(response_1['count'], 1)
self.assertEqual(response_1['results'][0]['name'], su_draft_1.name)
self.assertEqual(response_2['count'], 1)
self.assertEqual(response_2['results'][0]['name'], su_draft_2.name)
self.assertEqual(response_3['count'], 0)
class TaskDraftTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.scheduling_unit_draft_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitDraft(), '/scheduling_unit_draft/')
cls.template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
def test_task_draft_list_apiformat(self):
r = requests.get(BASE_URL + '/task_draft/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Draft List" in r.content.decode('utf8'))
def test_task_draft_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/1234321/', 404)
def test_task_draft_POST_and_GET(self):
taskdraft_test_data = test_data_creator.TaskDraft(scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data, 201, taskdraft_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, taskdraft_test_data)
def test_task_draft_PUT_invalid_raises_error(self):
taskdraft_test_data = test_data_creator.TaskDraft(scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
PUT_and_assert_expected_response(self, BASE_URL + '/task_draft/9876789876/', taskdraft_test_data, 404, {})
def test_task_draft_PUT(self):
taskdraft_test_data1 = test_data_creator.TaskDraft(name="the one", scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
taskdraft_test_data2 = test_data_creator.TaskDraft(name="the other", scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data1, 201, taskdraft_test_data1)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, taskdraft_test_data1)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, taskdraft_test_data2, 200, taskdraft_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, taskdraft_test_data2)
def test_task_draft_PATCH(self):
taskdraft_test_data = test_data_creator.TaskDraft(scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data, 201, taskdraft_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, taskdraft_test_data)
test_patch = {"description": "This is a new and improved description",
"specifications_doc": {"foo": "xyz"}}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(taskdraft_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_task_draft_DELETE(self):
taskdraft_test_data = test_data_creator.TaskDraft(scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data, 201, taskdraft_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, taskdraft_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_task_draft_CASCADE_behavior_on_task_template_deleted(self):
template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
taskdraft_test_data = test_data_creator.TaskDraft(name="task draft 2", template_url=template_url, scheduling_unit_draft_url=self.scheduling_unit_draft_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data, 201, taskdraft_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, taskdraft_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, template_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_draft_CASCADE_behavior_on_scheduling_unit_draft_deleted(self):
scheduling_unit_draft_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitDraft(), '/scheduling_unit_draft/')
taskdraft_test_data = test_data_creator.TaskDraft(name="task draft 2", scheduling_unit_draft_url=scheduling_unit_draft_url, template_url=self.template_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data, 201, taskdraft_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, taskdraft_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, scheduling_unit_draft_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_draft_SET_NULL_behavior_on_copies_deleted(self):
taskdraft_test_data1 = test_data_creator.TaskDraft(name="the one "+str(uuid.uuid4()), scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
taskdraft_test_data2 = test_data_creator.TaskDraft(name="the other "+str(uuid.uuid4()), scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
# POST new item with dependency
copy_url = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data2, 201, taskdraft_test_data2)['url']
test_data = dict(taskdraft_test_data1)
test_data['copies'] = copy_url
url = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', test_data, 201, test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, test_data)
# DELETE dependency
DELETE_and_assert_gone(self, copy_url)
# assert
test_data['copies'] = None
GET_OK_and_assert_equal_expected_response(self, url, test_data)
def test_GET_TaskDraft_list_view_shows_entry(self):
test_data_1 = TaskDraft_test_data("task draft")
obj = models.TaskDraft.objects.create(**test_data_1)
nbr_results = models.TaskDraft.objects.count()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/task_draft/', test_data_1, nbr_results, obj.id)
def test_GET_TaskDraft_view_returns_correct_entry(self):
# setup
test_data_1 = TaskDraft_test_data("task draft one")
test_data_2 = TaskDraft_test_data("task draft two")
id1 = models.TaskDraft.objects.create(**test_data_1).id
id2 = models.TaskDraft.objects.create(**test_data_2).id
# assert
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_draft/%s/' % id1, test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_draft/%s/' % id2, test_data_2)
def test_nested_TaskDraft_are_filtered_according_to_SchedulingUnitDraft(self):
# setup
test_data_1 = TaskDraft_test_data("task draft three")
sudt_test_data_1 = SchedulingUnitDraft_test_data("scheduling unit draft one")
scheduling_unit_draft_1 = models.SchedulingUnitDraft.objects.create(**sudt_test_data_1)
test_data_1 = dict(test_data_1)
test_data_1['scheduling_unit_draft'] = scheduling_unit_draft_1
task_draft_1 = models.TaskDraft.objects.create(**test_data_1)
# assert the returned list contains related items, A list of length 1 is retrieved
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/scheduling_unit_draft/%s/task_draft/' % scheduling_unit_draft_1.id, test_data_1, 1)
def test_TaskDraft_contains_list_of_related_TaskBlueprint(self):
# setup
test_data_1 = TaskDraft_test_data("task draft four")
tbt_test_data_1 = TaskBlueprint_test_data()
tbt_test_data_2 = TaskBlueprint_test_data()
task_draft = models.TaskDraft.objects.create(**test_data_1)
task_blueprint_1 = models.TaskBlueprint.objects.create(**tbt_test_data_1)
task_blueprint_1.draft = task_draft
task_blueprint_1.save()
task_blueprint_2 = models.TaskBlueprint.objects.create(**tbt_test_data_2)
task_blueprint_2.draft = task_draft
task_blueprint_2.save()
# assert
response_data = GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_draft/%s/' % task_draft.id, test_data_1)
assertUrlList(self, response_data['task_blueprints'], [task_blueprint_1, task_blueprint_2])
def test_TaskDraft_contains_lists_of_related_TaskRelationDraft(self):
# setup
test_data_1 = TaskDraft_test_data("task draft five")
task_draft = models.TaskDraft.objects.create(**test_data_1)
trdt_test_data_1 = TaskRelationDraft_test_data()
trdt_test_data_2 = TaskRelationDraft_test_data()
task_relation_draft_1 = models.TaskRelationDraft.objects.create(**trdt_test_data_1)
task_relation_draft_1.producer = task_draft
task_relation_draft_1.save()
task_relation_draft_2 = models.TaskRelationDraft.objects.create(**trdt_test_data_2)
task_relation_draft_2.consumer = task_draft
task_relation_draft_2.save()
# assert
response_data = GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_draft/%s/' % task_draft.id, test_data_1)
# consumed_by and produced_by might appear to be swapped, but they are actually correct. Depends on the angle you're looking at it.
assertUrlList(self, response_data['consumed_by'], [task_relation_draft_1])
assertUrlList(self, response_data['produced_by'], [task_relation_draft_2])
def test_GET_TaskDraft_view_filters_for_copy_reason(self):
"""
Test we can filter on this model field, because the parent LOFARViewSet uses filtering on __all__ fields
We only test that we get an error if we filter for an invalid option, as proof that filtering is enabled,
and assume that the filter backend does the correct thing.
"""
# assert
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/?copy_reason=template', 200)
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/?copy_reason=gibberish', 400)
def test_TaskDraft_filters_for_TaskBlueprints(self):
# setup
task_draft_1 = models.TaskDraft.objects.create(**TaskDraft_test_data("td_%s" % uuid.uuid4()))
task_draft_2 = models.TaskDraft.objects.create(**TaskDraft_test_data("td_%s" % uuid.uuid4()))
task_draft_3 = models.TaskDraft.objects.create(**TaskDraft_test_data("td_%s" % uuid.uuid4()))
task_blueprint_1 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data())
task_blueprint_1.draft = task_draft_1
task_blueprint_1.save()
task_blueprint_2 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data())
task_blueprint_2.draft = task_draft_2
task_blueprint_2.save()
task_blueprint_3 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data())
task_blueprint_3.draft = task_draft_3
task_blueprint_3.save()
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/?task_blueprints=%s' % task_blueprint_1.id, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/?task_blueprints=%s' % task_blueprint_2.id, 200)
response_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/?task_blueprints=%s&task_blueprints=%s' % (task_blueprint_1.id, task_blueprint_3.id), 200)
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/?task_blueprints=gibberish', 400)
self.assertIn(task_draft_1.name, str(response_1))
self.assertNotIn(task_draft_2.name, str(response_1))
self.assertNotIn(task_draft_3.name, str(response_1))
self.assertNotIn(task_draft_1.name, str(response_2))
self.assertIn(task_draft_2.name, str(response_2))
self.assertNotIn(task_draft_3.name, str(response_2))
self.assertIn(task_draft_1.name, str(response_3))
self.assertNotIn(task_draft_2.name, str(response_3))
self.assertIn(task_draft_3.name, str(response_3))
def test_TaskDraft_filters_for_SchedulingUnitDraft(self):
# setup
su_draft_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data("sud_%s" % uuid.uuid4()))
su_draft_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data("sud_%s" % uuid.uuid4()))
su_draft_3 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data("sud_%s" % uuid.uuid4()))
task_draft_1 = models.TaskDraft.objects.create(**TaskDraft_test_data("td_%s" % uuid.uuid4(), scheduling_unit_draft=su_draft_1))
task_draft_2 = models.TaskDraft.objects.create(**TaskDraft_test_data("td_%s" % uuid.uuid4(), scheduling_unit_draft=su_draft_2))
task_draft_3 = models.TaskDraft.objects.create(**TaskDraft_test_data("td_%s" % uuid.uuid4(), scheduling_unit_draft=su_draft_3))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/?scheduling_unit_draft=%s' % su_draft_1.id, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/?scheduling_unit_draft=%s,%s' % (su_draft_2.id, su_draft_3.id), 200)
response_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/?scheduling_unit_draft_min=%s&scheduling_unit_draft_max=%s' % (su_draft_3.id, su_draft_3.id), 200)
response_4 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/?scheduling_unit_draft_name=%s' % su_draft_2.name[:-3], 200)
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/?scheduling_unit_draft=%s' % 9999999, 400)
self.assertEqual(response_1['count'], 1)
self.assertIn(task_draft_1.name, str(response_1))
self.assertNotIn(task_draft_2.name, str(response_1))
self.assertNotIn(task_draft_3.name, str(response_1))
self.assertEqual(response_2['count'], 2)
self.assertNotIn(task_draft_1.name, str(response_2))
self.assertIn(task_draft_2.name, str(response_2))
self.assertIn(task_draft_3.name, str(response_2))
self.assertEqual(response_3['count'], 1)
self.assertNotIn(task_draft_1.name, str(response_3))
self.assertNotIn(task_draft_2.name, str(response_3))
self.assertIn(task_draft_3.name, str(response_3))
self.assertEqual(response_4['count'], 1)
self.assertNotIn(task_draft_1.name, str(response_4))
self.assertIn(task_draft_2.name, str(response_4))
self.assertNotIn(task_draft_3.name, str(response_4))
def test_TaskDraft_filters_for_project(self):
"""
Test we can filter on this property, which is explicitly named on the model-specific property filter
"""
# setup
project_1 = models.Project.objects.create(**Project_test_data(name='myproject_draft_partial_1_%s' % uuid.uuid4()))
project_2 = models.Project.objects.create(**Project_test_data(name='myproject_draft_partial_2_%s' % uuid.uuid4()))
scheduling_set_1 = models.SchedulingSet.objects.create(**SchedulingSet_test_data(project=project_1))
scheduling_set_2 = models.SchedulingSet.objects.create(**SchedulingSet_test_data(project=project_2))
su_draft_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud1_%s' % uuid.uuid4(), scheduling_set=scheduling_set_1))
su_draft_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud2_%s' % uuid.uuid4(), scheduling_set=scheduling_set_2))
task_draft_1 = models.TaskDraft.objects.create(**TaskDraft_test_data("td_%s" % uuid.uuid4(), scheduling_unit_draft=su_draft_1))
task_draft_2 = models.TaskDraft.objects.create(**TaskDraft_test_data("td_%s" % uuid.uuid4(), scheduling_unit_draft=su_draft_2))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/?project=%s' % project_1.name, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/?project=%s' % project_2.name, 200)
response_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/?project=foo', 200)
response_4 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_draft/?project=myproject_draft_partia', 200)
self.assertEqual(response_1['count'], 1)
self.assertEqual(response_1['results'][0]['name'], task_draft_1.name)
self.assertEqual(response_2['count'], 1)
self.assertEqual(response_2['results'][0]['name'], task_draft_2.name)
self.assertEqual(response_3['count'], 0)
self.assertEqual(response_4['count'], 2)
self.assertEqual({response_4['results'][0]['name'], response_4['results'][1]['name']},
{task_draft_1.name, task_draft_2.name})
class TaskRelationDraftTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.producer_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskDraft(), '/task_draft/')
cls.consumer_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskDraft(), '/task_draft/')
cls.template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskRelationSelectionTemplate(), '/task_relation_selection_template/')
cls.task_template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
cls.input_role_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(iotype="input", task_template_url=cls.task_template_url), '/task_connector_type/')
cls.output_role_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(iotype="output", task_template_url=cls.task_template_url), '/task_connector_type/')
def test_task_relation_draft_list_apiformat(self):
r = requests.get(BASE_URL + '/task_relation_draft/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Relation Draft List" in r.content.decode('utf8'))
def test_task_relation_draft_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_relation_draft/1234321/', 404)
def test_task_relation_draft_POST_and_GET(self):
trd_test_data = test_data_creator.TaskRelationDraft(producer_url=self.producer_url, consumer_url=self.consumer_url, template_url=self.template_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/', trd_test_data, 201, trd_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, trd_test_data)
def test_task_relation_draft_PUT_invalid_raises_error(self):
trd_test_data = test_data_creator.TaskRelationDraft(producer_url=self.producer_url, consumer_url=self.consumer_url, template_url=self.template_url, input_role_url=self.input_role_url)
PUT_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/9876789876/', trd_test_data, 404, {})
def test_task_relation_draft_PUT(self):
task_template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
input_role_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(iotype="input", task_template_url=task_template_url), '/task_connector_type/')
trd_test_data1 = test_data_creator.TaskRelationDraft(producer_url=self.producer_url, consumer_url=self.consumer_url, template_url=self.template_url, input_role_url=input_role_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/', trd_test_data1, 201, trd_test_data1)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, trd_test_data1)
# PUT new values, verify
task_template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
input_role_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(iotype="input", task_template_url=task_template_url), '/task_connector_type/')
new_consumer_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskDraft(), '/task_draft/')
trd_test_data2 = test_data_creator.TaskRelationDraft(producer_url=self.producer_url, consumer_url=new_consumer_url, template_url=self.template_url, input_role_url=input_role_url)
PUT_and_assert_expected_response(self, url, trd_test_data2, 200, trd_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, trd_test_data2)
def test_task_relation_draft_PATCH(self):
trd_test_data = test_data_creator.TaskRelationDraft(producer_url=self.producer_url, consumer_url=self.consumer_url, template_url=self.template_url, input_role_url=self.input_role_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/', trd_test_data, 201, trd_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, trd_test_data)
test_patch = {"selection_doc": {"foo": "patched"}}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(trd_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_task_relation_draft_DELETE(self):
trd_test_data = test_data_creator.TaskRelationDraft(producer_url=self.producer_url, consumer_url=self.consumer_url, template_url=self.template_url, input_role_url=self.input_role_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/', trd_test_data, 201, trd_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, trd_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_task_relation_draft_CASCADE_behavior_on_task_relation_selection_template_deleted(self):
template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskRelationSelectionTemplate(), '/task_relation_selection_template/')
trd_test_data = test_data_creator.TaskRelationDraft(template_url=template_url, producer_url=self.producer_url, consumer_url=self.consumer_url, input_role_url=self.input_role_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/', trd_test_data, 201, trd_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, trd_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, template_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_relation_draft_CASCADE_behavior_on_producer_deleted(self):
producer_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskDraft(), '/task_draft/')
trd_test_data = test_data_creator.TaskRelationDraft(producer_url=producer_url, consumer_url=self.consumer_url, template_url=self.template_url, input_role_url=self.input_role_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/',
trd_test_data, 201, trd_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, trd_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, producer_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_relation_draft_CASCADE_behavior_on_consumer_deleted(self):
consumer_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskDraft(), '/task_draft/')
trd_test_data = test_data_creator.TaskRelationDraft(consumer_url=consumer_url, producer_url=self.producer_url, template_url=self.template_url, input_role_url=self.input_role_url)
# POST new item with dependency
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/',
trd_test_data, 201, trd_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, trd_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, consumer_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_relation_draft_CASCADE_behavior_on_input_deleted(self):
task_template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
input_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(task_template_url=task_template_url), '/task_connector_type/')
trd_test_data = test_data_creator.TaskRelationDraft(input_role_url=input_url, producer_url=self.producer_url, consumer_url=self.consumer_url, template_url=self.template_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/',
trd_test_data, 201, trd_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, trd_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, input_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_relation_draft_CASCADE_behavior_on_output_deleted(self):
task_template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
output_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(task_template_url=task_template_url), '/task_connector_type/')
trd_test_data = test_data_creator.TaskRelationDraft(output_role_url=output_url, producer_url=self.producer_url, consumer_url=self.consumer_url, template_url=self.template_url, input_role_url=self.input_role_url)
# POST new item with dependency
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/',
trd_test_data, 201, trd_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, trd_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, output_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_GET_TaskRelationDraft_list_view_shows_entry(self):
test_data_1 = TaskRelationDraft_test_data()
models.TaskRelationDraft.objects.create(**test_data_1)
nbr_results = models.TaskRelationDraft.objects.count()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/task_relation_draft/', test_data_1, nbr_results)
def test_GET_TaskRelationDraft_view_returns_correct_entry(self):
# setup
test_data_1 = TaskRelationDraft_test_data()
test_data_2 = TaskRelationDraft_test_data()
id1 = models.TaskRelationDraft.objects.create(**test_data_1).id
id2 = models.TaskRelationDraft.objects.create(**test_data_2).id
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_relation_draft/%s/' % id1, test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_relation_draft/%s/' % id2, test_data_2)
def test_nested_TaskRelationDraft_are_filtered_according_to_TaskDraft(self):
# setup
test_data_1 = TaskRelationDraft_test_data()
tdt_test_data_1 = TaskDraft_test_data()
task_draft_1 = models.TaskDraft.objects.create(**tdt_test_data_1)
test_data_1 = dict(test_data_1)
test_data_1['producer'] = task_draft_1
task_relation_draft_1 = models.TaskRelationDraft.objects.create(**test_data_1)
# assert the returned list contains related items, A list of length 1 is retrieved
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/task_draft/%s/task_relation_draft/' % task_draft_1.id, test_data_1, 1)
# assert an existing related producer is returned
class SchedulingUnitBlueprintTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.scheduling_unit_draft_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitDraft(), '/scheduling_unit_draft/')
cls.template_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitTemplate(), '/scheduling_unit_template/')
def test_scheduling_unit_blueprint_list_apiformat(self):
r = requests.get(BASE_URL + '/scheduling_unit_blueprint/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Scheduling Unit Blueprint List" in r.content.decode('utf8'))
def test_scheduling_unit_blueprint_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/1234321/', 404)
def test_scheduling_unit_blueprint_POST_and_GET(self):
sub_test_data = test_data_creator.SchedulingUnitBlueprint(scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', sub_test_data, 201, sub_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, sub_test_data)
def test_scheduling_unit_blueprint_PUT_invalid_raises_error(self):
sub_test_data = test_data_creator.SchedulingUnitBlueprint(scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/9876789876/', sub_test_data, 404, {})
def test_scheduling_unit_blueprint_PUT(self):
sub_test_data1 = test_data_creator.SchedulingUnitBlueprint(name="the one "+str(uuid.uuid4()), scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
sub_test_data2 = test_data_creator.SchedulingUnitBlueprint(name="the other "+str(uuid.uuid4()), scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', sub_test_data1, 201, sub_test_data1)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, sub_test_data1)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, sub_test_data2, 200, sub_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, sub_test_data2)
def test_scheduling_unit_blueprint_PATCH(self):
sub_test_data = test_data_creator.SchedulingUnitBlueprint(scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', sub_test_data, 201, sub_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, sub_test_data)
test_patch = {"description": "This is an updated description"}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(sub_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_scheduling_unit_blueprint_DELETE(self):
sub_test_data = test_data_creator.SchedulingUnitBlueprint(scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', sub_test_data, 201, sub_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, sub_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_scheduling_unit_blueprint_CASCADE_behavior_on_scheduling_unit_template_deleted(self):
template_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitTemplate(), '/scheduling_unit_template/')
sub_test_data = test_data_creator.SchedulingUnitBlueprint(scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=template_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', sub_test_data, 201, sub_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, sub_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, template_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_scheduling_unit_blueprint_PROTECT_behavior_on_scheduling_unit_draft_deleted(self):
scheduling_unit_draft_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitDraft(), '/scheduling_unit_draft/')
sub_test_data = test_data_creator.SchedulingUnitBlueprint(scheduling_unit_draft_url=scheduling_unit_draft_url, template_url=self.template_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', sub_test_data, 201, sub_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, sub_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(scheduling_unit_draft_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
def test_GET_SchedulingUnitBlueprint_view_returns_correct_entry(self):
# setup
test_data_1 = SchedulingUnitBlueprint_test_data("scheduler unit blue print two one ")
test_data_2 = SchedulingUnitBlueprint_test_data("scheduler unit blue print two two ")
id1 = models.SchedulingUnitBlueprint.objects.create(**test_data_1).id
id2 = models.SchedulingUnitBlueprint.objects.create(**test_data_2).id
# assert
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/%s/' % id1, test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/%s/' % id2, test_data_2)
def test_GET_SchedulingUnitBlueprint_view_filters_for_time_range(self):
"""
Test we can filter on this property, which is explicitly named on the model-specific property filter
"""
# setup
subtask_1 = models.Subtask.objects.create(**Subtask_test_data(start_time=datetime(2050, 1, 1, 10, 0, 0), stop_time=datetime(2050, 1, 1, 14, 0, 0)))
subtask_2 = models.Subtask.objects.create(**Subtask_test_data(start_time=datetime(2050, 1, 5, 10, 0, 0), stop_time=datetime(2050, 1, 5, 14, 0, 0)))
task_blueprint_1 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data())
task_blueprint_2 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data())
subtask_1.task_blueprints.set([task_blueprint_1])
subtask_2.task_blueprints.set([task_blueprint_2])
subtask_1.save()
subtask_2.save()
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?start_time_after=2050-01-01T9:00:00&stop_time_before=2050-01-01T15:00:00', 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?start_time_after=2050-01-01T9:00:00&stop_time_before=2050-01-05T15:00:00', 200)
self.assertEqual(response_1['count'], 1)
self.assertEqual(response_2['count'], 2)
def test_GET_SchedulingUnitBlueprint_view_filters_for_project(self):
"""
Test we can filter on this property, which is explicitly named on the model-specific property filter
"""
# setup
project_1 = models.Project.objects.create(**Project_test_data(name='myproject_partial_1_%s' % uuid.uuid4()))
project_2 = models.Project.objects.create(**Project_test_data(name='myproject_partial_2_%s' % uuid.uuid4()))
scheduling_set_1 = models.SchedulingSet.objects.create(**SchedulingSet_test_data(project=project_1))
scheduling_set_2 = models.SchedulingSet.objects.create(**SchedulingSet_test_data(project=project_2))
su_draft_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(scheduling_set=scheduling_set_1))
su_draft_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(scheduling_set=scheduling_set_2))
su_blueprint_1 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(draft=su_draft_1, name='mysub1_%s' % uuid.uuid4()))
su_blueprint_2 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(draft=su_draft_2, name='mysub2_%s' % uuid.uuid4()))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?project=%s' % project_1.name, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?project=%s' % project_2.name, 200)
response_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?project=foo', 200)
response_4 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?project=myproject_partia', 200)
self.assertEqual(response_1['count'], 1)
self.assertEqual(response_1['results'][0]['name'], su_blueprint_1.name)
self.assertEqual(response_2['count'], 1)
self.assertEqual(response_2['results'][0]['name'], su_blueprint_2.name)
self.assertEqual(response_3['count'], 0)
self.assertEqual(response_4['count'], 2)
self.assertEqual(response_4['results'][0]['name'], su_blueprint_1.name)
self.assertEqual(response_4['results'][1]['name'], su_blueprint_2.name)
def test_GET_SchedulingUnitBlueprint_view_filters_for_output_pinned(self):
"""
Test we can filter on this regular field, because the model-specific property filter uses __all__
"""
# setup
models.SchedulingUnitBlueprint.objects.all().delete()
su_blueprint_true = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(name='mysub1_%s' % uuid.uuid4(), output_pinned=True))
su_blueprint_false = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(name='mysub2_%s' % uuid.uuid4(), output_pinned=False))
# assert
response = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/', 200)
response_true = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?output_pinned=true', 200)
response_false = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?output_pinned=false', 200)
self.assertEqual(response['count'], 2)
self.assertEqual(response_true['count'], 1)
self.assertEqual(response_true['results'][0]['name'], su_blueprint_true.name)
self.assertEqual(response_false['count'], 1)
self.assertEqual(response_false['results'][0]['name'], su_blueprint_false.name)
def test_GET_SchedulingUnitBlueprint_view_filters_for_draft(self):
"""
Test we can filter on this property, which is explicitly named on the model-specific property filter
"""
# setup
su_draft_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud1_%s' % uuid.uuid4()))
su_draft_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud2_%s' % uuid.uuid4()))
su_draft_3 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud3_%s' % uuid.uuid4()))
su_blueprint_1 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(name='sub1_%s' % uuid.uuid4()))
su_blueprint_2 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(name='sub2_%s' % uuid.uuid4()))
su_blueprint_3 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(name='sub3_%s' % uuid.uuid4()))
su_blueprint_1.draft = su_draft_1
su_blueprint_1.save()
su_blueprint_2.draft = su_draft_2
su_blueprint_2.save()
su_blueprint_3.draft = su_draft_3
su_blueprint_3.save()
# assertresponse_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/?scheduling_unit_blueprints=%s' % 999999, 200)
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?draft=%s' % su_draft_1.id, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?draft=%s,%s' % (su_draft_1.id, su_draft_3.id), 200)
response_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?draft=%s' % 999999, 400)
response_4 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?draft_min=%s&draft_max=%s' % (su_draft_2.id, su_draft_3.id), 200)
self.assertEqual(response_1['count'], 1)
self.assertEqual(response_1['results'][0]['name'], su_blueprint_1.name)
self.assertEqual(response_2['count'], 2)
self.assertEqual(response_2['results'][0]['name'], su_blueprint_1.name)
self.assertEqual(response_2['results'][1]['name'], su_blueprint_3.name)
self.assertIn('Select a valid choice', str(response_3))
self.assertEqual(response_4['count'], 2)
self.assertEqual(response_4['results'][0]['name'], su_blueprint_2.name)
self.assertEqual(response_4['results'][1]['name'], su_blueprint_3.name)
def test_GET_SchedulingUnitBlueprint_view_filters_for_id(self):
"""
Test we can filter on this field, which is explicitly named on the model-specific property filter
"""
# setup
su_blueprint_1 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(name='sub1_%s' % uuid.uuid4()))
su_blueprint_2 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(name='sub2_%s' % uuid.uuid4()))
su_blueprint_3 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(name='sub3_%s' % uuid.uuid4()))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?id=%s' % su_blueprint_1.id, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?id=%s,%s' % (su_blueprint_1.id, su_blueprint_3.id), 200)
response_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?id=%s' % 999999, 200)
response_4 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?id_min=%s&id_max=%s' % (su_blueprint_2.id, su_blueprint_3.id), 200)
self.assertEqual(response_1['count'], 1)
self.assertEqual(response_1['results'][0]['name'], su_blueprint_1.name)
self.assertEqual(response_2['count'], 2)
self.assertEqual(response_2['results'][0]['name'], su_blueprint_1.name)
self.assertEqual(response_2['results'][1]['name'], su_blueprint_3.name)
self.assertEqual(response_3['count'], 0)
self.assertEqual(response_4['count'], 2)
self.assertEqual(response_4['results'][0]['name'], su_blueprint_2.name)
self.assertEqual(response_4['results'][1]['name'], su_blueprint_3.name)
def test_GET_SchedulingUnitBlueprint_view_filters_for_strategy_template(self):
"""
Test we can filter on this field, which is explicitly named on the model-specific property filter
"""
# setup
template_1 = models.SchedulingUnitObservingStrategyTemplate.objects.create(**SchedulingUnitObservingStrategyTemplate_test_data(name='suost1_%s' % uuid.uuid4()))
template_2 = models.SchedulingUnitObservingStrategyTemplate.objects.create(**SchedulingUnitObservingStrategyTemplate_test_data(name='suost1_%s' % uuid.uuid4()))
su_draft_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud1_%s' % uuid.uuid4(), observation_strategy_template=template_1))
su_draft_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(name='sud2_%s' % uuid.uuid4(), observation_strategy_template=template_2))
su_blueprint_1 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(name='sub1_%s' % uuid.uuid4(), draft=su_draft_1))
su_blueprint_2 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(name='sub2_%s' % uuid.uuid4(), draft=su_draft_2))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?observation_strategy_template=%s' % template_1.id, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?observation_strategy_template_name=%s' % template_2.name[:-3], 200)
response_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?observation_strategy_template_name=%s' % 'gibberish', 200)
GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?observation_strategy_template=%s' % 999999, 400)
self.assertEqual(response_1['count'], 1)
self.assertEqual(response_1['results'][0]['name'], su_blueprint_1.name)
self.assertEqual(response_2['count'], 1)
self.assertEqual(response_2['results'][0]['name'], su_blueprint_2.name)
self.assertEqual(response_3['count'], 0)
class TaskBlueprintTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.draft_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskDraft(), '/task_draft/')
cls.template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
cls.scheduling_unit_blueprint_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitBlueprint(), '/scheduling_unit_blueprint/')
def test_task_blueprint_list_apiformat(self):
r = requests.get(BASE_URL + '/task_blueprint/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Blueprint List" in r.content.decode('utf8'))
def test_task_blueprint_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/1234321/', 404)
def test_task_blueprint_POST_and_GET(self):
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tb_test_data)
def test_task_blueprint_PUT_invalid_raises_error(self):
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
PUT_and_assert_expected_response(self, BASE_URL + '/task_blueprint/9876789876/', tb_test_data, 404, {})
def test_task_blueprint_PUT(self):
tb_test_data1 = test_data_creator.TaskBlueprint(name="the one "+str(uuid.uuid4()), draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
tb_test_data2 = test_data_creator.TaskBlueprint(name="the other "+str(uuid.uuid4()), draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data1, 201, tb_test_data1)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tb_test_data1)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, tb_test_data2, 200, tb_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, tb_test_data2)
def test_task_blueprint_PATCH(self):
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tb_test_data)
test_patch = {"description": "This is an updated description"}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(tb_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_task_blueprint_DELETE(self):
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tb_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_task_blueprint_prevents_missing_specification_template(self):
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# test data
test_data = dict(tb_test_data)
test_data['specifications_template'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['specifications_template']))
def test_task_blueprint_prevents_missing_draft(self):
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# test data
test_data = dict(tb_test_data)
test_data['draft'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['draft']))
def test_task_blueprint_prevents_missing_scheduling_unit_blueprint(self):
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# test data
test_data = dict(tb_test_data)
test_data['scheduling_unit_blueprint'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['scheduling_unit_blueprint']))
def test_task_blueprint_CASCADE_behavior_on_task_template_deleted(self):
template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, tb_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, template_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_blueprint_PROTECT_behavior_on_task_draft_deleted(self):
draft_data = test_data_creator.TaskDraft()
draft_url = test_data_creator.post_data_and_get_url(draft_data, '/task_draft/')
tb_test_data = test_data_creator.TaskBlueprint(draft_url=draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, tb_test_data)
# refresh draft_data, because it now has a reference to the blueprint
draft_data = test_data_creator.get_response_as_json_object(draft_url)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(draft_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, draft_url, draft_data)
def test_task_blueprint_CASCADE_behavior_on_scheduling_unit_blueprint_deleted(self):
scheduling_unit_blueprint_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitBlueprint(), '/scheduling_unit_blueprint/')
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=scheduling_unit_blueprint_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, tb_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, scheduling_unit_blueprint_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_GET_TaskBlueprint_list_view_shows_entry(self):
test_data_1 = TaskBlueprint_test_data()
tb = models.TaskBlueprint.objects.create(**test_data_1)
nbr_results = models.TaskBlueprint.objects.count()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/task_blueprint/', test_data_1, nbr_results, expected_id=tb.id)
def test_GET_TaskBlueprint_view_returns_correct_entry(self):
# setup
test_data_1 = TaskBlueprint_test_data("task blue print two one")
test_data_2 = TaskBlueprint_test_data("task blue print two two")
id1 = models.TaskBlueprint.objects.create(**test_data_1).id
id2 = models.TaskBlueprint.objects.create(**test_data_2).id
# assert
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_blueprint/%s/' % id1, test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_blueprint/%s/' % id2, test_data_2)
def test_nested_TaskBlueprint_are_filtered_according_to_TaskDraft(self):
# setup
test_data_1 = TaskBlueprint_test_data("task blue print three one")
tdt_test_data_1 = TaskDraft_test_data("task draft two one")
task_draft_1 = models.TaskDraft.objects.create(**tdt_test_data_1)
test_data_1 = dict(test_data_1)
test_data_1['draft'] = task_draft_1
task_blueprint_1 = models.TaskBlueprint.objects.create(**test_data_1)
# assert the returned list contains related items, A list of length 1 is retrieved
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/task_draft/%s/task_blueprint/' % task_draft_1.id, test_data_1, 1)
def test_TaskBlueprint_contains_list_of_related_Subtask(self):
# setup
test_data_1 = TaskBlueprint_test_data()
test_data_2 = TaskBlueprint_test_data()
st_test_data_1 = Subtask_test_data()
st_test_data_2 = Subtask_test_data()
task_blueprint = models.TaskBlueprint.objects.create(**test_data_1)
subtask_1 = models.Subtask.objects.create(**st_test_data_1)
subtask_1.task_blueprints.set([task_blueprint])
subtask_1.save()
subtask_2 = models.Subtask.objects.create(**st_test_data_2)
subtask_2.task_blueprints.set([task_blueprint])
subtask_2.save()
# assert
response_data = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/%s/' % task_blueprint.id, 200)
assertUrlList(self, response_data['subtasks'], [subtask_1, subtask_2])
def test_TaskBlueprint_contains_lists_of_related_TaskRelationBlueprint(self):
# setup
task_blueprint = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data())
task_relation_blueprint_1 = models.TaskRelationBlueprint.objects.create(**TaskRelationBlueprint_test_data(producer=task_blueprint))
task_relation_blueprint_2 = models.TaskRelationBlueprint.objects.create(**TaskRelationBlueprint_test_data(consumer=task_blueprint))
# assert
response_data = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/%s/' % task_blueprint.id, 200)
assertUrlList(self, response_data['consumed_by'], [task_relation_blueprint_1])
assertUrlList(self, response_data['produced_by'], [task_relation_blueprint_2])
def test_TaskBlueprint_filters_for_SchedulingUnitBlueprint(self):
# setup
su_blueprint_1 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data("sub_%s" % uuid.uuid4()))
su_blueprint_2 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data("sub_%s" % uuid.uuid4()))
su_blueprint_3 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data("sub_%s" % uuid.uuid4()))
task_blueprint_1 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data("tb_%s" % uuid.uuid4(), scheduling_unit_blueprint=su_blueprint_1))
task_blueprint_2 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data("tb_%s" % uuid.uuid4(), scheduling_unit_blueprint=su_blueprint_2))
task_blueprint_3 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data("tb_%s" % uuid.uuid4(), scheduling_unit_blueprint=su_blueprint_3))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?scheduling_unit_blueprint=%s' % su_blueprint_1.id, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?scheduling_unit_blueprint=%s,%s' % (su_blueprint_2.id, su_blueprint_3.id), 200)
response_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?scheduling_unit_blueprint_min=%s&scheduling_unit_blueprint_max=%s' % (su_blueprint_3.id, su_blueprint_3.id), 200)
response_4 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?scheduling_unit_blueprint_name=%s' % su_blueprint_2.name[:-3], 200)
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?scheduling_unit_blueprint=%s' % 99999, 400)
self.assertEqual(response_1['count'], 1)
self.assertIn(task_blueprint_1.name, str(response_1))
self.assertEqual(response_2['count'], 2)
self.assertNotIn(task_blueprint_1.name, str(response_2))
self.assertIn(task_blueprint_2.name, str(response_2))
self.assertIn(task_blueprint_3.name, str(response_2))
self.assertEqual(response_3['count'], 1)
self.assertNotIn(task_blueprint_1.name, str(response_3))
self.assertNotIn(task_blueprint_2.name, str(response_3))
self.assertIn(task_blueprint_3.name, str(response_3))
self.assertEqual(response_4['count'], 1)
self.assertNotIn(task_blueprint_1.name, str(response_4))
self.assertIn(task_blueprint_2.name, str(response_4))
self.assertNotIn(task_blueprint_3.name, str(response_4))
def test_TaskBlueprint_filters_for_TaskDraft(self):
# setup
task_draft_1 = models.TaskDraft.objects.create(**TaskDraft_test_data("td_%s" % uuid.uuid4()))
task_draft_2 = models.TaskDraft.objects.create(**TaskDraft_test_data("td_%s" % uuid.uuid4()))
task_draft_3 = models.TaskDraft.objects.create(**TaskDraft_test_data("td_%s" % uuid.uuid4()))
task_blueprint_1 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(name="tb_%s" % uuid.uuid4(), task_draft=task_draft_1))
task_blueprint_2 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(name="tb_%s" % uuid.uuid4(), task_draft=task_draft_2))
task_blueprint_3 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(name="tb_%s" % uuid.uuid4(), task_draft=task_draft_3))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?draft=%s' % task_draft_1.id, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?draft_min=%s&draft_max=%s' % (task_draft_2.id, task_draft_2.id), 200)
response_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?draft=%s,%s' % (task_draft_1.id, task_draft_3.id), 200)
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?draft=%s' % 9999999, 400)
self.assertIn(task_blueprint_1.name, str(response_1))
self.assertNotIn(task_blueprint_2.name, str(response_1))
self.assertNotIn(task_blueprint_3.name, str(response_1))
self.assertNotIn(task_blueprint_1.name, str(response_2))
self.assertIn(task_blueprint_2.name, str(response_2))
self.assertNotIn(task_blueprint_3.name, str(response_2))
self.assertIn(task_blueprint_1.name, str(response_3))
self.assertNotIn(task_blueprint_2.name, str(response_3))
self.assertIn(task_blueprint_3.name, str(response_3))
def test_TaskBlueprint_filters_for_Subtask(self):
# setup
task_blueprint_1 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(name="tb_%s" % uuid.uuid4()))
task_blueprint_2 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(name="tb_%s" % uuid.uuid4()))
task_blueprint_3 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(name="tb_%s" % uuid.uuid4()))
subtask_1 = models.Subtask.objects.create(**Subtask_test_data())
subtask_2 = models.Subtask.objects.create(**Subtask_test_data())
subtask_3 = models.Subtask.objects.create(**Subtask_test_data())
subtask_1.task_blueprints.set([task_blueprint_1])
subtask_2.task_blueprints.set([task_blueprint_2])
subtask_3.task_blueprints.set([task_blueprint_3])
subtask_1.save()
subtask_2.save()
subtask_3.save()
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?subtasks=%s' % subtask_1.id, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?subtasks_min=%s&subtasks_max=%s' % (subtask_2.id, subtask_2.id), 200)
response_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?subtasks=%s&subtasks=%s' % (subtask_1.id, subtask_3.id), 200)
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?subtasks=%s' % 9999999, 400)
self.assertIn(task_blueprint_1.name, str(response_1))
self.assertNotIn(task_blueprint_2.name, str(response_1))
self.assertNotIn(task_blueprint_3.name, str(response_1))
self.assertNotIn(task_blueprint_1.name, str(response_2))
self.assertIn(task_blueprint_2.name, str(response_2))
self.assertNotIn(task_blueprint_3.name, str(response_2))
self.assertIn(task_blueprint_1.name, str(response_3))
self.assertNotIn(task_blueprint_2.name, str(response_3))
self.assertIn(task_blueprint_3.name, str(response_3))
def test_TaskBlueprint_filters_for_status(self):
# setup
task_blueprint_1 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(name="tb_%s" % uuid.uuid4()))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?status=defined&status=finished', 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?status=obsolete', 200)
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?status=gibberish', 400)
self.assertGreater(response_1['count'], 0)
self.assertEqual(response_2['count'], 0)
def test_TaskBlueprint_filters_for_project(self):
"""
Test we can filter on this property, which is explicitly named on the model-specific property filter
"""
# setup
project_1 = models.Project.objects.create(**Project_test_data(name='myproject_partial_1_%s' % uuid.uuid4()))
project_2 = models.Project.objects.create(**Project_test_data(name='myproject_partial_2_%s' % uuid.uuid4()))
scheduling_set_1 = models.SchedulingSet.objects.create(**SchedulingSet_test_data(project=project_1))
scheduling_set_2 = models.SchedulingSet.objects.create(**SchedulingSet_test_data(project=project_2))
su_draft_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(scheduling_set=scheduling_set_1))
su_draft_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data(scheduling_set=scheduling_set_2))
su_blueprint_1 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(draft=su_draft_1, name='mysub1_%s' % uuid.uuid4()))
su_blueprint_2 = models.SchedulingUnitBlueprint.objects.create(**SchedulingUnitBlueprint_test_data(draft=su_draft_2, name='mysub2_%s' % uuid.uuid4()))
task_draft_1 = models.TaskDraft.objects.create(**TaskDraft_test_data("td_%s" % uuid.uuid4(), scheduling_unit_draft=su_draft_1))
task_draft_2 = models.TaskDraft.objects.create(**TaskDraft_test_data("td_%s" % uuid.uuid4(), scheduling_unit_draft=su_draft_2))
task_blueprint_1 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(name="tb_%s" % uuid.uuid4(), task_draft=task_draft_1, scheduling_unit_blueprint=su_blueprint_1))
task_blueprint_2 = models.TaskBlueprint.objects.create(**TaskBlueprint_test_data(name="tb_%s" % uuid.uuid4(), task_draft=task_draft_2, scheduling_unit_blueprint=su_blueprint_2))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?project=%s' % project_1.name, 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?project=%s' % project_2.name, 200)
response_3 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?project=foo', 200)
response_4 = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/?project=myproject_partia', 200)
self.assertEqual(response_1['count'], 1)
self.assertEqual(response_1['results'][0]['name'], task_blueprint_1.name)
self.assertEqual(response_2['count'], 1)
self.assertEqual(response_2['results'][0]['name'], task_blueprint_2.name)
self.assertEqual(response_3['count'], 0)
self.assertEqual(response_4['count'], 2)
self.assertEqual(response_4['results'][0]['name'], task_blueprint_1.name)
self.assertEqual(response_4['results'][1]['name'], task_blueprint_2.name)
class TaskRelationBlueprintTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.draft_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskRelationDraft(), '/task_relation_draft/')
cls.producer_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskBlueprint(), '/task_blueprint/')
cls.consumer_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskBlueprint(), '/task_blueprint/')
cls.template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskRelationSelectionTemplate(), '/task_relation_selection_template/')
cls.task_template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
cls.input_role_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(iotype="input", task_template_url=cls.task_template_url), '/task_connector_type/')
cls.output_role_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(iotype="output", task_template_url=cls.task_template_url), '/task_connector_type/')
def test_task_relation_blueprint_list_apiformat(self):
r = requests.get(BASE_URL + '/task_relation_blueprint/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Relation Blueprint List" in r.content.decode('utf8'))
def test_task_relation_blueprint_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_relation_blueprint/1234321/', 404)
def test_task_relation_blueprint_POST_and_GET(self):
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, producer_url=self.producer_url, consumer_url=self.consumer_url, template_url=self.template_url, input_role_url=self.input_role_url, output_role_url=self.output_role_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', trb_test_data, 201, trb_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, trb_test_data)
def test_task_relation_blueprint_PUT_invalid_raises_error(self):
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, consumer_url=self.consumer_url, producer_url=self.producer_url)
PUT_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/9876789876/', trb_test_data, 404, {})
def test_task_relation_blueprint_PUT(self):
task_template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
input_role_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(iotype="input", task_template_url=task_template_url), '/task_connector_type/')
trd_test_data1 = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, producer_url=self.producer_url, consumer_url=self.consumer_url, template_url=self.template_url, input_role_url=input_role_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', trd_test_data1, 201, trd_test_data1)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, trd_test_data1)
# PUT new values, verify
task_template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
input_role_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(iotype="input", task_template_url=task_template_url), '/task_connector_type/')
new_consumer_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskBlueprint(), '/task_blueprint/')
trd_test_data2 = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, producer_url=self.producer_url, consumer_url=new_consumer_url, template_url=self.template_url, input_role_url=input_role_url)
PUT_and_assert_expected_response(self, url, trd_test_data2, 200, trd_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, trd_test_data2)
def test_task_relation_blueprint_PATCH(self):
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, producer_url=self.producer_url, consumer_url=self.consumer_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', trb_test_data, 201, trb_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, trb_test_data)
test_patch = {"selection_doc": {"foo": "patched"}}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(trb_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_task_relation_blueprint_DELETE(self):
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, producer_url=self.producer_url, consumer_url=self.consumer_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', trb_test_data, 201, trb_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, trb_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_task_relation_blueprint_prevents_missing_selection_template(self):
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, consumer_url=self.consumer_url, producer_url=self.producer_url)
# test data
test_data = dict(trb_test_data)
test_data['selection_template'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['selection_template']))
def test_task_relation_blueprint_prevents_missing_draft(self):
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, consumer_url=self.consumer_url, producer_url=self.producer_url)
# test data
test_data = dict(trb_test_data)
test_data['draft'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['draft']))
def test_task_relation_blueprint_prevents_missing_producer(self):
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, consumer_url=self.consumer_url, producer_url=self.producer_url)
# test data
test_data = dict(trb_test_data)
test_data['producer'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['producer']))
def test_task_relation_blueprint_prevents_missing_consumer(self):
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, consumer_url=self.consumer_url, producer_url=self.producer_url)
# test data
test_data = dict(trb_test_data)
test_data['consumer'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['consumer']))
def test_task_relation_blueprint_prevents_missing_input(self):
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, consumer_url=self.consumer_url, producer_url=self.producer_url)
# test data
test_data = dict(trb_test_data)
test_data['input_role'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['input_role']))
def test_task_relation_blueprint_prevents_missing_output(self):
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, consumer_url=self.consumer_url, producer_url=self.producer_url)
# test data
test_data = dict(trb_test_data)
test_data['output_role'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['output_role']))
def test_task_relation_blueprint_CASCADE_behavior_on_task_relation_selection_template_deleted(self):
template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskRelationSelectionTemplate(), '/task_relation_selection_template/')
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=template_url, producer_url=self.producer_url, consumer_url=self.consumer_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', trb_test_data, 201, trb_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, trb_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, template_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_relation_blueprint_CASCADE_behavior_on_producer_deleted(self):
producer_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskBlueprint(), '/task_blueprint/')
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, consumer_url=self.consumer_url, producer_url=producer_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/',
trb_test_data, 201, trb_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, trb_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, producer_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_relation_blueprint_CASCADE_behavior_on_consumer_deleted(self):
consumer_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskBlueprint(), '/task_blueprint/')
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, consumer_url=consumer_url, producer_url=self.producer_url)
# POST new item with dependency
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/',
trb_test_data, 201, trb_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, trb_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, consumer_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_relation_blueprint_CASCADE_behavior_on_input_deleted(self):
task_template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
input_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(task_template_url=task_template_url), '/task_connector_type/')
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, input_role_url=input_url, producer_url=self.producer_url, consumer_url=self.consumer_url, template_url=self.template_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/',
trb_test_data, 201, trb_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, trb_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, input_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_relation_blueprint_CASCADE_behavior_on_output_deleted(self):
task_template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
output_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(task_template_url=task_template_url), '/task_connector_type/')
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, output_role_url=output_url, producer_url=self.producer_url, consumer_url=self.consumer_url, template_url=self.template_url, input_role_url=self.input_role_url)
# POST new item with dependency
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/',
trb_test_data, 201, trb_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, trb_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, output_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_GET_TaskRelationBlueprint_list_view_shows_entry(self):
test_data_1 = TaskRelationBlueprint_test_data()
models.TaskRelationBlueprint.objects.create(**test_data_1)
nbr_results = models.TaskRelationBlueprint.objects.count()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/task_relation_blueprint/', test_data_1, nbr_results)
def test_GET_TaskRelationBlueprint_view_returns_correct_entry(self):
# setup
test_data_1 = TaskRelationBlueprint_test_data()
test_data_2 = TaskRelationBlueprint_test_data()
id1 = models.TaskRelationBlueprint.objects.create(**test_data_1).id
id2 = models.TaskRelationBlueprint.objects.create(**test_data_2).id
# assert
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_relation_blueprint/%s/' % id1, test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_relation_blueprint/%s/' % id2, test_data_2)
def test_nested_TaskRelationBlueprint_are_filtered_according_to_TaskRelationDraft(self):
# setup
test_data_1 = TaskRelationBlueprint_test_data()
trdt_test_data_1 = TaskRelationDraft_test_data()
task_relation_draft_1 = models.TaskRelationDraft.objects.create(**trdt_test_data_1)
test_data_1 = dict(test_data_1)
test_data_1['draft'] = task_relation_draft_1
task_relation_blueprint_1 = models.TaskRelationBlueprint.objects.create(**test_data_1)
# assert the returned list contains related items, A list of length 1 is retrieved
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/task_relation_draft/%s/task_relation_blueprint/' % task_relation_draft_1.id, test_data_1, 1)
def test_nested_TaskRelationBlueprint_are_filtered_according_to_TaskBlueprint(self):
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, consumer_url=self.consumer_url, producer_url=self.producer_url)
POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', trb_test_data, 201, trb_test_data)['url']
# assert the returned list contains related producer
GET_and_assert_in_expected_response_result_list(self, '%s/task_relation_blueprint/' % self.producer_url, trb_test_data, 1)
# assert the returned list contains related consumer
GET_and_assert_in_expected_response_result_list(self, '%s/task_relation_blueprint/' % self.consumer_url, trb_test_data, 1)
class TaskSchedulingRelationBlueprintTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.task_blueprint = test_data_creator.post_data_and_get_response_as_json_object(test_data_creator.TaskBlueprint(), '/task_blueprint/')
@classmethod
def fast_create_new_task_blueprint_url(cls):
'''create a new task_blueprint object, based on existing one, with only new name, returning new url'''
tbp = dict(cls.task_blueprint)
tbp['name'] = str(uuid.uuid4())
return test_data_creator.post_data_and_get_url(tbp, '/task_blueprint/')
def test_task_scheduling_relation_blueprint_list_apiformat(self):
r = requests.get(BASE_URL + '/task_scheduling_relation_blueprint/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Scheduling Relation Blueprint List" in r.content.decode('utf8'))
def test_task_scheduling_relation_blueprint_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_scheduling_relation_blueprint/1234321/', 404)
def test_task_scheduling_relation_blueprint_POST_and_GET(self):
tsrb_test_data = test_data_creator.TaskSchedulingRelationBlueprint(first_url=self.fast_create_new_task_blueprint_url(), second_url=self.fast_create_new_task_blueprint_url(), placement="after")
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_blueprint/', tsrb_test_data, 201, tsrb_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tsrb_test_data)
def test_task_scheduling_relation_blueprint_unique_constraint(self):
tsrb_test_data = test_data_creator.TaskSchedulingRelationBlueprint(first_url=self.fast_create_new_task_blueprint_url(), second_url=self.fast_create_new_task_blueprint_url(), placement="after")
POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_blueprint/', tsrb_test_data, 201, tsrb_test_data)
# again should raise a unique constraint error, resulting in http 500
POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_blueprint/', tsrb_test_data, 500, tsrb_test_data)
def test_task_scheduling_relation_blueprint_PUT_invalid_raises_error(self):
tsrb_test_data = test_data_creator.TaskSchedulingRelationBlueprint(first_url=self.fast_create_new_task_blueprint_url(), second_url=self.fast_create_new_task_blueprint_url(), placement="after")
PUT_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_blueprint/9876789876/', tsrb_test_data, 404, {})
def test_task_scheduling_relation_blueprint_PUT(self):
tsrb_test_data1 = test_data_creator.TaskSchedulingRelationBlueprint(first_url=self.fast_create_new_task_blueprint_url(), second_url=self.fast_create_new_task_blueprint_url(), placement="after")
tsrb_test_data2 = test_data_creator.TaskSchedulingRelationBlueprint(first_url=self.fast_create_new_task_blueprint_url(), second_url=self.fast_create_new_task_blueprint_url(), placement="after")
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_blueprint/', tsrb_test_data1, 201, tsrb_test_data1)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tsrb_test_data1)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, tsrb_test_data2, 200, tsrb_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, tsrb_test_data2)
def test_task_scheduling_relation_blueprint_PATCH(self):
tsrb_test_data = test_data_creator.TaskSchedulingRelationBlueprint(first_url=self.fast_create_new_task_blueprint_url(), second_url=self.fast_create_new_task_blueprint_url(), placement="after")
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_blueprint/', tsrb_test_data, 201, tsrb_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tsrb_test_data)
test_patch = {"time_offset": 20}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(tsrb_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_task_scheduling_relation_blueprint_DELETE(self):
tsrb_test_data = test_data_creator.TaskSchedulingRelationBlueprint(first_url=self.fast_create_new_task_blueprint_url(), second_url=self.fast_create_new_task_blueprint_url(), placement="after")
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_blueprint/', tsrb_test_data, 201, tsrb_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tsrb_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_task_scheduling_relation_blueprint_prevents_missing_time_offset(self):
tsrb_test_data = test_data_creator.TaskSchedulingRelationBlueprint(first_url=self.fast_create_new_task_blueprint_url(), second_url=self.fast_create_new_task_blueprint_url(), placement="after")
# test data
test_data = dict(tsrb_test_data)
test_data['time_offset'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['time_offset']))
def test_task_scheduling_relation_blueprint_prevents_missing_time_first(self):
tsrb_test_data = test_data_creator.TaskSchedulingRelationBlueprint(first_url=self.fast_create_new_task_blueprint_url(), second_url=self.fast_create_new_task_blueprint_url(), placement="after")
# test data
test_data = dict(tsrb_test_data)
test_data['first'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['first']))
def test_task_scheduling_relation_blueprint_prevents_missing_second(self):
tsrb_test_data = test_data_creator.TaskSchedulingRelationBlueprint(first_url=self.fast_create_new_task_blueprint_url(), second_url=self.fast_create_new_task_blueprint_url(), placement="after")
# test data
test_data = dict(tsrb_test_data)
test_data['second'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['second']))
def test_task_scheduling_relation_blueprint_prevents_missing_placement(self):
tsrb_test_data = test_data_creator.TaskSchedulingRelationBlueprint(first_url=self.fast_create_new_task_blueprint_url(), second_url=self.fast_create_new_task_blueprint_url(), placement="after")
# test data
test_data = dict(tsrb_test_data)
test_data['placement'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['placement']))
def test_task_scheduling_relation_blueprint_CASCADE_behavior_on_task_blueprint_deleted(self):
#Create test data
first_task_blueprint_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskBlueprint(), '/task_blueprint/')
tsrb_test_data = test_data_creator.TaskSchedulingRelationBlueprint(first_url=first_task_blueprint_url, second_url=self.fast_create_new_task_blueprint_url(), placement="after")
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_blueprint/', tsrb_test_data, 201, tsrb_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, tsrb_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, first_task_blueprint_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_GET_TaskSchedulingRelationBlueprint_list_view_shows_entry(self):
test_data_1 = TaskSchedulingRelationBlueprint_test_data()
models.TaskSchedulingRelationBlueprint.objects.create(**test_data_1)
nbr_results = models.TaskSchedulingRelationBlueprint.objects.count()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/task_scheduling_relation_blueprint/', test_data_1, nbr_results)
def test_GET_TaskSchedulingRelationBlueprint_view_returns_correct_entry(self):
# setup
test_data_1 = TaskSchedulingRelationBlueprint_test_data()
test_data_2 = TaskSchedulingRelationBlueprint_test_data()
id1 = models.TaskSchedulingRelationBlueprint.objects.create(**test_data_1).id
id2 = models.TaskSchedulingRelationBlueprint.objects.create(**test_data_2).id
# assert
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_scheduling_relation_blueprint/%s/' % id1, test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_scheduling_relation_blueprint/%s/' % id2, test_data_2)
class TaskSchedulingRelationDraftTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.task_draft = test_data_creator.post_data_and_get_response_as_json_object(test_data_creator.TaskDraft(), '/task_draft/')
@classmethod
def fast_create_new_task_draft_url(cls):
'''create a new task_draft object, based on existing one, with only new name, returning new url'''
tbp = dict(cls.task_draft)
tbp['name'] = str(uuid.uuid4())
return test_data_creator.post_data_and_get_url(tbp, '/task_draft/')
def test_task_scheduling_relation_draft_list_apiformat(self):
r = requests.get(BASE_URL + '/task_scheduling_relation_draft/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Scheduling Relation Draft List" in r.content.decode('utf8'))
def test_task_scheduling_relation_draft_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_scheduling_relation_draft/1234321/', 404)
def test_task_scheduling_relation_draft_POST_and_GET(self):
tsrd_test_data = test_data_creator.TaskSchedulingRelationDraft( first_url=self.fast_create_new_task_draft_url(), second_url=self.fast_create_new_task_draft_url(), placement="after")
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_draft/', tsrd_test_data, 201, tsrd_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tsrd_test_data)
def test_task_scheduling_relation_draft_PUT_invalid_raises_error(self):
tsrd_test_data = test_data_creator.TaskSchedulingRelationDraft( first_url=self.fast_create_new_task_draft_url(), second_url=self.fast_create_new_task_draft_url(), placement="after")
PUT_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_draft/9876789876/', tsrd_test_data, 404, {})
def test_task_scheduling_relation_draft_PUT(self):
tsrd_test_data = test_data_creator.TaskSchedulingRelationDraft( first_url=self.fast_create_new_task_draft_url(), second_url=self.fast_create_new_task_draft_url(), placement="after")
tsrd_test_data2 = test_data_creator.TaskSchedulingRelationDraft( first_url=self.fast_create_new_task_draft_url(), second_url=self.fast_create_new_task_draft_url(), placement="after")
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_draft/', tsrd_test_data, 201, tsrd_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tsrd_test_data)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, tsrd_test_data2, 200, tsrd_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, tsrd_test_data2)
def test_task_scheduling_relation_draft_PATCH(self):
tsrd_test_data = test_data_creator.TaskSchedulingRelationDraft(first_url=self.fast_create_new_task_draft_url(), second_url=self.fast_create_new_task_draft_url(), placement="after")
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_draft/', tsrd_test_data, 201, tsrd_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tsrd_test_data)
test_patch = {"time_offset": 20}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(tsrd_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_task_scheduling_relation_draft_DELETE(self):
tsrd_test_data = test_data_creator.TaskSchedulingRelationDraft( first_url=self.fast_create_new_task_draft_url(), second_url=self.fast_create_new_task_draft_url(), placement="after")
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_draft/', tsrd_test_data, 201, tsrd_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tsrd_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_task_scheduling_relation_draft_prevents_missing_time_offset(self):
tsrd_test_data = test_data_creator.TaskSchedulingRelationDraft( first_url=self.fast_create_new_task_draft_url(), second_url=self.fast_create_new_task_draft_url(), placement="after")
# test data
test_data = dict(tsrd_test_data)
test_data['time_offset'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_draft/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['time_offset']))
def test_task_scheduling_relation_draft_prevents_missing_time_first(self):
tsrd_test_data = test_data_creator.TaskSchedulingRelationDraft( first_url=self.fast_create_new_task_draft_url(), second_url=self.fast_create_new_task_draft_url(), placement="after")
# test data
test_data = dict(tsrd_test_data)
test_data['first'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_draft/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['first']))
def test_task_scheduling_relation_draft_prevents_missing_second(self):
tsrd_test_data = test_data_creator.TaskSchedulingRelationDraft( first_url=self.fast_create_new_task_draft_url(), second_url=self.fast_create_new_task_draft_url(), placement="after")
# test data
test_data = dict(tsrd_test_data)
test_data['second'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_draft/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['second']))
def test_task_scheduling_relation_draft_prevents_missing_placement(self):
tsrd_test_data = test_data_creator.TaskSchedulingRelationDraft( first_url=self.fast_create_new_task_draft_url(), second_url=self.fast_create_new_task_draft_url(), placement="after")
# test data
test_data = dict(tsrd_test_data)
test_data['placement'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_draft/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['placement']))
def test_task_scheduling_relation_draft_CASCADE_behavior_on_task_draft_deleted(self):
task_draft_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskDraft(), '/task_draft/')
tsrd_test_data = test_data_creator.TaskSchedulingRelationDraft( first_url=task_draft_url, second_url=None, placement="after")
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_scheduling_relation_draft/', tsrd_test_data, 201, tsrd_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, tsrd_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, task_draft_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_GET_TaskSchedulingRelationDraft_list_view_shows_entry(self):
test_data_1 = TaskSchedulingRelationDraft_test_data()
models.TaskSchedulingRelationDraft.objects.create(**test_data_1)
nbr_results = models.TaskSchedulingRelationDraft.objects.count()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/task_scheduling_relation_draft/', test_data_1, nbr_results)
def test_GET_TaskSchedulingRelationDraft_view_returns_correct_entry(self):
# setup
test_data_1 = TaskSchedulingRelationDraft_test_data()
test_data_2 = TaskSchedulingRelationDraft_test_data()
id1 = models.TaskSchedulingRelationDraft.objects.create(**test_data_1).id
id2 = models.TaskSchedulingRelationDraft.objects.create(**test_data_2).id
# assert
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_scheduling_relation_draft/%s/' % id1, test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_scheduling_relation_draft/%s/' % id2, test_data_2)
class ReservationTestCase(unittest.TestCase):
def test_reservation_list_apiformat(self):
r = requests.get(BASE_URL + '/reservation/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Reservation List" in r.content.decode('utf8'))
def test_reservation_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/reservation/1234321/', 404)
def test_reservation_POST_and_GET(self):
reservation_test_data = test_data_creator.Reservation(duration=60)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/reservation/', reservation_test_data, 201, reservation_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, reservation_test_data)
def test_reservation_PUT_invalid_raises_error(self):
reservation_test_data = test_data_creator.Reservation(duration=60)
PUT_and_assert_expected_response(self, BASE_URL + '/reservation/9876789876/', reservation_test_data, 404, {})
def test_reservation_PUT(self):
project_url = test_data_creator.post_data_and_get_url(test_data_creator.Project(), '/project/')
reservation_test_data = test_data_creator.Reservation(name="reservation 1", duration=50, project_url=project_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/reservation/', reservation_test_data, 201, reservation_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, reservation_test_data)
reservation_test_data2 = test_data_creator.Reservation(name="reservation2", project_url=project_url)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, reservation_test_data2, 200, reservation_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, reservation_test_data2)
def test_reservation_PATCH(self):
reservation_test_data = test_data_creator.Reservation(duration=60)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/reservation/', reservation_test_data, 201, reservation_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, reservation_test_data)
test_patch = {"description": "This is a new and improved description",
"stop_time": None}
# PATCH item and verify
expected_patch_data = test_data_creator.update_schema_from_template("reservationtemplate", test_patch)
PATCH_and_assert_expected_response(self, url, test_patch, 200, expected_patch_data)
expected_data = dict(reservation_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_patch_data)
def test_reservation_DELETE(self):
reservation_test_data = test_data_creator.Reservation(duration=30, start_time=datetime.utcnow() + timedelta(days=1))
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/reservation/', reservation_test_data, 201, reservation_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, reservation_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_GET_Reservation_list_shows_entry(self):
test_data_1 = Reservation_test_data(duration=3600)
models.Reservation.objects.create(**test_data_1)
nbr_results = models.Reservation.objects.count()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/reservation/', test_data_1, nbr_results)
def test_GET_Reservation_view_returns_correct_entry(self):
test_data_1 = Reservation_test_data(name="Reservation 1", duration=60, start_time=datetime.utcnow() + timedelta(days=1))
test_data_2 = Reservation_test_data(name="Reservation 2", duration=120, start_time=datetime.utcnow() + timedelta(days=2))
id1 = models.Reservation.objects.create(**test_data_1).id
id2 = models.Reservation.objects.create(**test_data_2).id
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/reservation/' + str(id1) + '/', test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/reservation/' + str(id2) + '/', test_data_2)
class ExtendedViewTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
# create some connected objects
cls.sud_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitDraft(), '/scheduling_unit_draft/')
cls.sub_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitBlueprint(scheduling_unit_draft_url=cls.sud_url), '/scheduling_unit_blueprint/')
cls.td_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskDraft(scheduling_unit_draft_url=cls.sud_url), '/task_draft/')
cls.tb_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskBlueprint(draft_url=cls.td_url, scheduling_unit_blueprint_url=cls.sub_url), '/task_blueprint/')
test_data_creator.post_data_and_get_url(test_data_creator.Subtask(task_blueprint_urls=[cls.tb_url]), '/subtask/')
def test_GET_scheduling_unit_draft_serializes_referenced_objects(self):
# get the extended view on the su draft
sud_url = self.sud_url.replace('scheduling_unit_draft', 'scheduling_unit_draft_extended')
r = GET_and_assert_equal_expected_code(self, sud_url, 200)
# assert that task drafts are expanded
self.assertIn('specifications_doc', r['task_drafts'][0])
# assert that task blueprint inside task drafts are expanded
self.assertIn('specifications_doc', r['task_drafts'][0]['task_blueprints'][0])
# assert that subtasks inside nested task blueprints are expanded
self.assertIn('specifications_doc', r['task_drafts'][0]['task_blueprints'][0]['subtasks'][0])
# assert that task templates inside nested task blueprints are expanded
self.assertIn('schema', r['task_drafts'][0]['task_blueprints'][0]['specifications_template'])
def test_GET_scheduling_unit_blueprint_serializes_referenced_objects(self):
# get the extended view on the su blueprint
sub_url = self.sub_url.replace('scheduling_unit_blueprint', 'scheduling_unit_blueprint_extended')
r = GET_and_assert_equal_expected_code(self, sub_url, 200)
# assert that task blueprints are expanded
self.assertIn('specifications_doc', r['task_blueprints'][0])
# assert that subtasks inside task blueprints are expanded
self.assertIn('specifications_doc', r['task_blueprints'][0]['subtasks'][0])
# assert that task templates inside task blueprints are expanded
self.assertIn('schema', r['task_blueprints'][0]['specifications_template'])
# todo: move to t_permissions (I tried, but it broke)
class CyclePermissionTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.test_data_creator = TMSSRESTTestDataCreator(BASE_URL, requests.auth.HTTPBasicAuth('paulus', 'pauluspass'))
response = requests.get(cls.test_data_creator.django_api_url + '/', auth=cls.test_data_creator.auth)
cls.support_group = Group.objects.create(name='test_support')
cls.support_group.permissions.add(Permission.objects.get(codename='add_cycle'))
cls.admin_group = Group.objects.create(name='test_admin')
cls.admin_group.permissions.add(Permission.objects.get(codename='delete_cycle'))
def test_Cycle_cannot_be_added_without_group(self):
user = User.objects.get(username='paulus')
user.groups.set([])
# refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching
user = User.objects.get(username='paulus')
while user.has_perm('tmssapp.add_cycle'):
user = User.objects.get(username='paulus')
self.assertFalse(user.has_perm('tmssapp.add_cycle'))
test_data = self.test_data_creator.Cycle()
res = self.test_data_creator.post_data_and_get_response(test_data, '/cycle/')
self.assertEqual(res.status_code, 403)
def test_Cycle_can_be_added_by_support(self):
user = User.objects.get(username='paulus')
user.groups.set([self.support_group])
# refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching
user = User.objects.get(username='paulus')
while not user.has_perm('tmssapp.add_cycle'):
user = User.objects.get(username='paulus')
self.assertTrue(user.has_perm('tmssapp.add_cycle'))
test_data = self.test_data_creator.Cycle()
res = self.test_data_creator.post_data_and_get_response(test_data, '/cycle/')
self.assertEqual(res.status_code, 201)
def test_Cycle_cannot_be_deleted_without_group(self):
user = User.objects.get(username='paulus')
user.groups.set([self.support_group]) # can add, cannot delete
# refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching
user = User.objects.get(username='paulus')
while not user.has_perm('tmssapp.add_cycle'):
user = User.objects.get(username='paulus')
# add
count = len(models.Cycle.objects.all())
test_data = self.test_data_creator.Cycle()
url = self.test_data_creator.post_data_and_get_url(test_data, '/cycle/')
self.assertEqual(count+1, len(models.Cycle.objects.all()))
# delete
response = requests.delete(url, auth=self.test_data_creator.auth)
self.assertEqual(response.status_code, 403)
self.assertEqual(count + 1, len(models.Cycle.objects.all()))
def test_Cycle_can_be_deleted_by_admin(self):
user = User.objects.get(username='paulus')
user.groups.set([self.support_group, self.admin_group]) # can add and delete
# refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching
user = User.objects.get(username='paulus')
while not user.has_perm('tmssapp.add_cycle'):
user = User.objects.get(username='paulus')
# add
count = len(models.Cycle.objects.all())
test_data = self.test_data_creator.Cycle()
url = self.test_data_creator.post_data_and_get_url(test_data, '/cycle/')
self.assertEqual(count+1, len(models.Cycle.objects.all()))
# delete
response = requests.delete(url, auth=self.test_data_creator.auth)
self.assertEqual(response.status_code, 204)
self.assertEqual(count, len(models.Cycle.objects.all()))
class SchedulingUnitObservingStrategyTemplateTestCase(unittest.TestCase):
def test_create_draft_from_all_strategy_templates(self):
'''Try to create a scheduling_unit_draft for each known strategy template.
Check that it is created properly.'''
scheduling_set = models.SchedulingSet.objects.create(**SchedulingSet_test_data("scheduling set"))
for strategy_template in models.SchedulingUnitObservingStrategyTemplate.objects.all():
logger.info("test_create_draft_from_all_strategy_templates: checking template '%s'", strategy_template.name)
with tmss_test_env.create_tmss_client() as client:
# check normal creation with all defaults
scheduling_unit_draft = client.create_scheduling_unit_draft_from_strategy_template(strategy_template.id, scheduling_set.id)
self.assertIsNotNone(scheduling_unit_draft)
self.assertEqual(scheduling_set.id, scheduling_unit_draft['scheduling_set_id'])
self.assertEqual(strategy_template.id, scheduling_unit_draft['observation_strategy_template_id'])
# check creation with overrides on top of all defaults
overrides = client.get_scheduling_unit_observing_strategy_template_specification_with_just_the_parameters(strategy_template.name, strategy_template.version)
scheduling_unit_draft = client.create_scheduling_unit_draft_from_strategy_template(strategy_template.id, scheduling_set.id, requirements_doc_overrides=overrides)
self.assertIsNotNone(scheduling_unit_draft)
self.assertEqual(scheduling_set.id, scheduling_unit_draft['scheduling_set_id'])
self.assertEqual(strategy_template.id, scheduling_unit_draft['observation_strategy_template_id'])
# check that other overrides are forbidden
with self.assertRaises(Exception):
overrides = {'foo': 'bar'}
client.create_scheduling_unit_draft_from_strategy_template(strategy_template.id, scheduling_set.id, requirements_doc_overrides=overrides)
def test_nested_SchedulingUnitObservingStrategyTemplate_are_filtered_according_to_SchedulingSet(self):
# setup
template_test_data = SchedulingUnitObservingStrategyTemplate_test_data(name="my_unique_observing_strategy_template")
observing_strategy_template = models.SchedulingUnitObservingStrategyTemplate.objects.create(**template_test_data)
scheduling_set = models.SchedulingSet.objects.create(**SchedulingSet_test_data("scheduling set"))
scheduling_unit_draft = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data("scheduling unit draft", scheduling_set=scheduling_set, observation_strategy_template=observing_strategy_template))
# assert the returned list contains related items, A list of length 1 is retrieved
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/scheduling_set/%s/scheduling_unit_observing_strategy_template/'
% scheduling_set.id, template_test_data, 1)
class SubmitTriggerTestCase(unittest.TestCase):
def test_submit_trigger_for_all_strategy_templates(self):
'''Try to create a scheduling_unit_draft/blueprint for each known strategy template via a trigger submission.'''
project = models.Project.objects.create(**Project_test_data(can_trigger=True))
scheduling_set = models.SchedulingSet.objects.create(**SchedulingSet_test_data("scheduling set", project=project))
with tmss_test_env.create_tmss_client() as client:
for strategy_template in models.SchedulingUnitObservingStrategyTemplate.objects.all():
if strategy_template.template_doc_complete_with_defaults.get('scheduling_constraints_template', '') != 'constraints':
logger.info("test_submit_trigger_for_all_strategy_templates: skipping template '%s' which has no known scheduling_constraints_template which can be used while dynamic scheduling", strategy_template.name)
continue
logger.info("test_submit_trigger_for_all_strategy_templates: checking template '%s'", strategy_template.name)
trigger_doc = client.get_trigger_specification_doc_for_scheduling_unit_observing_strategy_template(strategy_template.name, strategy_template.version)
trigger_doc['scheduling_set_id'] = scheduling_set.id
trigger_doc['mode'] = 'test'
result = client.submit_trigger(trigger_doc)
self.assertIsNotNone(result)
self.assertTrue('scheduling_unit_draft' in result['url']) # it's a draft (because the mode is "test")
self.assertEqual(scheduling_set.id, result['scheduling_set_id'])
self.assertEqual(strategy_template.id, result['observation_strategy_template_id'])
trigger_doc['mode'] = 'run'
result = client.submit_trigger(trigger_doc)
self.assertIsNotNone(result)
self.assertTrue('scheduling_unit_blueprint' in result['url']) # it's a blueprint (because the mode is "run")
self.assertEqual('schedulable', result['status'])
# test time scheduling_constraints: 'at' constraint
run_at = datetime.utcnow()
trigger_doc['mode'] = 'test'
trigger_doc['scheduling_unit_observing_strategy_template']['overrides']['scheduling_constraints_doc']['time']['at'] = run_at.isoformat()
scheduling_unit_draft = client.submit_trigger(trigger_doc)
# check that the draft has a 'before' time constraint (and assume that the scheduling constraints work in the dynamic scheduler)
self.assertEqual(run_at, parser.parse(scheduling_unit_draft['scheduling_constraints_doc']['time']['at']))
# test time scheduling_constraints: 'before' constraint
end_before = datetime.utcnow()
trigger_doc['mode'] = 'test'
trigger_doc['scheduling_unit_observing_strategy_template']['overrides']['scheduling_constraints_doc']['time']['before'] = end_before.isoformat()
scheduling_unit_draft = client.submit_trigger(trigger_doc)
# check that the draft has a 'before' time constraint (and assume that the scheduling constraints work in the dynamic scheduler)
self.assertEqual(end_before, parser.parse(scheduling_unit_draft['scheduling_constraints_doc']['time']['before']))
# todo: move to t_permissions (I tried, but it broke)
class SystemRolePermissionTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls):
# as superuser
test_data = test_data_creator.Cycle()
cls.cycle_url = test_data_creator.post_data_and_get_url(test_data, '/cycle/')
test_data = test_data_creator.Project()
cls.project_url = test_data_creator.post_data_and_get_url(test_data, '/project/')
# create test_data_creator with regular user
cls.test_data_creator = TMSSRESTTestDataCreator(BASE_URL, requests.auth.HTTPBasicAuth('paulus', 'pauluspass'))
response = requests.get(cls.test_data_creator.django_api_url + '/', auth=cls.test_data_creator.auth)
tmss_test_env.populate_permissions()
cls.scientist_group = Group.objects.get(name='scientist')
# Cycle
def test_Cycle_cannot_be_viewed_without_scientist_group(self):
user = User.objects.get(username='paulus')
user.groups.set([])
# refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching
user = User.objects.get(username='paulus')
while user.has_perm('tmssapp.view_cycle'):
user = User.objects.get(username='paulus')
self.assertFalse(user.has_perm('tmssapp.view_cycle'))
GET_and_assert_equal_expected_code(self, BASE_URL + '/cycle/', 403, auth=self.test_data_creator.auth)
GET_and_assert_equal_expected_code(self, self.cycle_url, 403, auth=self.test_data_creator.auth)
def test_Cycle_can_be_viewed_with_scientist_group(self):
user = User.objects.get(username='paulus')
user.groups.set([self.scientist_group])
# refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching
user = User.objects.get(username='paulus')
while not user.has_perm('tmssapp.view_cycle'):
user = User.objects.get(username='paulus')
self.assertTrue(user.has_perm('tmssapp.view_cycle'))
GET_and_assert_equal_expected_code(self, BASE_URL + '/cycle/', 200, auth=self.test_data_creator.auth)
GET_and_assert_equal_expected_code(self, self.cycle_url, 200, auth=self.test_data_creator.auth)
def test_Cycle_cannot_be_added_with_scientist_group(self):
user = User.objects.get(username='paulus')
user.groups.set([self.scientist_group])
# refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching
user = User.objects.get(username='paulus')
while user.has_perm('tmssapp.add_cycle'):
user = User.objects.get(username='paulus')
self.assertFalse(user.has_perm('tmssapp.add_cycle'))
test_data = self.test_data_creator.Cycle()
res = self.test_data_creator.post_data_and_get_response(test_data, '/cycle/')
self.assertEqual(res.status_code, 403)
def test_Cycle_cannot_be_changed_with_scientist_group(self):
user = User.objects.get(username='paulus')
user.groups.set([self.scientist_group])
# refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching
user = User.objects.get(username='paulus')
while user.has_perm('tmssapp.change_cycle'):
user = User.objects.get(username='paulus')
self.assertFalse(user.has_perm('tmssapp.change_cycle'))
# PATCH item
test_patch = {"description": "Trololol!"}
PATCH_and_assert_expected_response(self, self.cycle_url, test_patch, 403, {}, auth=self.test_data_creator.auth)
def test_Cycle_cannot_be_deleted_with_scientist_group(self):
user = User.objects.get(username='paulus')
user.groups.set([self.scientist_group])
# refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching
user = User.objects.get(username='paulus')
while user.has_perm('tmssapp.delete_cycle'):
user = User.objects.get(username='paulus')
self.assertFalse(user.has_perm('tmssapp.delete_cycle'))
# try to delete
count = len(models.Cycle.objects.all())
response = requests.delete(self.cycle_url, auth=self.test_data_creator.auth)
self.assertEqual(response.status_code, 403)
self.assertEqual(count, len(models.Cycle.objects.all()))
# Project
def test_Project_cannot_be_viewed_without_scientist_group(self):
user = User.objects.get(username='paulus')
user.groups.set([])
# refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching
user = User.objects.get(username='paulus')
while user.has_perm('tmssapp.view_project'):
user = User.objects.get(username='paulus')
self.assertFalse(user.has_perm('tmssapp.view_project'))
# Note: with just the model permissions, you'd expect a permission denied error on a listing.
# But since we do also allow viewing projects based on project permissions (users should be able to list
# their own projects), we get an empty list instead.
#GET_and_assert_equal_expected_code(self, BASE_URL + '/project/', 403, auth=self.test_data_creator.auth)
list = GET_and_assert_equal_expected_code(self, BASE_URL + '/project/', 200, auth=self.test_data_creator.auth)
self.assertEqual(list['count'], 0)
GET_and_assert_equal_expected_code(self, self.project_url, 403, auth=self.test_data_creator.auth)
def test_Project_can_be_viewed_with_scientist_group(self):
user = User.objects.get(username='paulus')
user.groups.set([self.scientist_group])
# refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching
user = User.objects.get(username='paulus')
while not user.has_perm('tmssapp.view_project'):
user = User.objects.get(username='paulus')
self.assertTrue(user.has_perm('tmssapp.view_project'))
GET_and_assert_equal_expected_code(self, BASE_URL + '/project/', 200, auth=self.test_data_creator.auth)
GET_and_assert_equal_expected_code(self, self.project_url, 200, auth=self.test_data_creator.auth)
def test_Project_cannot_be_added_with_scientist_group(self):
user = User.objects.get(username='paulus')
user.groups.set([self.scientist_group])
# refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching
user = User.objects.get(username='paulus')
while user.has_perm('tmssapp.add_project'):
user = User.objects.get(username='paulus')
self.assertFalse(user.has_perm('tmssapp.add_project'))
test_data = self.test_data_creator.Project(cycle_urls=[self.cycle_url])
res = self.test_data_creator.post_data_and_get_response(test_data, '/project/')
self.assertEqual(res.status_code, 403)
def test_Project_cannot_be_changed_with_scientist_group(self):
user = User.objects.get(username='paulus')
user.groups.set([self.scientist_group])
# refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching
user = User.objects.get(username='paulus')
while user.has_perm('tmssapp.change_project'):
user = User.objects.get(username='paulus')
self.assertFalse(user.has_perm('tmssapp.change_project'))
# PATCH item
test_patch = {"description": "Trololol!"}
PATCH_and_assert_expected_response(self, self.project_url, test_patch, 403, {}, auth=self.test_data_creator.auth)
def test_Project_cannot_be_deleted_with_scientist_group(self):
user = User.objects.get(username='paulus')
user.groups.set([self.scientist_group])
# refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching
user = User.objects.get(username='paulus')
while user.has_perm('tmssapp.delete_project'):
user = User.objects.get(username='paulus')
self.assertFalse(user.has_perm('tmssapp.delete_project'))
# try to delete
count = len(models.Project.objects.all())
response = requests.delete(self.project_url, auth=self.test_data_creator.auth)
self.assertEqual(response.status_code, 403)
self.assertEqual(count, len(models.Project.objects.all()))
if __name__ == "__main__":
unittest.main()