-
Jorrit Schaap authoredJorrit Schaap authored
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
t_tmssapp_specification_REST_API.py 137.57 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.
# todo: I am still a bit under the impression that we re-test Django functionality that we can expect to just work
# todo: with some of these tests. On the other hand a lot of these provide us a nice basis for differentiating out
# todo: behavior in a controlled way.
# todo: We should probably also fully test behavior wrt mandatory and nullable fields.
from datetime import datetime
import unittest
import logging
logger = logging.getLogger(__name__)
logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO)
from lofar.common.test_utils import skip_integration_tests
if skip_integration_tests():
exit(3)
# 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
# import and setup test data creator
from lofar.sas.tmss.test.tmss_test_data_rest import TMSSRESTTestDataCreator
test_data_creator = TMSSRESTTestDataCreator(BASE_URL, AUTH)
class BasicFunctionTestCase(unittest.TestCase):
# todo: test_welcome_page (once we have one :))
pass
class GeneratorTemplateTestCase(unittest.TestCase):
def test_generator_template_list_apiformat(self):
r = requests.get(BASE_URL + '/generator_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Generator Template List" in r.content.decode('utf8'))
def test_generator_template_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/generator_template/1234321/', 404)
def test_generator_template_POST_and_GET(self):
# POST and GET a new item and assert correctness
test_data = test_data_creator.GeneratorTemplate()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, test_data)
def test_generator_template_PUT_invalid_raises_error(self):
test_data = test_data_creator.GeneratorTemplate()
PUT_and_assert_expected_response(self, BASE_URL + '/generator_template/9876789876/', test_data, 404, {})
def test_generator_template_PUT(self):
# POST new item, verify
test_data = test_data_creator.GeneratorTemplate()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, test_data)
# PUT new values, verify
test_data2 = test_data_creator.GeneratorTemplate("generatortemplate2")
PUT_and_assert_expected_response(self, url, test_data2, 200, test_data2)
GET_OK_and_assert_equal_expected_response(self, url, test_data2)
def test_generator_template_PATCH(self):
# POST new item, verify
test_data = test_data_creator.GeneratorTemplate()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, test_data)
test_patch = {"version": 'v6.28318530718',
"schema": {"mykey": "my better value"}}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(test_data_creator.GeneratorTemplate())
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_generator_template_DELETE(self):
# POST new item, verify
test_data = test_data_creator.GeneratorTemplate()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_GET_generator_template_view_returns_correct_entry(self):
test_data_1 = GeneratorTemplate_test_data("test_generator_template_1")
test_data_2 = GeneratorTemplate_test_data("test_generator_template_2")
id1 = models.GeneratorTemplate.objects.create(**test_data_1).id
id2 = models.GeneratorTemplate.objects.create(**test_data_2).id
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/generator_template/' + str(id1), test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/generator_template/' + str(id2), test_data_2)
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()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url+'?format=json', test_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()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, test_data)
# PUT new values, verify
test_data2 = test_data_creator.SchedulingUnitTemplate("schedulingunittemplate2")
PUT_and_assert_expected_response(self, url, test_data2, 200, test_data2)
GET_OK_and_assert_equal_expected_response(self, url, test_data2)
def test_scheduling_unit_template_PATCH(self):
# POST new item, verify
test_data = test_data_creator.SchedulingUnitTemplate()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, test_data)
test_patch = {"version": 'v6.28318530718',
"schema": {"mykey": "my better value"}}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(test_data_creator.SchedulingUnitTemplate())
expected_data.update(test_patch)
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()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_GET_scheduling_unit_template_view_returns_correct_entry(self):
test_data_1 = SchedulingUnitTemplate_test_data("scheduling_unit_template_1")
test_data_2 = SchedulingUnitTemplate_test_data("scheduling_unit_template_2")
id1 = models.SchedulingUnitTemplate.objects.create(**test_data_1).id
id2 = models.SchedulingUnitTemplate.objects.create(**test_data_2).id
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/scheduling_unit_template/' + str(id1), test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/scheduling_unit_template/' + str(id2), test_data_2)
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()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url + '?format=json', test_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()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, test_data)
# PUT new values, verify
test_data2 = test_data_creator.TaskTemplate("tasktemplate2")
PUT_and_assert_expected_response(self, url, test_data2, 200, test_data2)
GET_OK_and_assert_equal_expected_response(self, url, test_data2)
def test_task_template_PATCH(self):
# POST new item, verify
test_data = test_data_creator.TaskTemplate()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, test_data)
test_patch = {"version": 'v6.28318530718',
"schema": {"mykey": "my better value"},
}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(test_data_creator.TaskTemplate())
expected_data.update(test_patch)
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()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, test_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)
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()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_selection_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url+'?format=json', test_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()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_selection_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, test_data)
# PUT new values, verify
test_data2 = test_data_creator.TaskRelationSelectionTemplate("taskrelationselectiontemplate2")
PUT_and_assert_expected_response(self, url, test_data2, 200, test_data2)
GET_OK_and_assert_equal_expected_response(self, url, test_data2)
def test_task_relation_selection_template_PATCH(self):
# POST new item, verify
test_data = test_data_creator.TaskRelationSelectionTemplate()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_selection_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, test_data)
test_patch = {"version": 'v6.28318530718',
"schema": {"mykey": "my better value"},
}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(test_data_creator.TaskRelationSelectionTemplate())
expected_data.update(test_patch)
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()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_selection_template/', test_data, 201, test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, test_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.input_of_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
cls.output_of_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(input_of_url=self.input_of_url, output_of_url=self.output_of_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(input_of_url=self.input_of_url, output_of_url=self.output_of_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(input_of_url=self.input_of_url, output_of_url=self.output_of_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(input_of_url=self.input_of_url, output_of_url=self.output_of_url))
test_data_invalid['dataformats'] = [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['dataformats']))
def test_task_connector_POST_nonexistant_input_of_raises_error(self):
# POST a new item with wrong reference
test_data_invalid = dict(test_data_creator.TaskConnectorType(input_of_url=self.input_of_url, output_of_url=self.output_of_url))
test_data_invalid['input_of'] = 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['input_of']))
def test_task_connector_POST_nonexistant_output_of_raises_error(self):
# POST a new item with wrong reference
test_data_invalid = dict(test_data_creator.TaskConnectorType(input_of_url=self.input_of_url, output_of_url=self.output_of_url))
test_data_invalid['output_of'] = 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['output_of']))
def test_task_connector_POST_existing_outputs_works(self):
# First POST a new item to reference
test_data = test_data_creator.TaskTemplate()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', test_data, 201, test_data)
url = r_dict['url']
# POST a new item with correct reference
test_data_valid = dict(test_data_creator.TaskConnectorType(input_of_url=self.input_of_url, output_of_url=self.output_of_url))
test_data_valid['output_of'] = 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(input_of_url=self.input_of_url, output_of_url=self.output_of_url), 404, {})
def test_task_connector_PUT(self):
tc_test_data1 = test_data_creator.TaskConnectorType(role="correlator", input_of_url=self.input_of_url, output_of_url=self.output_of_url)
tc_test_data2 = test_data_creator.TaskConnectorType(role="beamformer", input_of_url=self.input_of_url, output_of_url=self.output_of_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(input_of_url=self.input_of_url, output_of_url=self.output_of_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/',
"dataformats": [BASE_URL + '/dataformat/Beamformed/',
BASE_URL + '/dataformat/MeasurementSet/']}
# 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(input_of_url=self.input_of_url, output_of_url=self.output_of_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_inputs_template_deleted(self):
input_of_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
tc_test_data = test_data_creator.TaskConnectorType(input_of_url=input_of_url, output_of_url=self.output_of_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, input_of_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_relation_blueprint_CASCADE_behavior_on_outputs_template_deleted(self):
output_of_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
tc_test_data = test_data_creator.TaskConnectorType(input_of_url=self.input_of_url, output_of_url=output_of_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, output_of_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_GET_task_connector_view_returns_correct_entry(self):
test_data_1 = TaskConnectorType_test_data()
test_data_2 = TaskConnectorType_test_data()
id1 = models.TaskConnectorType.objects.create(**test_data_1).id
id2 = models.TaskConnectorType.objects.create(**test_data_2).id
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_connector_type/' + str(id1), test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_connector_type/' + str(id2), test_data_2)
class DefaultTemplates(unittest.TestCase):
def test_default_generator_template_POST(self):
test_data = test_data_creator.GeneratorTemplate()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', test_data, 201, test_data)
url = r_dict['url']
test_data_1 = dict(test_data_creator.DefaultTemplates())
test_data_1['template'] = url
POST_and_assert_expected_response(self, BASE_URL + '/default_generator_template/', test_data_1, 201, test_data_1)
def test_default_scheduling_unit_template_POST(self):
test_data = test_data_creator.SchedulingUnitTemplate()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/',
test_data, 201,
test_data)
url = r_dict['url']
test_data_1 = dict(test_data_creator.DefaultTemplates())
test_data_1['template'] = url
POST_and_assert_expected_response(self, BASE_URL + '/default_scheduling_unit_template/', test_data_1, 201, test_data_1)
def test_default_task_template_POST(self):
test_data = test_data_creator.TaskTemplate()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/',
test_data, 201,
test_data)
url = r_dict['url']
test_data_1 = dict(test_data_creator.DefaultTemplates())
test_data_1['template'] = url
POST_and_assert_expected_response(self, BASE_URL + '/default_task_template/', test_data_1, 201, test_data_1)
def test_default_task_relation_selection_template_POST(self):
test_data = test_data_creator.TaskRelationSelectionTemplate()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_selection_template/',
test_data, 201,
test_data)
url = r_dict['url']
test_data_1 = dict(test_data_creator.DefaultTemplates())
test_data_1['template'] = url
POST_and_assert_expected_response(self, BASE_URL + '/default_task_relation_selection_template/', test_data_1, 201, test_data_1)
def test_default_generator_template_PROTECT_behavior_on_template_deleted(self):
# POST with dependency
test_data = test_data_creator.GeneratorTemplate()
template_url = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/',
test_data, 201,
test_data)['url']
test_data2 = dict(test_data_creator.DefaultTemplates("defaulttemplate2"))
test_data2['template'] = template_url
POST_and_assert_expected_response(self, BASE_URL + '/default_generator_template/',
test_data2, 201, test_data2)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, template_url, test_data)
def test_default_scheduling_unit_template_PROTECT_behavior_on_template_deleted(self):
# POST with dependency
test_data = test_data_creator.SchedulingUnitTemplate()
template_url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/',
test_data, 201,
test_data)['url']
test_data2 = dict(test_data_creator.DefaultTemplates("defaulttemplate2"))
test_data2['template'] = template_url
POST_and_assert_expected_response(self, BASE_URL + '/default_scheduling_unit_template/',
test_data2, 201, test_data2)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, template_url, test_data)
def test_default_task_template_PROTECT_behavior_on_template_deleted(self):
# POST with dependency
test_data = test_data_creator.TaskTemplate()
template_url = POST_and_assert_expected_response(self, BASE_URL + '/task_template/',
test_data, 201,
test_data)['url']
test_data2 = dict(test_data_creator.DefaultTemplates("defaulttemplate2"))
test_data2['template'] = template_url
POST_and_assert_expected_response(self, BASE_URL + '/default_task_template/',
test_data2, 201, test_data2)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, template_url, test_data)
def test_default_task_relation_selection_template_PROTECT_behavior_on_template_deleted(self):
# POST with dependency
test_data = test_data_creator.TaskRelationSelectionTemplate()
template_url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_selection_template/',
test_data, 201,
test_data)['url']
test_data2 = dict(test_data_creator.DefaultTemplates("defaulttemplate2"))
test_data2['template'] = template_url
POST_and_assert_expected_response(self, BASE_URL + '/default_task_relation_selection_template/',
test_data2, 201, test_data2)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, template_url, test_data)
class CycleTestCase(unittest.TestCase):
def test_cycle_list_apiformat(self):
r = requests.get(BASE_URL + '/cycle/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Cycle List" in r.content.decode('utf8'))
def test_cycle_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/cycle/1234321/', 404)
def test_cycle_POST_and_GET(self):
# POST and GET a new item and assert correctness
cycle_test_data = test_data_creator.Cycle()
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', cycle_test_data, 201, cycle_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, cycle_test_data)
def test_cycle_PUT_invalid_raises_error(self):
PUT_and_assert_expected_response(self, BASE_URL + '/cycle/9876789876/', test_data_creator.Cycle(), 404, {})
def test_cycle_PUT(self):
cycle_test_data = test_data_creator.Cycle()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', cycle_test_data, 201, cycle_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, cycle_test_data)
# PUT new values, verify
test_data = dict(test_data_creator.Cycle("other description"))
test_data['name'] = cycle_test_data['name'] # since name is PK, need to keep that unchanged
PUT_and_assert_expected_response(self, url, test_data, 200, test_data)
GET_OK_and_assert_equal_expected_response(self, url, test_data)
def test_cycle_PATCH(self):
cycle_test_data = test_data_creator.Cycle()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', cycle_test_data, 201, cycle_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, cycle_test_data)
test_patch = {"start": datetime(year=2015, month=10, day=21).isoformat()}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(cycle_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_cycle_DELETE(self):
cycle_test_data = test_data_creator.Cycle()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', cycle_test_data, 201, cycle_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, cycle_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_GET_cycle_list_shows_entry(self):
test_data_1 = Cycle_test_data() # uuid makes name unique
test_data_1["start"] = datetime(2999, 1, 1) # cycles are ordered by start, so make this the latest date and hence the latest cycle
models.Cycle.objects.create(**test_data_1)
nbr_results = models.Cycle.objects.count()
test_data_1["start"] = test_data_1["start"].isoformat()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/cycle/', test_data_1, nbr_results)
def test_GET_cycle_view_returns_correct_entry(self):
test_data_1 = Cycle_test_data() # uuid makes name unique
test_data_2 = Cycle_test_data()
id1 = models.Cycle.objects.create(**test_data_1).name # name is pk
id2 = models.Cycle.objects.create(**test_data_2).name # name is pk
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/cycle/' + str(id1), test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/cycle/' + str(id2), test_data_2)
def test_cycle_contains_list_of_releated_projects(self):
cycle_test_data_1 = Cycle_test_data()
project_test_data_1 = Project_test_data() # uuid makes name unique
project_test_data_2 = Project_test_data() # uuid makes name unique
cycle = models.Cycle.objects.create(**cycle_test_data_1)
project1 = models.Project.objects.create(**project_test_data_1)
project1.cycles.set([cycle])
project1.save()
project2 = models.Project.objects.create(**project_test_data_2)
project2.cycles.set([cycle])
project2.save()
response_data = GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/cycle/' + cycle.name, cycle_test_data_1)
assertUrlList(self, response_data['projects'], [project1, project2])
class ProjectTestCase(unittest.TestCase):
def test_project_list_apiformat(self):
r = requests.get(BASE_URL + '/project/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Project List" in r.content.decode('utf8'))
def test_project_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/project/1234321/', 404)
def test_project_POST_and_GET(self):
project_test_data = test_data_creator.Project()
# POST and GET a new item and assert correctness
expected = project_test_data.copy()
expected.pop('quota') # exclude quota from comparison, because these get auto-generated
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', project_test_data, 201, expected)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected)
def test_project_PUT_invalid_raises_error(self):
PUT_and_assert_expected_response(self, BASE_URL + '/project/9876789876/', test_data_creator.Project(), 404, {})
def test_project_PUT(self):
project_test_data = test_data_creator.Project()
# POST new item, verify
expected = project_test_data.copy()
expected.pop('quota') # exclude quota from comparison, because these get auto-generated
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', project_test_data, 201, expected)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected)
# PUT new values, verify
test_data = dict(test_data_creator.Project("other description"))
test_data['name'] = project_test_data['name'] # since name is PK, need to keep that unchanged
expected = test_data.copy()
expected.pop('quota') # exclude quota from comparison, because these get auto-generated
PUT_and_assert_expected_response(self, url, test_data, 200, expected)
GET_OK_and_assert_equal_expected_response(self, url, expected)
def test_project_PATCH(self):
project_test_data = test_data_creator.Project()
# POST new item, verify
expected = project_test_data.copy()
expected.pop('quota') # exclude quota from comparison, because these get auto-generated
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', project_test_data, 201, expected)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected)
test_patch = {"priority_rank": 1.0,
"tags": ["SUPERIMPORTANT"]}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected)
def test_project_DELETE(self):
project_test_data = test_data_creator.Project()
# POST new item, verify
expected = project_test_data.copy()
expected.pop('quota') # exclude quota from comparison, because these get auto-generated
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', project_test_data, 201, expected)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected)
# DELETE related auto-generated quota first
quotas = r_dict['quota']
for quota in quotas:
DELETE_and_assert_gone(self, quota)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_project_PROTECT_behavior_on_cycle_deleted(self):
# POST new item with dependencies
cycle_test_data = test_data_creator.Cycle()
cycle_url = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', cycle_test_data, 201, cycle_test_data)['url']
test_data = dict(test_data_creator.Project())
test_data['cycles'] = [cycle_url]
expected = test_data.copy()
expected.pop('quota') # exclude quota from comparison, because these get auto-generated
url = POST_and_assert_expected_response(self, BASE_URL + '/project/', test_data, 201, expected)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, expected)
# add project reference to cycle test data (we make Django add that to the cycle in serializer)
cycle_test_data['projects'] = [url]
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(cycle_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, cycle_url, cycle_test_data)
def test_GET_project_list_shows_entry(self):
test_data_1 = Project_test_data() # uuid makes name unique
test_data_1["name"] = "ZZZZZZZZZZZZZZZ" # projects are ordered by name, so make this the latest project (in sorted alphabetical order)
models.Project.objects.create(**test_data_1)
nbr_results = models.Project.objects.count()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/project/', test_data_1, nbr_results)
def test_GET_project_view_returns_correct_entry(self):
test_data_1 = Project_test_data() # uuid makes name unique
test_data_2 = Project_test_data()
id1 = models.Project.objects.create(**test_data_1).name # name is pk
id2 = models.Project.objects.create(**test_data_2).name
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/project/' + str(id1), test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/project/' + str(id2), test_data_2)
def test_nested_projects_are_filtered_according_to_cycle(self):
test_data_1 = dict(Project_test_data()) # uuid makes project unique
project_1 = models.Project.objects.create(**test_data_1)
cycle_1 = models.Cycle.objects.create(**Cycle_test_data())
project_1.cycles.set([cycle_1])
# assert the returned list contains related items, A list of length 1 is retrieved
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/cycle/%s/project/' % cycle_1.name, test_data_1, 1)
class ResourceTypeTestCase(unittest.TestCase):
def test_resource_type_list_apiformat(self):
r = requests.get(BASE_URL + '/resource_type/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Resource Type List" in r.content.decode('utf8'))
def test_resource_type_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/resource_type/1234321/', 404)
def test_resource_type_POST_and_GET(self):
resource_type_test_data = test_data_creator.ResourceType()
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/resource_type/', resource_type_test_data, 201, resource_type_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, resource_type_test_data)
class ProjectQuotaTestCase(unittest.TestCase):
def test_project_quota_list_apiformat(self):
r = requests.get(BASE_URL + '/project_quota/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Project Quota List" in r.content.decode('utf8'))
def test_project_quota_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/project_quota/1234321/', 404)
def test_project_quota_POST_and_GET(self):
project_quota_test_data = test_data_creator.ProjectQuota()
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project_quota/', project_quota_test_data, 201, project_quota_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, project_quota_test_data)
def test_project_quota_POST_and_GET(self):
project_quota_test_data = test_data_creator.ProjectQuota()
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project_quota/', project_quota_test_data, 201, project_quota_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, project_quota_test_data)
def test_project_quota_PUT_invalid_raises_error(self):
PUT_and_assert_expected_response(self, BASE_URL + '/project_quota/9876789876/', test_data_creator.Project(), 404, {})
def test_project_quota_PUT(self):
project_quota_test_data = test_data_creator.ProjectQuota()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project_quota/', project_quota_test_data, 201, project_quota_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, project_quota_test_data)
# PUT new values, verify
test_data = dict(test_data_creator.ProjectQuota("other description"))
PUT_and_assert_expected_response(self, url, test_data, 200, test_data)
GET_OK_and_assert_equal_expected_response(self, url, test_data)
def test_project_quota_PATCH(self):
project_quota_test_data = test_data_creator.ProjectQuota()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project_quota/', project_quota_test_data, 201, project_quota_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, project_quota_test_data)
test_patch = {"value": 500}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(project_quota_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_project_quota_DELETE(self):
project_quota_test_data = test_data_creator.ProjectQuota()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project_quota/', project_quota_test_data, 201, project_quota_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, project_quota_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_project_quota_PROTECT_behavior_on_project_deleted(self):
# POST new item with dependencies
project_test_data = test_data_creator.Project()
expected = project_test_data.copy()
expected.pop('quota') # exclude quota from comparison, because these get auto-generated
project_url = POST_and_assert_expected_response(self, BASE_URL + '/project/', project_test_data, 201, expected)['url']
project_quota_test_data = dict(test_data_creator.ProjectQuota(project_url=project_url))
project_quota_url = POST_and_assert_expected_response(self, BASE_URL + '/project_quota/', project_quota_test_data, 201, project_quota_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, project_quota_url, project_quota_test_data)
project_test_data['project'] = [project_quota_url] # add the
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(project_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, project_quota_url, project_quota_test_data)
class SchedulingSetTestCase(unittest.TestCase):
def test_scheduling_set_list_apiformat(self):
r = requests.get(BASE_URL + '/scheduling_set/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Scheduling Set List" in r.content.decode('utf8'))
def test_scheduling_set_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_set/1234321/', 404)
def test_scheduling_set_POST_and_GET(self):
schedulingset_test_data = test_data_creator.SchedulingSet()
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', schedulingset_test_data, 201, schedulingset_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingset_test_data)
def test_scheduling_set_PUT_invalid_raises_error(self):
schedulingset_test_data = test_data_creator.SchedulingSet()
PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_set/9876789876/', schedulingset_test_data, 404, {})
def test_scheduling_set_PUT(self):
project_url = test_data_creator.post_data_and_get_url(test_data_creator.Project(), '/project/')
schedulingset_test_data = test_data_creator.SchedulingSet(project_url=project_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', schedulingset_test_data, 201, schedulingset_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingset_test_data)
schedulingset_test_data2 = test_data_creator.SchedulingSet("schedulingset2", project_url=project_url)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, schedulingset_test_data2, 200, schedulingset_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, schedulingset_test_data2)
def test_scheduling_set_PATCH(self):
schedulingset_test_data = test_data_creator.SchedulingSet()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', schedulingset_test_data, 201, schedulingset_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingset_test_data)
test_patch = {"description": "This is a new and improved description",
"generator_doc": '{"para": "meter"}'}
# 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, expected_data)
def test_scheduling_set_DELETE(self):
schedulingset_test_data = test_data_creator.SchedulingSet()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', schedulingset_test_data, 201, schedulingset_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingset_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_scheduling_set_PROTECT_behavior_on_project_deleted(self):
project_url = test_data_creator.post_data_and_get_url(test_data_creator.Project(), '/project/')
schedulingset_test_data = test_data_creator.SchedulingSet(project_url=project_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', schedulingset_test_data, 201, schedulingset_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, schedulingset_test_data)
# fetch project data before we delete it (for later comparison)
project_test_data = GET_and_assert_equal_expected_code(self, project_url, 200)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(project_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, project_url, project_test_data)
def test_scheduling_set_SET_NULL_behavior_on_generator_template_deleted(self):
generator_template_url = test_data_creator.post_data_and_get_url(test_data_creator.GeneratorTemplate(), '/generator_template/')
schedulingset_test_data = test_data_creator.SchedulingSet(generator_template_url=generator_template_url)
# POST new item
test_data = dict(schedulingset_test_data)
url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', test_data, 201, test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, test_data)
# DELETE dependency
DELETE_and_assert_gone(self, generator_template_url)
# assert
test_data['generator_template'] = None
GET_OK_and_assert_equal_expected_response(self, url, test_data)
def test_GET_SchedulingSet_list_shows_entry(self):
test_data_1 = SchedulingSet_test_data()
models.SchedulingSet.objects.create(**test_data_1)
nbr_results = models.SchedulingSet.objects.count()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/scheduling_set/', test_data_1, nbr_results)
def test_GET_SchedulingSet_view_returns_correct_entry(self):
test_data_1 = SchedulingSet_test_data() # uuid makes name unique
test_data_2 = SchedulingSet_test_data()
id1 = models.SchedulingSet.objects.create(**test_data_1).id
id2 = models.SchedulingSet.objects.create(**test_data_2).id
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/scheduling_set/' + str(id1), test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/scheduling_set/' + str(id2), test_data_2)
def test_SchedulingSet_contains_list_of_releated_SchedulingUnitDraft(self):
test_data_1 = SchedulingSet_test_data()
scheduling_set = models.SchedulingSet.objects.create(**test_data_1)
scheduling_unit_draft_1 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data("scheduler draft one"))
scheduling_unit_draft_1.scheduling_set = scheduling_set
scheduling_unit_draft_1.save()
scheduling_unit_draft_2 = models.SchedulingUnitDraft.objects.create(**SchedulingUnitDraft_test_data("scheduler draft one"))
scheduling_unit_draft_2.scheduling_set = scheduling_set
scheduling_unit_draft_2.save()
response_data = GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/scheduling_set/%d' % scheduling_set.id, test_data_1)
assertUrlList(self, response_data['scheduling_unit_drafts'], [scheduling_unit_draft_1, scheduling_unit_draft_2])
class SchedulingUnitDraftTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.scheduling_set_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingSet(), '/scheduling_set/')
cls.template_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitTemplate(), '/scheduling_unit_template/')
def test_scheduling_unit_draft_list_apiformat(self):
r = requests.get(BASE_URL + '/scheduling_unit_draft/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Scheduling Unit Draft List" in r.content.decode('utf8'))
def test_scheduling_unit_draft_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_draft/1234321/', 404)
def test_scheduling_unit_draft_POST_and_GET(self):
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(scheduling_set_url=self.scheduling_set_url, template_url=self.template_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingunitdraft_test_data)
def test_scheduling_unit_draft_PUT_invalid_raises_error(self):
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(scheduling_set_url=self.scheduling_set_url, template_url=self.template_url)
PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/9876789876/', schedulingunitdraft_test_data, 404, {})
def test_scheduling_unit_draft_PUT(self):
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(scheduling_set_url=self.scheduling_set_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingunitdraft_test_data)
schedulingunitdraft_test_data2 = test_data_creator.SchedulingUnitDraft("my_scheduling_unit_draft2", scheduling_set_url=self.scheduling_set_url, template_url=self.template_url)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, schedulingunitdraft_test_data2, 200, schedulingunitdraft_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, schedulingunitdraft_test_data2)
def test_scheduling_unit_draft_PATCH(self):
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(scheduling_set_url=self.scheduling_set_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingunitdraft_test_data)
test_patch = {"description": "This is a new and improved description",
"requirements_doc": '{"para": "meter"}'}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(schedulingunitdraft_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_scheduling_unit_draft_DELETE(self):
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(scheduling_set_url=self.scheduling_set_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, schedulingunitdraft_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_scheduling_unit_draft_CASCADE_behavior_on_scheduling_unit_template_deleted(self):
template_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitTemplate(), '/scheduling_unit_template/')
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(template_url=template_url, scheduling_set_url=self.scheduling_set_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, schedulingunitdraft_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, template_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_scheduling_unit_draft_CASCADE_behavior_on_scheduling_set_deleted(self):
scheduling_set_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingSet(), '/scheduling_set/')
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(scheduling_set_url=scheduling_set_url, template_url=self.template_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, schedulingunitdraft_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, scheduling_set_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_scheduling_unit_draft_SET_NULL_behavior_on_copies_deleted(self):
schedulingunitdraft_test_data = test_data_creator.SchedulingUnitDraft(scheduling_set_url=self.scheduling_set_url, template_url=self.template_url)
# POST new item with dependency
copy_url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)['url']
test_data = dict(schedulingunitdraft_test_data)
test_data['copies'] = copy_url
url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', test_data, 201, test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, test_data)
# DELETE dependency
DELETE_and_assert_gone(self, copy_url)
# assert
test_data['copies'] = None
GET_OK_and_assert_equal_expected_response(self, url, test_data)
def test_GET_SchedulingUnitDraft_list_view_shows_entry(self):
test_data_1 = SchedulingUnitDraft_test_data("scheduler unit draft one")
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)
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")
tdt_test_data_2 = TaskDraft_test_data("task draft two")
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])
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": '{"para": "meter"}'}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(taskdraft_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_task_draft_DELETE(self):
taskdraft_test_data = test_data_creator.TaskDraft(scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data, 201, taskdraft_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, taskdraft_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_task_draft_CASCADE_behavior_on_task_template_deleted(self):
template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
taskdraft_test_data = test_data_creator.TaskDraft(name="task draft 2", template_url=template_url, scheduling_unit_draft_url=self.scheduling_unit_draft_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data, 201, taskdraft_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, taskdraft_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, template_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_draft_CASCADE_behavior_on_scheduling_unit_draft_deleted(self):
scheduling_unit_draft_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitDraft(), '/scheduling_unit_draft/')
taskdraft_test_data = test_data_creator.TaskDraft(name="task draft 2", scheduling_unit_draft_url=scheduling_unit_draft_url, template_url=self.template_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data, 201, taskdraft_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, taskdraft_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, scheduling_unit_draft_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_draft_SET_NULL_behavior_on_copies_deleted(self):
taskdraft_test_data1 = test_data_creator.TaskDraft(name="the one", 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 with dependency
copy_url = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data2, 201, taskdraft_test_data2)['url']
test_data = dict(taskdraft_test_data1)
test_data['copies'] = copy_url
url = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', test_data, 201, test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, test_data)
# DELETE dependency
DELETE_and_assert_gone(self, copy_url)
# assert
test_data['copies'] = None
GET_OK_and_assert_equal_expected_response(self, url, test_data)
def test_GET_TaskDraft_list_view_shows_entry(self):
test_data_1 = TaskDraft_test_data("task draft")
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)
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 one")
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 one")
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 one")
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])
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.input_role_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(), '/task_connector_type/')
cls.output_role_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(), '/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, 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_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, output_role_url=self.output_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):
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=self.input_role_url, output_role_url=self.output_role_url)
trd_test_data2 = 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, output_role_url=self.output_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
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, output_role_url=self.output_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": '{"para": "meter"}'}
# 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, output_role_url=self.output_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, output_role_url=self.output_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, output_role_url=self.output_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, output_role_url=self.output_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):
input_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(), '/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, output_role_url=self.output_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, input_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_relation_draft_CASCADE_behavior_on_output_deleted(self):
output_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(), '/task_connector_type/')
trd_test_data = test_data_creator.TaskRelationDraft(output_role_url=output_url, producer_url=self.producer_url, consumer_url=self.consumer_url, template_url=self.template_url, input_role_url=self.input_role_url)
# POST new item with dependency
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/',
trd_test_data, 201, trd_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, trd_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, output_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_GET_TaskRelationDraft_list_view_shows_entry(self):
test_data_1 = TaskRelationDraft_test_data()
models.TaskRelationDraft.objects.create(**test_data_1)
nbr_results = models.TaskRelationDraft.objects.count()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/task_relation_draft/', test_data_1, nbr_results)
def test_GET_TaskRelationDraft_view_returns_correct_entry(self):
# setup
test_data_1 = TaskRelationDraft_test_data()
test_data_2 = TaskRelationDraft_test_data()
id1 = models.TaskRelationDraft.objects.create(**test_data_1).id
id2 = models.TaskRelationDraft.objects.create(**test_data_2).id
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_relation_draft/%s/' % id1, test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_relation_draft/%s/' % id2, test_data_2)
def test_nested_TaskRelationDraft_are_filtered_according_to_TaskDraft(self):
# setup
test_data_1 = TaskRelationDraft_test_data()
tdt_test_data_1 = TaskDraft_test_data()
task_draft_1 = models.TaskDraft.objects.create(**tdt_test_data_1)
test_data_1 = dict(test_data_1)
test_data_1['producer'] = task_draft_1
task_relation_draft_1 = models.TaskRelationDraft.objects.create(**test_data_1)
# assert the returned list contains related items, A list of length 1 is retrieved
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/task_draft/%s/task_relation_draft/' % task_draft_1.id, test_data_1, 1)
# assert an existing related producer is returned
class SchedulingUnitBlueprintTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.scheduling_unit_draft_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitDraft(), '/scheduling_unit_draft/')
cls.template_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitTemplate(), '/scheduling_unit_template/')
def test_scheduling_unit_blueprint_list_apiformat(self):
r = requests.get(BASE_URL + '/scheduling_unit_blueprint/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Scheduling Unit Blueprint List" in r.content.decode('utf8'))
def test_scheduling_unit_blueprint_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/scheduling_unit_blueprint/1234321/', 404)
def test_scheduling_unit_blueprint_POST_and_GET(self):
sub_test_data = test_data_creator.SchedulingUnitBlueprint(scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', sub_test_data, 201, sub_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, sub_test_data)
def test_scheduling_unit_blueprint_PUT_invalid_raises_error(self):
sub_test_data = test_data_creator.SchedulingUnitBlueprint(scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/9876789876/', sub_test_data, 404, {})
def test_scheduling_unit_blueprint_PUT(self):
sub_test_data1 = test_data_creator.SchedulingUnitBlueprint(name="the one", scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
sub_test_data2 = test_data_creator.SchedulingUnitBlueprint(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 + '/scheduling_unit_blueprint/', sub_test_data1, 201, sub_test_data1)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, sub_test_data1)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, sub_test_data2, 200, sub_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, sub_test_data2)
def test_scheduling_unit_blueprint_PATCH(self):
sub_test_data = test_data_creator.SchedulingUnitBlueprint(scheduling_unit_draft_url=self.scheduling_unit_draft_url, template_url=self.template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', sub_test_data, 201, sub_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, sub_test_data)
test_patch = {"description": "This is an updated description",
"do_cancel": True}
# 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_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/')
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)
# DELETE dependency
DELETE_and_assert_gone(self, scheduling_unit_draft_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_GET_SchedulingUnitBlueprint_list_view_shows_entry(self):
test_data_1 = SchedulingUnitBlueprint_test_data("scheduler unit blue print one")
models.SchedulingUnitBlueprint.objects.create(**test_data_1)
nbr_results = models.SchedulingUnitBlueprint.objects.count()
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/scheduling_unit_blueprint/', test_data_1, nbr_results)
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_nested_SchedulingUnitBlueprint_are_filtered_according_to_SchedulingUnitDraft(self):
# setup
test_data_1 = SchedulingUnitBlueprint_test_data("scheduler unit blue print three one")
sudt_test_data_1 = SchedulingUnitDraft_test_data()
scheduling_unit_draft_1 = models.SchedulingUnitDraft.objects.create(**sudt_test_data_1)
test_data_1 = dict(test_data_1)
test_data_1['draft'] = scheduling_unit_draft_1
scheduling_unit_blueprint_1 = models.SchedulingUnitBlueprint.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/scheduling_unit_blueprint/' % scheduling_unit_draft_1.id, test_data_1, 1)
class TaskBlueprintTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.draft_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskDraft(), '/task_draft/')
cls.template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
cls.scheduling_unit_blueprint_url = test_data_creator.post_data_and_get_url(test_data_creator.SchedulingUnitBlueprint(), '/scheduling_unit_blueprint/')
def test_task_blueprint_list_apiformat(self):
r = requests.get(BASE_URL + '/task_blueprint/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Blueprint List" in r.content.decode('utf8'))
def test_task_blueprint_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/1234321/', 404)
def test_task_blueprint_POST_and_GET(self):
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tb_test_data)
def test_task_blueprint_PUT_invalid_raises_error(self):
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
PUT_and_assert_expected_response(self, BASE_URL + '/task_blueprint/9876789876/', tb_test_data, 404, {})
def test_task_blueprint_PUT(self):
tb_test_data1 = test_data_creator.TaskBlueprint(name="the one", draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
tb_test_data2 = test_data_creator.TaskBlueprint(name="the other", draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data1, 201, tb_test_data1)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tb_test_data1)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, tb_test_data2, 200, tb_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, tb_test_data2)
def test_task_blueprint_PATCH(self):
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tb_test_data)
test_patch = {"description": "This is an updated description",
"do_cancel": True}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(tb_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_task_blueprint_DELETE(self):
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, tb_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_task_blueprint_prevents_missing_specification_template(self):
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# test data
test_data = dict(tb_test_data)
test_data['specifications_template'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['specifications_template']))
def test_task_blueprint_prevents_missing_draft(self):
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# test data
test_data = dict(tb_test_data)
test_data['draft'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['draft']))
def test_task_blueprint_prevents_missing_scheduling_unit_blueprint(self):
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=self.template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# test data
test_data = dict(tb_test_data)
test_data['scheduling_unit_blueprint'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['scheduling_unit_blueprint']))
def test_task_blueprint_CASCADE_behavior_on_task_template_deleted(self):
template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskTemplate(), '/task_template/')
tb_test_data = test_data_creator.TaskBlueprint(draft_url=self.draft_url, template_url=template_url, scheduling_unit_blueprint_url=self.scheduling_unit_blueprint_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)['url']
# verify
GET_OK_and_assert_equal_expected_response(self, url, tb_test_data)
# DELETE dependency
DELETE_and_assert_gone(self, template_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_blueprint_CASCADE_behavior_on_task_draft_deleted(self):
draft_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskDraft(), '/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)
# DELETE dependency
DELETE_and_assert_gone(self, draft_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
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()
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)
def test_GET_TaskBlueprint_view_returns_correct_entry(self):
# setup
test_data_1 = TaskBlueprint_test_data("task blue print two one")
test_data_2 = TaskBlueprint_test_data("task blue print two two")
id1 = models.TaskBlueprint.objects.create(**test_data_1).id
id2 = models.TaskBlueprint.objects.create(**test_data_2).id
# assert
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_blueprint/%s/' % id1, test_data_1)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/task_blueprint/%s/' % id2, test_data_2)
def test_nested_TaskBlueprint_are_filtered_according_to_TaskDraft(self):
# setup
test_data_1 = TaskBlueprint_test_data("task blue print three one")
tdt_test_data_1 = TaskDraft_test_data("task draft two one")
task_draft_1 = models.TaskDraft.objects.create(**tdt_test_data_1)
test_data_1 = dict(test_data_1)
test_data_1['draft'] = task_draft_1
task_blueprint_1 = models.TaskBlueprint.objects.create(**test_data_1)
# assert the returned list contains related items, A list of length 1 is retrieved
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/task_draft/%s/task_blueprint/' % task_draft_1.id, test_data_1, 1)
def test_TaskBlueprint_contains_list_of_related_Subtask(self):
# setup
test_data_1 = TaskBlueprint_test_data()
test_data_2 = TaskBlueprint_test_data()
st_test_data_1 = Subtask_test_data()
st_test_data_2 = Subtask_test_data()
task_blueprint = models.TaskBlueprint.objects.create(**test_data_1)
subtask_1 = models.Subtask.objects.create(**st_test_data_1)
subtask_1.task_blueprint = task_blueprint
subtask_1.save()
subtask_2 = models.Subtask.objects.create(**st_test_data_2)
subtask_2.task_blueprint = task_blueprint
subtask_2.save()
# assert
response_data = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/%s/' % task_blueprint.id, 200)
assertUrlList(self, response_data['subtasks'], [subtask_1, subtask_2])
def test_TaskBlueprint_contains_lists_of_related_TaskRelationBlueprint(self):
# setup
test_data_1 = TaskBlueprint_test_data()
test_data_2 = TaskBlueprint_test_data()
trbt_test_data_1 = TaskRelationBlueprint_test_data()
trbt_test_data_2 = TaskRelationBlueprint_test_data()
task_blueprint = models.TaskBlueprint.objects.create(**test_data_1)
task_relation_blueprint_1 = models.TaskRelationBlueprint.objects.create(**trbt_test_data_1)
task_relation_blueprint_1.producer = task_blueprint
task_relation_blueprint_1.save()
task_relation_blueprint_2 = models.TaskRelationBlueprint.objects.create(**trbt_test_data_2)
task_relation_blueprint_2.consumer = task_blueprint
task_relation_blueprint_2.save()
# assert
response_data = GET_and_assert_equal_expected_code(self, BASE_URL + '/task_blueprint/%s/' % task_blueprint.id, 200)
assertUrlList(self, response_data['produced_by'], [task_relation_blueprint_1])
assertUrlList(self, response_data['consumed_by'], [task_relation_blueprint_2])
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.input_role_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(), '/task_connector_type/')
cls.output_role_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(), '/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, template_url=self.template_url, input_role_url=self.input_role_url, output_role_url=self.output_role_url, consumer_url=self.consumer_url, producer_url=self.producer_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, output_role_url=self.output_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):
trb_test_data1 = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, output_role_url=self.output_role_url, consumer_url=self.consumer_url, producer_url=self.producer_url)
trb_test_data2 = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, output_role_url=self.output_role_url, consumer_url=self.consumer_url, producer_url=self.producer_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', trb_test_data1, 201, trb_test_data1)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, trb_test_data1)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, trb_test_data2, 200, trb_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, trb_test_data2)
def test_task_relation_blueprint_PATCH(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, output_role_url=self.output_role_url, consumer_url=self.consumer_url, producer_url=self.producer_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', trb_test_data, 201, trb_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, trb_test_data)
test_patch = {"selection_doc": '{"new": "doc"}'}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(trb_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_task_relation_blueprint_DELETE(self):
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, output_role_url=self.output_role_url, consumer_url=self.consumer_url, producer_url=self.producer_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, output_role_url=self.output_role_url, consumer_url=self.consumer_url, producer_url=self.producer_url)
# test data
test_data = dict(trb_test_data)
test_data['selection_template'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['selection_template']))
def test_task_relation_blueprint_prevents_missing_draft(self):
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, output_role_url=self.output_role_url, consumer_url=self.consumer_url, producer_url=self.producer_url)
# test data
test_data = dict(trb_test_data)
test_data['draft'] = None
# POST invalid data and assert response
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', test_data, 400, {})
self.assertTrue('This field may not be null' in str(r_dict['draft']))
def test_task_relation_blueprint_prevents_missing_producer(self):
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, output_role_url=self.output_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, output_role_url=self.output_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, output_role_url=self.output_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, output_role_url=self.output_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, input_role_url=self.input_role_url, output_role_url=self.output_role_url, consumer_url=self.consumer_url, producer_url=self.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, 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, output_role_url=self.output_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, output_role_url=self.output_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):
input_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(), '/task_connector_type/')
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=input_url, output_role_url=self.output_role_url, consumer_url=self.consumer_url, producer_url=self.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, input_url)
# assert
GET_and_assert_equal_expected_code(self, url, 404)
def test_task_relation_blueprint_CASCADE_behavior_on_output_deleted(self):
output_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskConnectorType(), '/task_connector_type/')
trb_test_data = test_data_creator.TaskRelationBlueprint(draft_url=self.draft_url, template_url=self.template_url, input_role_url=self.input_role_url, output_role_url=output_url, consumer_url=self.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, 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):
# setup
test_data_1 = TaskRelationBlueprint_test_data()
test_data_2 = TaskRelationBlueprint_test_data()
tbt_test_data_1 = TaskBlueprint_test_data()
tbt_test_data_2 = TaskBlueprint_test_data()
task_blueprint_1 = models.TaskBlueprint.objects.create(**tbt_test_data_1)
task_blueprint_2 = models.TaskBlueprint.objects.create(**tbt_test_data_2)
test_data_1 = dict(test_data_1)
test_data_1['producer'] = task_blueprint_1
task_relation_blueprint_1 = models.TaskRelationBlueprint.objects.create(**test_data_1)
test_data_2 = dict(test_data_2)
test_data_2['consumer'] = task_blueprint_2
task_relation_blueprint_2 = models.TaskRelationBlueprint.objects.create(**test_data_2)
# assert the returned list contains related producer
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/task_blueprint/%s/task_relation_blueprint/' % task_blueprint_1.id, test_data_1, 1)
# assert the returned list contains related consumer
GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/task_blueprint/%s/task_relation_blueprint/' % task_blueprint_2.id, test_data_2, 1)
class TaskSchedulingRelationBlueprintTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.first_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskBlueprint(), '/task_blueprint/')
cls.second_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskBlueprint(), '/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=None, second_url=None, 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_PUT_invalid_raises_error(self):
tsrb_test_data = test_data_creator.TaskSchedulingRelationBlueprint(first_url=None, second_url=None, 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=None, second_url=None, placement="after")
tsrb_test_data2 = test_data_creator.TaskSchedulingRelationBlueprint(first_url=None, second_url=None, 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=None, second_url=None, 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=None, second_url=None, 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=None, second_url=None, 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=None, second_url=None, 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=None, second_url=None, 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=None, second_url=None, 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
tsrb_test_data = test_data_creator.TaskSchedulingRelationBlueprint(first_url=None, second_url=None, 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)
#Get the URL of first task blueprint
test_data = dict(tsrb_test_data)
task_blueprint_url=test_data['first']
# DELETE dependency
DELETE_and_assert_gone(self, 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.first_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskDraft(), '/task_draft/')
cls.second_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskDraft(), '/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=None, second_url=None, 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=None, second_url=None, 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=None, second_url=None, placement="after")
tsrd_test_data2 = test_data_creator.TaskSchedulingRelationDraft( first_url=None, second_url=None, 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=None, second_url=None, 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=None, second_url=None, 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=None, second_url=None, 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=None, second_url=None, 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=None, second_url=None, 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=None, second_url=None, 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)
if __name__ == "__main__":
logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s',
level=logging.INFO)
unittest.main()