Select Git revision
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
t_tmssapp_specification_REST_API.py 203.28 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.
import copy
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_and_connectors()
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 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 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"}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(schedulingset_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, test_patch)
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_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"}
# 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_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.Subtask.objects.all().delete()
models.TaskBlueprint.objects.all().delete()
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
scheduling_set = models.SchedulingSet.objects.create(**SchedulingSet_test_data())
strategy_template = models.SchedulingUnitObservingStrategyTemplate.objects.get(name="UC1 CTC+pipelines")
from lofar.sas.tmss.tmss.tmssapp.tasks import create_scheduling_unit_draft_from_observing_strategy_template
su_draft_1 = create_scheduling_unit_draft_from_observing_strategy_template(strategy_template, scheduling_set, name='sud1_%s' % uuid.uuid4(), description="<none>",
specifications_doc_overrides={'tasks': {'Target Observation': {'specifications_doc': {'SAPs': [{'digital_pointing': {'angle1': 0.333, 'angle2': 0.999, 'direction_type': 'J2000'}}]}}}})
su_draft_2 = create_scheduling_unit_draft_from_observing_strategy_template(strategy_template, scheduling_set, name='sud1_%s' % uuid.uuid4(), description="<none>",
specifications_doc_overrides={'tasks': {'Target Observation': {'specifications_doc': {'SAPs': [{'digital_pointing': {'angle1': 0.111, 'angle2': 0.777, 'direction_type': 'J2000'}}]}}}})
# 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_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_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_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 = {"output_pinned": not r_dict['output_pinned']}
# 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(scheduled_on_sky_start_time=datetime(2050, 1, 1, 10, 0, 0), scheduled_on_sky_stop_time=datetime(2050, 1, 1, 14, 0, 0), task_blueprint=models.TaskBlueprint.objects.create(**TaskBlueprint_test_data())))
subtask_2 = models.Subtask.objects.create(**Subtask_test_data(scheduled_on_sky_start_time=datetime(2050, 1, 5, 10, 0, 0), scheduled_on_sky_stop_time=datetime(2050, 1, 5, 14, 0, 0), task_blueprint=models.TaskBlueprint.objects.create(**TaskBlueprint_test_data())))
# assert
response_1 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?scheduled_on_sky_start_time_after=2050-01-01T9:00:00&scheduled_on_sky_stop_time_before=2050-01-01T15:00:00', 200)
response_2 = GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/?scheduled_on_sky_start_time_after=2050-01-01T9:00:00&scheduled_on_sky_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.Subtask.objects.all().delete()
models.TaskBlueprint.objects.all().delete()
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_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 = {"output_pinned": not r_dict['output_pinned']}
# 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_allows_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 data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', test_data, 201, {})
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_SETNULL_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 successful
response = requests.delete(draft_url, auth=AUTH)
self.assertEqual(204, response.status_code)
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()
task_blueprint = models.TaskBlueprint.objects.create(**test_data_1)
st_test_data_1 = Subtask_test_data(task_blueprint=task_blueprint, primary=True)
st_test_data_2 = Subtask_test_data(task_blueprint=task_blueprint, primary=False)
subtask_1 = models.Subtask.objects.create(**st_test_data_1)
subtask_2 = models.Subtask.objects.create(**st_test_data_2)
# 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(task_blueprint=task_blueprint_1))
subtask_2 = models.Subtask.objects.create(**Subtask_test_data(task_blueprint=task_blueprint_2))
subtask_3 = models.Subtask.objects.create(**Subtask_test_data(task_blueprint=task_blueprint_3))
# 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 should fail on immutable blueprint
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, 500, trd_test_data1)
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)
# PATCH item should be forbidden (immutable blueprint)
test_patch = {"selection_doc": {"foo": "patched"}}
PATCH_and_assert_expected_response(self, url, test_patch, 500, trb_test_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_allows_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, 201, {})
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_url=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, specifications_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, specifications_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()