-
Jorrit Schaap authoredTMSS-573: Seperated TMSS into sub-packages TMSSBackend, TMSSFrontend, TMSSClient and TMSSServices and used the proper dependencies on using-lofar-packages like QA and SCU where TMSSFrontend is not needed.
Jorrit Schaap authoredTMSS-573: Seperated TMSS into sub-packages TMSSBackend, TMSSFrontend, TMSSClient and TMSSServices and used the proper dependencies on using-lofar-packages like QA and SCU where TMSSFrontend is not needed.
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
t_tmssapp_scheduling_REST_API.py 89.09 KiB
#!/usr/bin/env python3
# Copyright (C) 2018 ASTRON (Netherlands Institute for Radio Astronomy)
# P.O. Box 2, 7990 AA Dwingeloo, The Netherlands
#
# This file is part of the LOFAR software suite.
# The LOFAR software suite is free software: you can redistribute it and/or
# modify it under the terms of the GNU General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# The LOFAR software suite is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.
# $Id: $
# This functional test talks to the API like a regular user would.
# It is supposed to cover all REST http methods for all ViewSets.
# I am still a bit under the impression that we re-test Django functionality that we can expect to just work
# with some of these tests. On the other hand a lot of these provide us a nice basis for differentiating out
# behavior in a controlled way.
# We should probably also fully test behavior wrt mandatory and nullable fields.
import 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 exit_with_skipped_code_if_skip_integration_tests
exit_with_skipped_code_if_skip_integration_tests()
# Do Mandatory setup step:
# use setup/teardown magic for tmss test database, ldap server and django server
# (ignore pycharm unused import statement, python unittests does use at RunTime the tmss_test_environment_unittest_setup module)
from lofar.sas.tmss.test.tmss_test_environment_unittest_setup import *
from lofar.sas.tmss.tmss.tmssapp import models
from lofar.common.datetimeutils import formatDatetime
# import and setup test data creator
from lofar.sas.tmss.test.tmss_test_data_rest import TMSSRESTTestDataCreator
test_data_creator = TMSSRESTTestDataCreator(BASE_URL, AUTH)
from lofar.sas.tmss.test.test_utils import minimal_json_schema
from datetime import datetime, timedelta
class SubtaskTemplateTestCase(unittest.TestCase):
def test_subtask_template_list_apiformat(self):
r = requests.get(BASE_URL + '/subtask_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Subtask Template List" in r.content.decode('utf8'))
def test_subtask_template_template_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask_template/1234321/', 404)
def test_subtask_template_POST_and_GET(self):
st_test_data = test_data_creator.SubtaskTemplate()
expected_data = test_data_creator.update_schema_from_template("subtasktemplate", st_test_data)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/', st_test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_subtask_template_PUT_invalid_raises_error(self):
st_test_data = test_data_creator.SubtaskTemplate()
PUT_and_assert_expected_response(self, BASE_URL + '/subtask_template/9876789876/', st_test_data, 404, {})
def test_subtask_template_PUT(self):
st_test_data = test_data_creator.SubtaskTemplate(name="the_one")
st_test_data2 = test_data_creator.SubtaskTemplate(name="the_other")
expected_data = test_data_creator.update_schema_from_template("subtasktemplate", st_test_data)
expected_data2 = test_data_creator.update_schema_from_template("subtasktemplate", st_test_data2)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/', st_test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, st_test_data2, 200, expected_data2)
GET_OK_and_assert_equal_expected_response(self, url, expected_data2)
def test_subtask_template_PATCH(self):
st_test_data = test_data_creator.SubtaskTemplate()
expected_data = test_data_creator.update_schema_from_template("subtasktemplate", st_test_data)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/', st_test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
test_patch = {"name": "new_name",
"description": "better description",
"type": BASE_URL + '/subtask_type/inspection',
"schema": minimal_json_schema(properties={"mykey": {"type":"string", "default":"my better value"}})
}
# PATCH item and verify
expected_patch_data = test_data_creator.update_schema_from_template("subtasktemplate", test_patch)
PATCH_and_assert_expected_response(self, url, test_patch, 200, expected_patch_data)
expected_data = dict(st_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_patch_data)
def test_subtask_template_DELETE(self):
st_test_data = test_data_creator.SubtaskTemplate()
expected_data = test_data_creator.update_schema_from_template("subtasktemplate", st_test_data)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/', st_test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_subtask_template_PROTECT_behavior_on_type_choice_deleted(self):
st_test_data = test_data_creator.SubtaskTemplate()
# create dependency that is safe to delete (enums are not populated / re-established between tests)
type_data = {'value': 'kickme'}
POST_and_assert_expected_response(self, BASE_URL + '/subtask_type/', type_data, 201, type_data)
type_url = BASE_URL + '/subtask_type/kickme'
# POST new item and verify
test_data = dict(st_test_data)
test_data['type'] = type_url
expected_data = test_data_creator.update_schema_from_template("subtasktemplate", test_data)
url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/', test_data, 201, expected_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(type_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, type_url, type_data)
class DataproductSpecificationsTemplateTestCase(unittest.TestCase):
def test_dataproduct_specifications_template_list_apiformat(self):
r = requests.get(BASE_URL + '/dataproduct_specifications_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Dataproduct Specifications Template List" in r.content.decode('utf8'))
def test_dataproduct_specifications_template_template_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/dataproduct_specifications_template/1234321/', 404)
def test_dataproduct_specifications_template_POST_and_GET(self):
dst_test_data = test_data_creator.DataproductSpecificationsTemplate()
expected_data = test_data_creator.update_schema_from_template("dataproductspecificationstemplate", dst_test_data)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/', dst_test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_dataproduct_specifications_template_PUT_invalid_raises_error(self):
dst_test_data = test_data_creator.DataproductSpecificationsTemplate()
PUT_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/9876789876/', dst_test_data, 404, {})
def test_dataproduct_specifications_template_PUT(self):
dst_test_data = test_data_creator.DataproductSpecificationsTemplate(name="the_one")
dst_test_data2 = test_data_creator.DataproductSpecificationsTemplate(name="the_other")
expected_data = test_data_creator.update_schema_from_template("dataproductspecificationstemplate", dst_test_data)
expected_data2 = test_data_creator.update_schema_from_template("dataproductspecificationstemplate", dst_test_data2)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/', dst_test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, dst_test_data2, 200, expected_data2)
GET_OK_and_assert_equal_expected_response(self, url, expected_data2)
def test_dataproduct_specifications_template_PATCH(self):
dst_test_data = test_data_creator.DataproductSpecificationsTemplate()
expected_data = test_data_creator.update_schema_from_template("dataproductspecificationstemplate", dst_test_data)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/', dst_test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
test_patch = {"name": "new_name",
"description": "better description",
"schema": minimal_json_schema(properties={"mykey": {"type":"string", "default":"my better value"}})
}
# PATCH item and verify
expected_patch_data = test_data_creator.update_schema_from_template("dataproductspecificationstemplate", test_patch)
PATCH_and_assert_expected_response(self, url, test_patch, 200, expected_patch_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_patch_data)
def test_dataproduct_specifications_template_DELETE(self):
dst_test_data = test_data_creator.DataproductSpecificationsTemplate()
expected_data = test_data_creator.update_schema_from_template("dataproductspecificationstemplate", dst_test_data)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/', dst_test_data, 201, expected_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
class DefaultSubtaskTemplatesTestCase(unittest.TestCase):
def test_default_subtask_template_POST(self):
template_url = test_data_creator.post_data_and_get_url(test_data_creator.SubtaskTemplate(), '/subtask_template/')
dst_test_data = test_data_creator.DefaultSubtaskTemplates(template_url=template_url)
POST_and_assert_expected_response(self, BASE_URL + '/default_subtask_template/', dst_test_data, 201, dst_test_data)
def test_default_dataproduct_specifications_template_POST(self):
template_url = test_data_creator.post_data_and_get_url(test_data_creator.DataproductSpecificationsTemplate(), '/dataproduct_specifications_template/')
dst_test_data = test_data_creator.DefaultSubtaskTemplates(template_url=template_url)
POST_and_assert_expected_response(self, BASE_URL + '/default_dataproduct_specifications_template/', dst_test_data, 201, dst_test_data)
def test_default_subtask_template_PROTECT_behavior_on_template_deleted(self):
st_test_data = test_data_creator.SubtaskTemplate()
expected_data = test_data_creator.update_schema_from_template("subtasktemplate", st_test_data)
template_url = test_data_creator.post_data_and_get_url(st_test_data, '/subtask_template/')
dst_test_data = test_data_creator.DefaultSubtaskTemplates(template_url=template_url)
# POST with dependency
POST_and_assert_expected_response(self, BASE_URL + '/default_subtask_template/', dst_test_data, 201, dst_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, template_url, expected_data)
def test_default_dataproduct_specifications_template_PROTECT_behavior_on_template_deleted(self):
dpst_test_data = test_data_creator.DataproductSpecificationsTemplate()
expected_data = test_data_creator.update_schema_from_template("dataproductspecificationstemplate", dpst_test_data)
template_url = test_data_creator.post_data_and_get_url(dpst_test_data, '/dataproduct_specifications_template/')
dst_test_data = test_data_creator.DefaultSubtaskTemplates(template_url=template_url)
# POST with dependency
POST_and_assert_expected_response(self, BASE_URL + '/default_dataproduct_specifications_template/', dst_test_data, 201, dst_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, template_url, expected_data)
class SubtaskTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
# we should not depend on "previous" data
models.SubtaskInput.objects.all().delete()
models.DataproductHash.objects.all().delete()
models.DataproductArchiveInfo.objects.all().delete()
models.DataproductTransform.objects.all().delete()
models.Dataproduct.objects.all().delete()
models.Subtask.objects.all().delete()
cls.cluster_url = test_data_creator.post_data_and_get_url(test_data_creator.Cluster(), '/cluster/')
cls.task_blueprint_data = test_data_creator.TaskBlueprint()
cls.task_blueprint_url = test_data_creator.post_data_and_get_url(cls.task_blueprint_data, '/task_blueprint/')
cls.specifications_template_url = test_data_creator.post_data_and_get_url(test_data_creator.SubtaskTemplate(), '/subtask_template/')
def test_subtask_list_apiformat(self):
r = requests.get(BASE_URL + '/subtask/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Subtask List" in r.content.decode('utf8'))
def test_subtask_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/1234321/', 404)
def test_subtask_POST_and_GET(self):
st_test_data = test_data_creator.Subtask(cluster_url=self.cluster_url, task_blueprint_url=self.task_blueprint_url, specifications_template_url=self.specifications_template_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', st_test_data, 201, st_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, st_test_data)
minimium_subtaskid = 2000000
subtask_id = url.split("subtask/")[1].replace("/","")
self.assertGreaterEqual(int(subtask_id), minimium_subtaskid)
subtask_id = r_dict['id']
self.assertGreaterEqual(int(subtask_id), minimium_subtaskid)
def test_subtask_PUT_invalid_raises_error(self):
st_test_data = test_data_creator.Subtask(cluster_url=self.cluster_url, task_blueprint_url=self.task_blueprint_url, specifications_template_url=self.specifications_template_url)
PUT_and_assert_expected_response(self, BASE_URL + '/subtask/9876789876/', st_test_data, 404, {})
def test_subtask_PUT(self):
st_test_data = test_data_creator.Subtask(cluster_url=self.cluster_url, task_blueprint_url=self.task_blueprint_url, specifications_template_url=self.specifications_template_url)
st_test_data2 = test_data_creator.Subtask(cluster_url=self.cluster_url, task_blueprint_url=self.task_blueprint_url, specifications_template_url=self.specifications_template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', st_test_data, 201, st_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, st_test_data)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, st_test_data2, 200, st_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, st_test_data2)
def test_subtask_PATCH(self):
st_test_data = test_data_creator.Subtask(cluster_url=self.cluster_url, task_blueprint_url=self.task_blueprint_url, specifications_template_url=self.specifications_template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', st_test_data, 201, st_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, st_test_data)
test_patch = {"specifications_doc": {"somespec": "somevalue"}}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(st_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_subtask_DELETE(self):
st_test_data = test_data_creator.Subtask(cluster_url=self.cluster_url, task_blueprint_url=self.task_blueprint_url, specifications_template_url=self.specifications_template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', st_test_data, 201, st_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, st_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_subtask_PROTECT_behavior_on_state_choice_deleted(self):
st_test_data = test_data_creator.Subtask(cluster_url=self.cluster_url, task_blueprint_url=self.task_blueprint_url, specifications_template_url=self.specifications_template_url)
# create dependency that is safe to delete (enums are not populated / re-established between tests)
state_data = {'value': 'kickme'}
POST_and_assert_expected_response(self, BASE_URL + '/subtask_state/', state_data, 201, state_data)
state_url = BASE_URL + '/subtask_state/kickme'
# POST new item and verify
test_data = dict(st_test_data)
test_data['state'] = state_url
url = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', test_data, 201, test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(state_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, state_url, state_data)
def test_subtask_SET_NULL_behavior_on_task_blueprint_deleted(self):
# make new task_blueprint_url instance, but reuse related data for speed
tbp_test_data = test_data_creator.TaskBlueprint(draft_url=self.task_blueprint_data['draft'],
template_url=self.task_blueprint_data['specifications_template'],
scheduling_unit_blueprint_url=self.task_blueprint_data['scheduling_unit_blueprint'])
task_blueprint_url = test_data_creator.post_data_and_get_url(tbp_test_data, '/task_blueprint/')
st_test_data = test_data_creator.Subtask(task_blueprint_url=task_blueprint_url, cluster_url=self.cluster_url, specifications_template_url=self.specifications_template_url)
# POST new item and verify
url = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', st_test_data, 201, st_test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, st_test_data)
# DELETE dependency and check it's gone
DELETE_and_assert_gone(self, task_blueprint_url)
# assert item reference is set null
expected_data = dict(st_test_data)
expected_data['task_blueprint'] = None
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_subtask_PROTECT_behavior_on_template_deleted(self):
stt_test_data = test_data_creator.SubtaskTemplate()
expected_data = test_data_creator.update_schema_from_template("subtasktemplate", stt_test_data)
specifications_template_url = test_data_creator.post_data_and_get_url(stt_test_data, '/subtask_template/')
st_test_data = test_data_creator.Subtask(specifications_template_url=specifications_template_url, cluster_url=self.cluster_url, task_blueprint_url=self.task_blueprint_url)
# POST new item and verify
url = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', st_test_data, 201, st_test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, st_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(specifications_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, specifications_template_url, expected_data)
def test_subtask_state_log_records(self):
st_test_data = test_data_creator.Subtask()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', st_test_data, 201, st_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, st_test_data)
# Verify state log count is 1
segments = url.split('/')
identifier = ''
while identifier == '':
identifier = segments.pop()
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/subtask_state_log/?subtask=' + identifier, {"count":1})
# PATCH item with something else than state and verify no log record is created
test_patch = {"specifications_doc": {"somespec": "somevalue"}}
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/subtask_state_log/?subtask=' + identifier, {"count": 1})
# PATCH item with state update and verify log record is created
test_patch = {"state": BASE_URL + "/subtask_state/finishing"}
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
GET_OK_and_assert_equal_expected_response(self, BASE_URL + '/subtask_state_log/?subtask=' + identifier, {"count": 2})
class DataproductTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.specifications_template_url = test_data_creator.post_data_and_get_url(test_data_creator.SubtaskTemplate(), '/dataproduct_specifications_template/')
cls.subtask_output_url = test_data_creator.post_data_and_get_url(test_data_creator.SubtaskOutput(), '/subtask_output/')
cls.dataproduct_feedback_template_url = test_data_creator.post_data_and_get_url(test_data_creator.DataproductFeedbackTemplate(), '/dataproduct_feedback_template/')
def test_dataproduct_list_apiformat(self):
r = requests.get(BASE_URL + '/dataproduct/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Dataproduct List" in r.content.decode('utf8'))
def test_dataproduct_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/dataproduct/1234321/', 404)
def test_dataproduct_POST_and_GET(self):
dp_test_data = test_data_creator.Dataproduct(specifications_template_url=self.specifications_template_url, subtask_output_url=self.subtask_output_url, dataproduct_feedback_template_url=self.dataproduct_feedback_template_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', dp_test_data, 201, dp_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dp_test_data)
def test_dataproduct_PUT_invalid_raises_error(self):
dp_test_data = test_data_creator.Dataproduct(specifications_template_url=self.specifications_template_url, subtask_output_url=self.subtask_output_url, dataproduct_feedback_template_url=self.dataproduct_feedback_template_url)
PUT_and_assert_expected_response(self, BASE_URL + '/dataproduct/9876789876/', dp_test_data, 404, {})
def test_dataproduct_PUT(self):
dp_test_data = test_data_creator.Dataproduct(specifications_template_url=self.specifications_template_url, subtask_output_url=self.subtask_output_url, dataproduct_feedback_template_url=self.dataproduct_feedback_template_url)
dp_test_data2 = test_data_creator.Dataproduct(specifications_template_url=self.specifications_template_url, subtask_output_url=self.subtask_output_url, dataproduct_feedback_template_url=self.dataproduct_feedback_template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', dp_test_data, 201, dp_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dp_test_data)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, dp_test_data2, 200, dp_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, dp_test_data2)
def test_dataproduct_PATCH(self):
dp_test_data = test_data_creator.Dataproduct(specifications_template_url=self.specifications_template_url, subtask_output_url=self.subtask_output_url, dataproduct_feedback_template_url=self.dataproduct_feedback_template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', dp_test_data, 201, dp_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dp_test_data)
test_patch = {"filename": 'my_better.filename',
"deleted_since": datetime.utcnow().isoformat()}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(dp_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_dataproduct_DELETE(self):
dp_test_data = test_data_creator.Dataproduct(specifications_template_url=self.specifications_template_url, subtask_output_url=self.subtask_output_url, dataproduct_feedback_template_url=self.dataproduct_feedback_template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', dp_test_data, 201, dp_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dp_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_dataproduct_PROTECT_behavior_on_dataformat_deleted(self):
dp_test_data = test_data_creator.Dataproduct(specifications_template_url=self.specifications_template_url, subtask_output_url=self.subtask_output_url, dataproduct_feedback_template_url=self.dataproduct_feedback_template_url)
# create dependency that is safe to delete (enums are not populated / re-established between tests)
dataformat_data = {'value': 'kickme'}
POST_and_assert_expected_response(self, BASE_URL + '/dataformat/', dataformat_data, 201, dataformat_data)
dataformat_url = BASE_URL + '/dataformat/kickme'
# POST new item and verify
test_data = dict(dp_test_data)
test_data['dataformat'] = dataformat_url
url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', test_data, 201, test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(dataformat_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, dataformat_url, dataformat_data)
def test_dataproduct_CASCADE_behavior_on_specifications_template_deleted(self):
specifications_template_url = test_data_creator.post_data_and_get_url(test_data_creator.SubtaskTemplate(), '/dataproduct_specifications_template/')
dp_test_data = test_data_creator.Dataproduct(specifications_template_url=specifications_template_url, subtask_output_url=self.subtask_output_url, dataproduct_feedback_template_url=self.dataproduct_feedback_template_url)
# POST new item, verify
url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', dp_test_data, 201, dp_test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, dp_test_data)
# DELETE dependency and check it's gone
DELETE_and_assert_gone(self, specifications_template_url)
# assert item gone
GET_and_assert_equal_expected_code(self, url, 404)
class SubtaskInputTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.subtask_data = test_data_creator.Subtask()
cls.subtask_url = test_data_creator.post_data_and_get_url(cls.subtask_data, '/subtask/')
cls.task_relation_blueprint_data = test_data_creator.TaskRelationBlueprint()
cls.task_relation_blueprint_url = test_data_creator.post_data_and_get_url(cls.task_relation_blueprint_data, '/task_relation_blueprint/')
cls.dataproduct_urls = [test_data_creator.post_data_and_get_url(test_data_creator.Dataproduct(), '/dataproduct/'), test_data_creator.post_data_and_get_url(test_data_creator.Dataproduct(), '/dataproduct/')]
cls.subtask_output_data = test_data_creator.SubtaskOutput()
cls.subtask_output_url = test_data_creator.post_data_and_get_url(cls.subtask_output_data, '/subtask_output/')
cls.task_relation_selection_template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskRelationSelectionTemplate(), '/task_relation_selection_template/')
def test_subtask_input_list_apiformat(self):
r = requests.get(BASE_URL + '/subtask_input/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Subtask Input List" in r.content.decode('utf8'))
def test_subtask_input_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask_input/1234321/', 404)
def test_subtask_input_POST_and_GET(self):
sti_test_data = test_data_creator.SubtaskInput(subtask_url=self.subtask_url, task_relation_blueprint_url=self.task_relation_blueprint_url, dataproduct_urls=self.dataproduct_urls, subtask_output_url=self.subtask_output_url, task_relation_selection_template_url=self.task_relation_selection_template_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, sti_test_data)
def test_subtask_input_PUT_invalid_raises_error(self):
sti_test_data = test_data_creator.SubtaskInput(subtask_url=self.subtask_url, task_relation_blueprint_url=self.task_relation_blueprint_url, dataproduct_urls=self.dataproduct_urls, subtask_output_url=self.subtask_output_url, task_relation_selection_template_url=self.task_relation_selection_template_url)
PUT_and_assert_expected_response(self, BASE_URL + '/subtask_input/9876789876/', sti_test_data, 404, {})
def test_subtask_input_PUT(self):
sti_test_data = test_data_creator.SubtaskInput(subtask_url=self.subtask_url, task_relation_blueprint_url=self.task_relation_blueprint_url, dataproduct_urls=self.dataproduct_urls, subtask_output_url=self.subtask_output_url, task_relation_selection_template_url=self.task_relation_selection_template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, sti_test_data)
# PUT new values, verify
sti_test_data2 = test_data_creator.SubtaskInput(subtask_url=self.subtask_url, task_relation_blueprint_url=self.task_relation_blueprint_url, dataproduct_urls=self.dataproduct_urls, subtask_output_url=self.subtask_output_url, task_relation_selection_template_url=self.task_relation_selection_template_url)
PUT_and_assert_expected_response(self, url, sti_test_data2, 200, sti_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, sti_test_data2)
def test_subtask_input_PATCH(self):
sti_test_data = test_data_creator.SubtaskInput(subtask_url=self.subtask_url, task_relation_blueprint_url=self.task_relation_blueprint_url, dataproduct_urls=self.dataproduct_urls, subtask_output_url=self.subtask_output_url, task_relation_selection_template_url=self.task_relation_selection_template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, sti_test_data)
# make new subtask_url instance, but reuse related data for speed
subtask_url = test_data_creator.post_data_and_get_url(test_data_creator.Subtask(cluster_url=self.subtask_data['cluster'],
task_blueprint_url=self.subtask_data['task_blueprint'],
specifications_template_url=self.subtask_data['specifications_template'],
specifications_doc=self.subtask_data['specifications_doc']), '/subtask/')
test_patch = {"subtask": subtask_url,
"tags": ['FANCYTAG'],
}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(sti_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_subtask_input_DELETE(self):
sti_test_data = test_data_creator.SubtaskInput(subtask_url=self.subtask_url, task_relation_blueprint_url=self.task_relation_blueprint_url, dataproduct_urls=self.dataproduct_urls, subtask_output_url=self.subtask_output_url, task_relation_selection_template_url=self.task_relation_selection_template_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, sti_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_subtask_input_CASCADE_behavior_on_subtask_deleted(self):
# make new subtask_url instance, but reuse related data for speed
subtask_url = test_data_creator.post_data_and_get_url(test_data_creator.Subtask(cluster_url=self.subtask_data['cluster'],
task_blueprint_url=self.subtask_data['task_blueprint'],
specifications_template_url=self.subtask_data['specifications_template'],
specifications_doc=self.subtask_data['specifications_doc']), '/subtask/')
sti_test_data = test_data_creator.SubtaskInput(subtask_url=subtask_url, task_relation_blueprint_url=self.task_relation_blueprint_url, dataproduct_urls=self.dataproduct_urls, subtask_output_url=self.subtask_output_url, task_relation_selection_template_url=self.task_relation_selection_template_url)
# POST new item, verify
url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, sti_test_data)
# DELETE dependency and check it's gone
DELETE_and_assert_gone(self, subtask_url)
# assert item gone
GET_and_assert_equal_expected_code(self, url, 404)
def test_subtask_input_SET_NULL_behavior_on_task_relation_blueprint_deleted(self):
# make new task_relation_blueprint instance, but reuse related data for speed
task_relation_blueprint_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskRelationBlueprint(draft_url=self.task_relation_blueprint_data['draft'], template_url=self.task_relation_blueprint_data['selection_template'],
input_role_url=self.task_relation_blueprint_data['input_role'], output_role_url=self.task_relation_blueprint_data['output_role'],
consumer_url=self.task_relation_blueprint_data['consumer']), '/task_relation_blueprint/')
sti_test_data = test_data_creator.SubtaskInput(task_relation_blueprint_url=task_relation_blueprint_url, subtask_url=self.subtask_url, dataproduct_urls=self.dataproduct_urls, subtask_output_url=self.subtask_output_url, task_relation_selection_template_url=self.task_relation_selection_template_url)
# POST new item, verify
url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, sti_test_data)
# DELETE dependency and check it's gone
DELETE_and_assert_gone(self, task_relation_blueprint_url)
# assert item reference is set null
expected_data = dict(sti_test_data)
expected_data['task_relation_blueprint'] = None
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_subtask_input_PROTECT_behavior_on_producer_deleted(self):
# make new subtask_output_url instance, but reuse related data for speed
subtask_output_url = test_data_creator.post_data_and_get_url(test_data_creator.SubtaskOutput(subtask_url=self.subtask_output_data['subtask']), '/subtask_output/')
sti_test_data = test_data_creator.SubtaskInput(subtask_output_url=subtask_output_url, subtask_url=self.subtask_url, task_relation_blueprint_url=self.task_relation_blueprint_url, dataproduct_urls=self.dataproduct_urls, task_relation_selection_template_url=self.task_relation_selection_template_url)
# POST with dependency
url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, sti_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(subtask_output_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_and_assert_equal_expected_code(self, subtask_output_url, 200)
def test_subtask_input_PROTECT_behavior_on_selection_template_deleted(self):
task_relation_selection_template_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskRelationSelectionTemplate(), '/task_relation_selection_template/')
sti_test_data = test_data_creator.SubtaskInput(task_relation_selection_template_url=task_relation_selection_template_url,
subtask_url=self.subtask_url,
task_relation_blueprint_url=self.task_relation_blueprint_url,
dataproduct_urls=self.dataproduct_urls,
subtask_output_url=self.subtask_output_url)
# POST with dependency
url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, sti_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(task_relation_selection_template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_and_assert_equal_expected_code(self, task_relation_selection_template_url, 200)
class SubtaskOutputTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.subtask_data = test_data_creator.Subtask()
cls.subtask_url = test_data_creator.post_data_and_get_url(cls.subtask_data, '/subtask/')
def test_subtask_output_list_apiformat(self):
r = requests.get(BASE_URL + '/subtask_output/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Subtask Output List" in r.content.decode('utf8'))
def test_subtask_output_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask_output/1234321/', 404)
def test_subtask_output_POST_and_GET(self):
sto_test_data = test_data_creator.SubtaskOutput(subtask_url=self.subtask_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', sto_test_data, 201,
sto_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, sto_test_data)
def test_subtask_output_PUT_invalid_raises_error(self):
sto_test_data = test_data_creator.SubtaskOutput(subtask_url=self.subtask_url)
PUT_and_assert_expected_response(self, BASE_URL + '/subtask_output/9876789876/', sto_test_data, 404, {})
def test_subtask_output_PUT(self):
sto_test_data = test_data_creator.SubtaskOutput(subtask_url=self.subtask_url)
sto_test_data2 = test_data_creator.SubtaskOutput(subtask_url=self.subtask_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', sto_test_data, 201,sto_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, sto_test_data)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, sto_test_data2, 200, sto_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, sto_test_data2)
def test_subtask_output_PATCH(self):
sto_test_data = test_data_creator.SubtaskOutput(subtask_url=self.subtask_url)
sto_test_data2 = test_data_creator.SubtaskOutput(subtask_url=self.subtask_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', sto_test_data, 201,
sto_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, sto_test_data)
test_patch = {"subtask": sto_test_data2["subtask"],
"tags": ['FANCYTAG'], }
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(sto_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_subtask_output_DELETE(self):
sto_test_data = test_data_creator.SubtaskOutput(subtask_url=self.subtask_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', sto_test_data, 201,
sto_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, sto_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_subtask_output_CASCADE_behavior_on_subtask_deleted(self):
# make new subtask_url instance, but reuse related data for speed
subtask_url = test_data_creator.post_data_and_get_url(self.subtask_data, '/subtask/')
sto_test_data = test_data_creator.SubtaskOutput(subtask_url=subtask_url)
# POST new item, verify
url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', sto_test_data, 201, sto_test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, sto_test_data)
# DELETE dependency and check it's gone
DELETE_and_assert_gone(self, subtask_url)
# assert item gone
GET_and_assert_equal_expected_code(self, url, 404)
class AntennaSetTestCase(unittest.TestCase):
def test_antenna_set_list_apiformat(self):
r = requests.get(BASE_URL + '/antenna_set/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Antenna Set List" in r.content.decode('utf8'))
def test_antenna_set_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/antenna_set/1234321/', 404)
def test_antenna_set_POST_and_GET(self):
antennaset_test_data = test_data_creator.AntennaSet()
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/antenna_set/', antennaset_test_data, 201, antennaset_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, antennaset_test_data)
def test_antenna_set_PUT_invalid_raises_error(self):
antennaset_test_data = test_data_creator.AntennaSet()
PUT_and_assert_expected_response(self, BASE_URL + '/antenna_set/9876789876/', antennaset_test_data, 404, {})
def test_antenna_set_PUT(self):
antennaset_test_data = test_data_creator.AntennaSet(name="the_one")
antennaset_test_data2 = test_data_creator.AntennaSet(name="the_other")
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/antenna_set/', antennaset_test_data, 201, antennaset_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, antennaset_test_data)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, antennaset_test_data2, 200, antennaset_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, antennaset_test_data2)
def test_antenna_set_PATCH(self):
antennaset_test_data = test_data_creator.AntennaSet()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/antenna_set/', antennaset_test_data, 201, antennaset_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, antennaset_test_data)
test_patch = {"rcus": [11, 12, 13, 14, 15],
"station_type": BASE_URL + '/station_type/remote'}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(antennaset_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_antenna_set_DELETE(self):
antennaset_test_data = test_data_creator.AntennaSet()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/antenna_set/', antennaset_test_data, 201, antennaset_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, antennaset_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_antenna_set_PROTECT_behavior_on_station_type_deleted(self):
antennaset_test_data = test_data_creator.AntennaSet()
# create dependency that is safe to delete (enums are not populated / re-established between tests)
dataformat_data = {'value': 'kickme'}
POST_and_assert_expected_response(self, BASE_URL + '/station_type/', dataformat_data, 201, dataformat_data)
dataformat_url = BASE_URL + '/station_type/kickme'
# POST new item and verify
test_data = dict(antennaset_test_data)
test_data['station_type'] = dataformat_url
url = POST_and_assert_expected_response(self, BASE_URL + '/antenna_set/', test_data, 201, test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(dataformat_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, dataformat_url, dataformat_data)
class DataproductTransformTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.input_dataproduct_data = test_data_creator.Dataproduct()
cls.input_dataproduct_url = test_data_creator.post_data_and_get_url(cls.input_dataproduct_data, '/dataproduct/')
cls.output_dataproduct_data = test_data_creator.Dataproduct()
cls.output_dataproduct_url = test_data_creator.post_data_and_get_url(cls.output_dataproduct_data, '/dataproduct/')
def test_dataproduct_transform_list_apiformat(self):
r = requests.get(BASE_URL + '/dataproduct_transform/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Dataproduct Transform List" in r.content.decode('utf8'))
def test_dataproduct_transform_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/dataproduct_transform/1234321/', 404)
def test_dataproduct_transform_POST_and_GET(self):
dpt_test_data = test_data_creator.DataproductTransform(input_dataproduct_url=self.input_dataproduct_url, output_dataproduct_url=self.output_dataproduct_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', dpt_test_data, 201, dpt_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dpt_test_data)
def test_dataproduct_transform_PUT_invalid_raises_error(self):
dpt_test_data = test_data_creator.DataproductTransform(input_dataproduct_url=self.input_dataproduct_url, output_dataproduct_url=self.output_dataproduct_url)
PUT_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/9876789876/', dpt_test_data, 404, {})
def test_dataproduct_transform_PUT(self):
dpt_test_data = test_data_creator.DataproductTransform(input_dataproduct_url=self.input_dataproduct_url, output_dataproduct_url=self.output_dataproduct_url)
dpt_test_data2 = test_data_creator.DataproductTransform(input_dataproduct_url=self.input_dataproduct_url, output_dataproduct_url=self.output_dataproduct_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', dpt_test_data, 201, dpt_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dpt_test_data)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, dpt_test_data2, 200, dpt_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, dpt_test_data2)
def test_dataproduct_transform_PATCH(self):
dpt_test_data = test_data_creator.DataproductTransform(input_dataproduct_url=self.input_dataproduct_url, output_dataproduct_url=self.output_dataproduct_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', dpt_test_data, 201, dpt_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dpt_test_data)
# make new output_dataproduct_url instance, but reuse related data for speed
output_dp_test_data = test_data_creator.Dataproduct(specifications_template_url=self.output_dataproduct_data['specifications_template'],
subtask_output_url=self.output_dataproduct_data['producer'],
dataproduct_feedback_template_url=self.output_dataproduct_data['feedback_template'])
output_dataproduct_url = test_data_creator.post_data_and_get_url(output_dp_test_data, '/dataproduct/')
test_patch = {"output": output_dataproduct_url,
"identity": False }
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(dpt_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_dataproduct_transform_DELETE(self):
dpt_test_data = test_data_creator.DataproductTransform(input_dataproduct_url=self.input_dataproduct_url, output_dataproduct_url=self.output_dataproduct_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', dpt_test_data, 201, dpt_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dpt_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_dataproduct_transform_PROTECT_behavior_on_input_deleted(self):
# make new input_dataproduct_url instance, but reuse related data for speed
input_dp_test_data = test_data_creator.Dataproduct(specifications_template_url=self.input_dataproduct_data['specifications_template'],
subtask_output_url=self.input_dataproduct_data['producer'],
dataproduct_feedback_template_url=self.input_dataproduct_data['feedback_template'])
input_dataproduct_url = test_data_creator.post_data_and_get_url(input_dp_test_data, '/dataproduct/')
dpt_test_data = test_data_creator.DataproductTransform(input_dataproduct_url=input_dataproduct_url, output_dataproduct_url=self.output_dataproduct_url)
# POST new item and verify
url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', dpt_test_data, 201, dpt_test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, dpt_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(input_dataproduct_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, input_dataproduct_url, input_dp_test_data)
def test_dataproduct_transform_PROTECT_behavior_on_output_deleted(self):
# make new output_dataproduct_url instance, but reuse related data for speed
output_dp_test_data = test_data_creator.Dataproduct(specifications_template_url=self.output_dataproduct_data['specifications_template'],
subtask_output_url=self.output_dataproduct_data['producer'],
dataproduct_feedback_template_url=self.output_dataproduct_data['feedback_template'])
output_dataproduct_url = test_data_creator.post_data_and_get_url(output_dp_test_data, '/dataproduct/')
dpt_test_data = test_data_creator.DataproductTransform(output_dataproduct_url=output_dataproduct_url, input_dataproduct_url=self.input_dataproduct_url)
# POST new item and verify
url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', dpt_test_data, 201, dpt_test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, dpt_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(output_dataproduct_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_OK_and_assert_equal_expected_response(self, output_dataproduct_url, output_dp_test_data)
class FilesystemTestCase(unittest.TestCase):
def test_filesystem_list_apiformat(self):
r = requests.get(BASE_URL + '/filesystem/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Filesystem List" in r.content.decode('utf8'))
def test_filesystem_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/filesystem/1234321/', 404)
def test_filesystem_POST_and_GET(self):
fs_test_data = test_data_creator.Filesystem()
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/filesystem/', fs_test_data, 201, fs_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, fs_test_data)
def test_filesystem_PUT_invalid_raises_error(self):
fs_test_data = test_data_creator.Filesystem()
PUT_and_assert_expected_response(self, BASE_URL + '/filesystem/9876789876/', fs_test_data,
404, {})
def test_filesystem_PUT(self):
fs_test_data = test_data_creator.Filesystem()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/filesystem/', fs_test_data,
201, fs_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, fs_test_data)
fs_test_data2 = test_data_creator.Filesystem()
# PUT new values, verify
PUT_and_assert_expected_response(self, url, fs_test_data2, 200, fs_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, fs_test_data2)
def test_filesystem_PATCH(self):
cluster_url = test_data_creator.post_data_and_get_url(test_data_creator.Cluster(), '/cluster/')
fs_test_data = test_data_creator.Filesystem(cluster_url=cluster_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/filesystem/', fs_test_data,
201, fs_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, fs_test_data)
cluster_url2 = test_data_creator.post_data_and_get_url(test_data_creator.Cluster(), '/cluster/')
test_patch = {"cluster": cluster_url2,
"capacity": 3333333333}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(fs_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_filesystem_DELETE(self):
fs_test_data = test_data_creator.Filesystem()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/filesystem/', fs_test_data,
201, fs_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, fs_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_filesystem_PROTECT_behavior_on_cluster_deleted(self):
fs_test_data = test_data_creator.Filesystem()
# POST new item and verify
url = POST_and_assert_expected_response(self, BASE_URL + '/filesystem/', fs_test_data, 201,
fs_test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, fs_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(fs_test_data['cluster'], auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_and_assert_equal_expected_code(self, fs_test_data['cluster'], 200)
class ClusterTestCase(unittest.TestCase):
def test_cluster_list_apiformat(self):
r = requests.get(BASE_URL + '/cluster/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Cluster List" in r.content.decode('utf8'))
def test_cluster_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/cluster/1234321/', 404)
def test_cluster_POST_and_GET(self):
c_test_data = test_data_creator.Cluster()
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cluster/', c_test_data, 201, c_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, c_test_data)
def test_cluster_PUT_invalid_raises_error(self):
c_test_data = test_data_creator.Cluster()
PUT_and_assert_expected_response(self, BASE_URL + '/cluster/9876789876/', c_test_data, 404, {})
def test_cluster_PUT(self):
c_test_data = test_data_creator.Cluster()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cluster/', c_test_data, 201, c_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, c_test_data)
c_test_data2 = test_data_creator.Cluster()
# PUT new values, verify
PUT_and_assert_expected_response(self, url, c_test_data2, 200, c_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, c_test_data2)
def test_cluster_PATCH(self):
c_test_data = test_data_creator.Cluster()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cluster/', c_test_data, 201, c_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, c_test_data)
test_patch = {"location": 'at the other end of the universe'}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(c_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_cluster_DELETE(self):
c_test_data = test_data_creator.Cluster()
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cluster/', c_test_data, 201, c_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, c_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
class DataproductHashTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.dataproduct_url = test_data_creator.post_data_and_get_url(test_data_creator.Dataproduct(), '/dataproduct/')
def test_dataproduct_hash_list_apiformat(self):
r = requests.get(BASE_URL + '/dataproduct_hash/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Dataproduct Hash List" in r.content.decode('utf8'))
def test_dataproduct_hash_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/dataproduct_hash/1234321/', 404)
def test_dataproduct_hash_POST_and_GET(self):
dph_test_data = test_data_creator.DataproductHash(dataproduct_url=self.dataproduct_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', dph_test_data,
201, dph_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dph_test_data)
def test_dataproduct_hash_PUT_invalid_raises_error(self):
dph_test_data = test_data_creator.DataproductHash(dataproduct_url=self.dataproduct_url)
PUT_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/9876789876/', dph_test_data,
404, {})
def test_dataproduct_hash_PUT(self):
dph_test_data = test_data_creator.DataproductHash(hash="the_one", dataproduct_url=self.dataproduct_url)
dph_test_data2 = test_data_creator.DataproductHash(hash="the_other", dataproduct_url=self.dataproduct_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', dph_test_data,
201, dph_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dph_test_data)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, dph_test_data2, 200, dph_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, dph_test_data2)
def test_dataproduct_hash_PATCH(self):
dph_test_data = test_data_creator.DataproductHash(dataproduct_url=self.dataproduct_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', dph_test_data,
201, dph_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dph_test_data)
test_patch = {"algorithm": BASE_URL + '/algorithm/aes256',
"hash": 'bender-was-here'}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(dph_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_dataproduct_hash_DELETE(self):
dph_test_data = test_data_creator.DataproductHash(dataproduct_url=self.dataproduct_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', dph_test_data,
201, dph_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dph_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_dataproduct_hash_PROTECT_behavior_on_dataproduct_deleted(self):
dph_test_data = test_data_creator.DataproductHash(dataproduct_url=self.dataproduct_url)
# POST new item and verify
url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', dph_test_data, 201,
dph_test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, dph_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(dph_test_data['dataproduct'], auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_and_assert_equal_expected_code(self, dph_test_data['dataproduct'], 200)
def test_dataproduct_hash_PROTECT_behavior_on_algorithm_deleted(self):
dph_test_data = test_data_creator.DataproductHash(dataproduct_url=self.dataproduct_url)
# POST new item and verify
url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', dph_test_data, 201,
dph_test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, dph_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(dph_test_data['algorithm'], auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_and_assert_equal_expected_code(self, dph_test_data['algorithm'], 200)
class DataproductArchiveInfoTestCase(unittest.TestCase):
def test_dataproduct_archive_info_list_apiformat(self):
r = requests.get(BASE_URL + '/dataproduct_archive_info/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Dataproduct Archive Info List" in r.content.decode('utf8'))
def test_dataproduct_archive_info_GET_nonexistant_raises_error(self):
GET_and_assert_equal_expected_code(self, BASE_URL + '/dataproduct_archive_info/1234321/', 404)
def test_dataproduct_archive_info_POST_and_GET(self):
dataproduct_url = test_data_creator.post_data_and_get_url(test_data_creator.Dataproduct(), '/dataproduct/')
dpai_test_data = test_data_creator.DataproductArchiveInfo(dataproduct_url=dataproduct_url)
# POST and GET a new item and assert correctness
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', dpai_test_data,
201, dpai_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dpai_test_data)
def test_dataproduct_archive_info_PUT_invalid_raises_error(self):
dataproduct_url = test_data_creator.post_data_and_get_url(test_data_creator.Dataproduct(), '/dataproduct/')
dpai_test_data = test_data_creator.DataproductArchiveInfo(dataproduct_url=dataproduct_url)
PUT_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/9876789876/', dpai_test_data,
404, {})
def test_dataproduct_archive_info_PUT(self):
dataproduct_url = test_data_creator.post_data_and_get_url(test_data_creator.Dataproduct(), '/dataproduct/')
dpai_test_data = test_data_creator.DataproductArchiveInfo(dataproduct_url=dataproduct_url)
dataproduct_url = test_data_creator.post_data_and_get_url(test_data_creator.Dataproduct(), '/dataproduct/')
dpai_test_data2 = test_data_creator.DataproductArchiveInfo(dataproduct_url=dataproduct_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', dpai_test_data,
201, dpai_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dpai_test_data)
# PUT new values, verify
PUT_and_assert_expected_response(self, url, dpai_test_data2, 200, dpai_test_data2)
GET_OK_and_assert_equal_expected_response(self, url, dpai_test_data2)
def test_dataproduct_archive_info_PATCH(self):
dataproduct_url = test_data_creator.post_data_and_get_url(test_data_creator.Dataproduct(), '/dataproduct/')
dpai_test_data = test_data_creator.DataproductArchiveInfo(dataproduct_url=dataproduct_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', dpai_test_data,
201, dpai_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dpai_test_data)
test_patch = {"storage_ticket": "mygoldenticket"}
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(dpai_test_data)
expected_data.update(test_patch)
GET_OK_and_assert_equal_expected_response(self, url, expected_data)
def test_dataproduct_archive_info_DELETE(self):
dataproduct_url = test_data_creator.post_data_and_get_url(test_data_creator.Dataproduct(), '/dataproduct/')
dpai_test_data = test_data_creator.DataproductArchiveInfo(dataproduct_url=dataproduct_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', dpai_test_data,
201, dpai_test_data)
url = r_dict['url']
GET_OK_and_assert_equal_expected_response(self, url, dpai_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_dataproduct_archive_info_PROTECT_behavior_on_dataproduct_deleted(self):
dataproduct_url = test_data_creator.post_data_and_get_url(test_data_creator.Dataproduct(), '/dataproduct/')
dpai_test_data = test_data_creator.DataproductArchiveInfo(dataproduct_url=dataproduct_url)
# POST new item and verify
url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', dpai_test_data, 201,
dpai_test_data)['url']
GET_OK_and_assert_equal_expected_response(self, url, dpai_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
response = requests.delete(dpai_test_data['dataproduct'], auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_and_assert_equal_expected_code(self, dpai_test_data['dataproduct'], 200)
class SubtaskQueryTestCase(unittest.TestCase):
"""
Test queries on the subtask REST api:
- query cluster only
- query start and stop time and cluster
- query start and stop time
- query start time and cluster
- query stop time and cluster
- query with incorrect input
"""
#TODO: add proper indexes on start and stop time
def check_response_OK_and_result_count(self, response, expected_count):
"""
Check http response on status_code OK and the expected count number of the results list
"""
self.assertEqual(200, response.status_code)
json_response = response.json()
self.assertEqual(expected_count, json_response.get('count'))
@staticmethod
def get_total_number_of_subtasks():
"""
Retrieve the total number of current subtasks objects
"""
response = requests.get(BASE_URL + '/subtask/', auth=AUTH)
json_response = response.json()
return json_response.get('count')
@staticmethod
def create_multiple_subtask_object(total_number: int, cluster_name: str):
"""
Create multiple subtasks for a given number of days with start_time 2 hours from now and
stop_time 4 hours from now
"""
cluster_url = test_data_creator.post_data_and_get_url(test_data_creator.Cluster(name=cluster_name), '/cluster/')
task_blueprint_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskBlueprint(), '/task_blueprint/')
for day_idx in range(0, total_number):
start_time = datetime.now() + timedelta(hours=2, days=day_idx)
stop_time = datetime.now() + timedelta(hours=4, days=day_idx)
test_data_creator.post_data_and_get_url(test_data_creator.Subtask(start_time=start_time, stop_time=stop_time,
cluster_url=cluster_url, task_blueprint_url=task_blueprint_url), '/subtask/')
subtasks_test_data_with_start_stop_time = {'clusterB': 50, 'clusterC': 30 }
@classmethod
def setUpClass(cls) -> None:
"""
Setup once before test is running. Create multiple subtask objects
clusterA 1 subtasks with start and stop time now
clusterB 50 subtasks with start 2hr and stop time 4hr from now, recurring 'every day'
clusterC 30 subtasks with start 2hr and stop time 4hr from now, recurring 'every day'
"""
# we're counting (filtered) subtasks, so we should not depend on "previous" data
models.SubtaskInput.objects.all().delete()
models.DataproductHash.objects.all().delete()
models.DataproductArchiveInfo.objects.all().delete()
models.DataproductTransform.objects.all().delete()
models.Dataproduct.objects.all().delete()
models.Subtask.objects.all().delete()
cluster_url = test_data_creator.post_data_and_get_url(test_data_creator.Cluster(name="clusterA"), '/cluster/')
test_data_creator.post_data_and_get_url(test_data_creator.Subtask(start_time=datetime.utcnow(), stop_time=datetime.utcnow(),
cluster_url=cluster_url), '/subtask/')
for cluster_name, period_length_in_days in SubtaskQueryTestCase.subtasks_test_data_with_start_stop_time.items():
SubtaskQueryTestCase.create_multiple_subtask_object(period_length_in_days, cluster_name)
def test_query_cluster_only(self):
"""
Check the query on cluster name. Check status code and response length
"""
logger.info("Check query on clusterA")
response = requests.get(BASE_URL + '/subtask/?cluster__name=clusterA', auth=AUTH)
self.check_response_OK_and_result_count(response, 1)
for cluster_name, period_length_in_days in SubtaskQueryTestCase.subtasks_test_data_with_start_stop_time.items():
logger.info("Check query on %s" % cluster_name)
response = requests.get(BASE_URL + '/subtask/?cluster__name=%s' % cluster_name, auth=AUTH)
self.check_response_OK_and_result_count(response, period_length_in_days)
def test_query_start_and_stop_time_and_cluster(self):
"""
Check if I can query on the start and stop time and cluster name (B and C) over a period
Check status code and response length
"""
for cluster_name, period_length_in_days in SubtaskQueryTestCase.subtasks_test_data_with_start_stop_time.items():
start_time = datetime.now()
stop_time = start_time + timedelta(days=period_length_in_days)
expected_count = period_length_in_days
logger.info("Check query in a period (%s until %s) for %s", formatDatetime(start_time), formatDatetime(stop_time), cluster_name)
response = requests.get(BASE_URL + '/subtask/?start_time__gt=%s&stop_time__lt=%s&cluster__name=%s' %
(start_time, stop_time, cluster_name), auth=AUTH)
self.check_response_OK_and_result_count(response, expected_count)
logger.info("Check number of subtasks every day for %s" % cluster_name)
for day_idx in range(0, period_length_in_days):
start_time = datetime.now() + timedelta(days=day_idx)
stop_time = start_time + timedelta(days=1)
response = requests.get(BASE_URL + '/subtask/?start_time__gt=%s&stop_time__lt=%s&cluster__name=%s' %
(start_time, stop_time, cluster_name), auth=AUTH)
self.check_response_OK_and_result_count(response, 1)
logger.info("Check query in a period (%s until %s) for clusterNotExist" %
(formatDatetime(start_time), formatDatetime(stop_time)))
response = requests.get(BASE_URL + '/subtask/?start_time__gt=%s&stop_time__lt=%s&cluster__name=%s' %
(start_time, stop_time, "clusterNotExist"), auth=AUTH)
self.check_response_OK_and_result_count(response, 0)
def test_query_start_and_stop_time(self):
"""
Check if I can query on the start and stop time over a period
Check status code and response length
"""
period_length_in_days = 50 # max(B+C)
expected_count = 80 # B+C
start_time = datetime.now()
stop_time = start_time + timedelta(days=period_length_in_days)
logger.info("Check query in a period (%s until %s)" %
(formatDatetime(start_time), formatDatetime(stop_time)))
response = requests.get(BASE_URL + '/subtask/?start_time__gt=%s&stop_time__lt=%s' %
(start_time, stop_time), auth=AUTH)
self.check_response_OK_and_result_count(response, expected_count)
logger.info("Check number of subtasks every day")
for day_idx in range(0, period_length_in_days):
start_time = datetime.now() + timedelta(days=day_idx)
stop_time = start_time + timedelta(days=1)
response = requests.get(BASE_URL + '/subtask/?start_time__gt=%s&stop_time__lt=%s' %
(start_time, stop_time), auth=AUTH)
if day_idx >= 30:
expected_count = 1 # B
else:
expected_count = 2 # B+C
self.check_response_OK_and_result_count(response, expected_count)
def test_query_start_and_cluster(self):
"""
Check if I can query on the start time and cluster name (B and C) over a period
Check status code and response length
"""
for cluster_name, period_length_in_days in SubtaskQueryTestCase.subtasks_test_data_with_start_stop_time.items():
start_time = datetime.now()
expected_count = period_length_in_days
logger.info("Check query greater than start_time (%s) for %s " %
(formatDatetime(start_time), cluster_name))
response = requests.get(BASE_URL + '/subtask/?start_time__gt=%s&cluster__name=%s' %
(start_time, cluster_name), auth=AUTH)
self.check_response_OK_and_result_count(response, expected_count)
logger.info("Check number of subtasks every day for %s" % cluster_name)
expected_count = period_length_in_days
for day_idx in range(0, period_length_in_days):
start_time = datetime.now() + timedelta(days=day_idx)
response = requests.get(BASE_URL + '/subtask/?start_time__gt=%s&cluster__name=%s' %
(start_time, cluster_name), auth=AUTH)
self.check_response_OK_and_result_count(response, expected_count)
expected_count -= 1 # every another day one less
def test_query_stop_and_cluster(self):
"""
Check if I can query on the stop time and cluster name (B and C) over a period
Check status code and response length
"""
for cluster_name, period_length_in_days in SubtaskQueryTestCase.subtasks_test_data_with_start_stop_time.items():
stop_time = datetime.now() + timedelta(days=period_length_in_days)
logger.info("Check query less than stop_time (%s) for %s " %
(formatDatetime(stop_time), cluster_name))
response = requests.get(BASE_URL + '/subtask/?stop_time__lt=%s&cluster__name=%s' %
(stop_time, cluster_name), auth=AUTH)
self.check_response_OK_and_result_count(response, period_length_in_days)
logger.info("Check number of subtasks every day for %s" % cluster_name)
expected_count = 1
for day_idx in range(0, period_length_in_days):
stop_time = datetime.now() + timedelta(days=day_idx+1)
response = requests.get(BASE_URL + '/subtask/?stop_time__lt=%s&cluster__name=%s' %
(stop_time, cluster_name), auth=AUTH)
self.check_response_OK_and_result_count(response, expected_count)
expected_count += 1 # every another day one more
def test_query_wrong_input(self):
"""
Check the query when wrong input is given;
- query on a none existing cluster name
- query start time larger than stop time
- query start_time less than and stop_time greater than
- wrong query name
Note! when error in query name, REST will return ALL (in this case 82 objects)
"""
response = requests.get(BASE_URL + '/subtask/?cluster__name=clusterNotExist', auth=AUTH)
self.check_response_OK_and_result_count(response, 0)
# Check how many is 'ALL'
total_subtasks = SubtaskQueryTestCase.get_total_number_of_subtasks()
response = requests.get(BASE_URL + '/subtask/?cluster__error_in_query=clusterA', auth=AUTH)
self.check_response_OK_and_result_count(response, total_subtasks)
period_length_in_days = 50 # max(B+C)
stop_time = datetime.now()
start_time = stop_time + timedelta(days=period_length_in_days)
logger.info("Check 'wrong' query in a period (%s until %s)" %
(formatDatetime(start_time), formatDatetime(stop_time)))
response = requests.get(BASE_URL + '/subtask/?start_time__gt=%s&stop_time__lt=%s' %
(start_time, stop_time), auth=AUTH)
self.check_response_OK_and_result_count(response, 0)
start_time = datetime.now()
stop_time = start_time + timedelta(days=period_length_in_days)
logger.info("Check 'wrong' query in a period (%s until %s)" %
(formatDatetime(start_time), formatDatetime(stop_time)))
response = requests.get(BASE_URL + '/subtask/?start_time__lt=%s&stop_time__gt=%s' %
(start_time, stop_time), auth=AUTH)
self.check_response_OK_and_result_count(response, 0)
def test_query_state_only(self):
"""
Check the query on state value. Check status code and response length
All states are defining (by default), None are defined
"""
logger.info("Check query on state defining")
total_number_of_subtasks = SubtaskQueryTestCase.get_total_number_of_subtasks()
response = requests.get(BASE_URL + '/subtask/?state__value=defining', auth=AUTH)
self.check_response_OK_and_result_count(response, total_number_of_subtasks)
response = requests.get(BASE_URL + '/subtask/?state__value=defined', auth=AUTH)
self.check_response_OK_and_result_count(response, 0)
def test_query_ordering_start_time(self):
"""
Check the query on ordering of start_time in ascending (default) and descending order
Check status code and response length
Check if next start_time in response is 'younger' in ascending order
Check if next start_time in response is 'older' in descending order
"""
logger.info("Check query on ordering ascending start time")
response = requests.get(BASE_URL + '/subtask/?ordering=start_time', auth=AUTH)
self.check_response_OK_and_result_count(response, SubtaskQueryTestCase.get_total_number_of_subtasks())
previous_start_time = "2000-01-01T00:00:00"
for item in response.json().get('results'):
start_time = item['start_time']
self.assertGreater(start_time, previous_start_time, "The start time should be greater than the previous one")
previous_start_time = start_time
logger.info("Check query on ordering descending start time")
response = requests.get(BASE_URL + '/subtask/?ordering=-start_time', auth=AUTH)
self.check_response_OK_and_result_count(response, SubtaskQueryTestCase.get_total_number_of_subtasks())
previous_start_time = "2100-01-01T00:00:00"
for item in response.json().get('results'):
start_time = item['start_time']
self.assertLess(start_time, previous_start_time, "The start time should be smaller than the previous one")
previous_start_time = start_time
if __name__ == "__main__":
logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO)
unittest.main()