Skip to content
Snippets Groups Projects

Resolve TMSS-146

Merged Resolve TMSS-146
TMSS-146 into master
Merged Jorrit Schaap requested to merge TMSS-146 into master
Viewing commit fdc67d63
Prev Next
Show latest version
1 file
+ 158
− 82
Compare changes
  • Side-by-side
  • Inline
SAS/TMSS/test/t_tmssapp_scheduling_functional.py
+ 158
− 82
@@ -39,13 +39,12 @@ logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=loggin
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.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)
DJANGO_TIMEFORMAT = "%Y-%m-%dT%H:%M:%S"
class SubtaskTemplateTestCase(unittest.TestCase):
def test_subtask_template_list_apiformat(self):
@@ -251,6 +250,13 @@ class DefaultSubtaskTemplatesTestCase(unittest.TestCase):
class SubtaskTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
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)
@@ -260,7 +266,7 @@ class SubtaskTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/subtask/1234321/', 404, {})
def test_subtask_POST_and_GET(self):
st_test_data = test_data_creator.Subtask()
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)
@@ -271,13 +277,13 @@ class SubtaskTestCase(unittest.TestCase):
self.assertGreaterEqual(int(subtask_id), minimium_subtaskid)
def test_subtask_PUT_invalid_raises_error(self):
st_test_data = test_data_creator.Subtask()
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()
st_test_data2 = test_data_creator.Subtask()
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)
@@ -289,7 +295,7 @@ class SubtaskTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, st_test_data2)
def test_subtask_PATCH(self):
st_test_data = test_data_creator.Subtask()
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)
@@ -305,7 +311,7 @@ class SubtaskTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_subtask_DELETE(self):
st_test_data = test_data_creator.Subtask()
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)
@@ -316,7 +322,7 @@ class SubtaskTestCase(unittest.TestCase):
DELETE_and_assert_gone(self, url)
def test_subtask_PROTECT_behavior_on_state_choice_deleted(self):
st_test_data = test_data_creator.Subtask()
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'}
@@ -337,9 +343,12 @@ class SubtaskTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, state_url, 200, state_data)
def test_subtask_SET_NULL_behavior_on_task_blueprint_deleted(self):
tbp_test_data = test_data_creator.TaskBlueprint()
# 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)
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']
@@ -356,7 +365,7 @@ class SubtaskTestCase(unittest.TestCase):
def test_subtask_PROTECT_behavior_on_template_deleted(self):
stt_test_data = test_data_creator.SubtaskTemplate()
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)
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']
@@ -371,6 +380,12 @@ class SubtaskTestCase(unittest.TestCase):
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)
@@ -380,7 +395,7 @@ class DataproductTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/dataproduct/1234321/', 404, {})
def test_dataproduct_POST_and_GET(self):
dp_test_data = test_data_creator.Dataproduct()
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)
@@ -388,13 +403,13 @@ class DataproductTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, dp_test_data)
def test_dataproduct_PUT_invalid_raises_error(self):
dp_test_data = test_data_creator.Dataproduct()
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()
dp_test_data2 = test_data_creator.Dataproduct()
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)
@@ -406,7 +421,7 @@ class DataproductTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, dp_test_data2)
def test_dataproduct_PATCH(self):
dp_test_data = test_data_creator.Dataproduct()
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)
@@ -423,7 +438,7 @@ class DataproductTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_dataproduct_DELETE(self):
dp_test_data = test_data_creator.Dataproduct()
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)
@@ -434,7 +449,7 @@ class DataproductTestCase(unittest.TestCase):
DELETE_and_assert_gone(self, url)
def test_dataproduct_PROTECT_behavior_on_dataformat_deleted(self):
dp_test_data = test_data_creator.Dataproduct()
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'}
@@ -456,7 +471,7 @@ class DataproductTestCase(unittest.TestCase):
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)
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']
@@ -578,6 +593,18 @@ class SubtaskConnectorTestCase(unittest.TestCase):
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_connector_url = test_data_creator.post_data_and_get_url(test_data_creator.SubtaskConnector(), '/subtask_connector/')
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.subtask_input_selection_template_url = test_data_creator.post_data_and_get_url(test_data_creator.SubtaskInputSelectionTemplate(), '/subtask_input_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)
@@ -587,7 +614,7 @@ class SubtaskInputTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/subtask_input/1234321/', 404, {})
def test_subtask_input_POST_and_GET(self):
sti_test_data = test_data_creator.SubtaskInput()
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_connector_url=self.subtask_connector_url, subtask_output_url=self.subtask_output_url, subtask_input_selection_template_url=self.subtask_input_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)
@@ -595,12 +622,12 @@ class SubtaskInputTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, sti_test_data)
def test_subtask_input_PUT_invalid_raises_error(self):
sti_test_data = test_data_creator.SubtaskInput()
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_connector_url=self.subtask_connector_url, subtask_output_url=self.subtask_output_url, subtask_input_selection_template_url=self.subtask_input_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()
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_connector_url=self.subtask_connector_url, subtask_output_url=self.subtask_output_url, subtask_input_selection_template_url=self.subtask_input_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)
@@ -608,19 +635,23 @@ class SubtaskInputTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, sti_test_data)
# PUT new values, verify
sti_test_data2 = test_data_creator.SubtaskInput()
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_connector_url=self.subtask_connector_url, subtask_output_url=self.subtask_output_url, subtask_input_selection_template_url=self.subtask_input_selection_template_url)
PUT_and_assert_expected_response(self, url, sti_test_data2, 200, sti_test_data2)
GET_and_assert_expected_response(self, url, 200, sti_test_data2)
def test_subtask_input_PATCH(self):
sti_test_data = test_data_creator.SubtaskInput()
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_connector_url=self.subtask_connector_url, subtask_output_url=self.subtask_output_url, subtask_input_selection_template_url=self.subtask_input_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_and_assert_expected_response(self, url, 200, sti_test_data)
subtask_url = test_data_creator.post_data_and_get_url(test_data_creator.Subtask(), '/subtask/')
# 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'],
}
@@ -632,7 +663,7 @@ class SubtaskInputTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_subtask_input_DELETE(self):
sti_test_data = test_data_creator.SubtaskInput()
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_connector_url=self.subtask_connector_url, subtask_output_url=self.subtask_output_url, subtask_input_selection_template_url=self.subtask_input_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)
@@ -643,8 +674,12 @@ class SubtaskInputTestCase(unittest.TestCase):
DELETE_and_assert_gone(self, url)
def test_subtask_input_CASCADE_behavior_on_subtask_deleted(self):
subtask_url = test_data_creator.post_data_and_get_url(test_data_creator.Subtask(), '/subtask/')
sti_test_data = test_data_creator.SubtaskInput(subtask_url=subtask_url)
# 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_connector_url=self.subtask_connector_url, subtask_output_url=self.subtask_output_url, subtask_input_selection_template_url=self.subtask_input_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']
@@ -658,7 +693,7 @@ class SubtaskInputTestCase(unittest.TestCase):
def test_subtask_input_SET_NULL_behavior_on_connector_deleted(self):
subtask_connector_url = test_data_creator.post_data_and_get_url(test_data_creator.SubtaskConnector(), '/subtask_connector/')
sti_test_data = test_data_creator.SubtaskInput(subtask_connector_url=subtask_connector_url)
sti_test_data = test_data_creator.SubtaskInput(subtask_connector_url=subtask_connector_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, subtask_input_selection_template_url=self.subtask_input_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']
@@ -673,8 +708,11 @@ class SubtaskInputTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_subtask_input_SET_NULL_behavior_on_task_relation_blueprint_deleted(self):
task_relation_blueprint_url = test_data_creator.post_data_and_get_url(test_data_creator.TaskRelationBlueprint(), '/task_relation_blueprint/')
sti_test_data = test_data_creator.SubtaskInput(task_relation_blueprint_url=task_relation_blueprint_url)
# 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_url=self.task_relation_blueprint_data['input'], output_url=self.task_relation_blueprint_data['output'],
consumer_url=self.task_relation_blueprint_data['consumer'], producer_url=self.task_relation_blueprint_data['producer']), '/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_connector_url=self.subtask_connector_url, subtask_output_url=self.subtask_output_url, subtask_input_selection_template_url=self.subtask_input_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']
@@ -689,8 +727,9 @@ class SubtaskInputTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_subtask_input_PROTECT_behavior_on_producer_deleted(self):
subtask_output_url = test_data_creator.post_data_and_get_url(test_data_creator.SubtaskOutput(), '/subtask_output/')
sti_test_data = test_data_creator.SubtaskInput(subtask_output_url=subtask_output_url)
# 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_connector_url=self.subtask_output_data['connector']), '/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, subtask_connector_url=self.subtask_connector_url, subtask_input_selection_template_url=self.subtask_input_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']
@@ -705,7 +744,12 @@ class SubtaskInputTestCase(unittest.TestCase):
def test_subtask_input_PROTECT_behavior_on_selection_template_deleted(self):
subtask_input_selection_template_url = test_data_creator.post_data_and_get_url(test_data_creator.SubtaskInputSelectionTemplate(), '/subtask_input_selection_template/')
sti_test_data = test_data_creator.SubtaskInput(subtask_input_selection_template_url=subtask_input_selection_template_url)
sti_test_data = test_data_creator.SubtaskInput(subtask_input_selection_template_url=subtask_input_selection_template_url,
subtask_url=self.subtask_url,
task_relation_blueprint_url=self.task_relation_blueprint_url,
dataproduct_urls=self.dataproduct_urls,
subtask_connector_url=self.subtask_connector_url,
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']
@@ -720,6 +764,13 @@ class SubtaskInputTestCase(unittest.TestCase):
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/')
cls.subtask_connector_data = test_data_creator.SubtaskConnector()
cls.subtask_connector_url = test_data_creator.post_data_and_get_url(cls.subtask_connector_data, '/subtask_connector/')
def test_subtask_output_list_apiformat(self):
r = requests.get(BASE_URL + '/subtask_output/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
@@ -729,7 +780,7 @@ class SubtaskOutputTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/subtask_output/1234321/', 404, {})
def test_subtask_output_POST_and_GET(self):
sto_test_data = test_data_creator.SubtaskOutput()
sto_test_data = test_data_creator.SubtaskOutput(subtask_url=self.subtask_url, subtask_connector_url=self.subtask_connector_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,
@@ -738,12 +789,12 @@ class SubtaskOutputTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, sto_test_data)
def test_subtask_output_PUT_invalid_raises_error(self):
sto_test_data = test_data_creator.SubtaskOutput()
sto_test_data = test_data_creator.SubtaskOutput(subtask_url=self.subtask_url, subtask_connector_url=self.subtask_connector_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()
sto_test_data2 = test_data_creator.SubtaskOutput()
sto_test_data = test_data_creator.SubtaskOutput(subtask_url=self.subtask_url, subtask_connector_url=self.subtask_connector_url)
sto_test_data2 = test_data_creator.SubtaskOutput(subtask_url=self.subtask_url, subtask_connector_url=self.subtask_connector_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', sto_test_data, 201,sto_test_data)
@@ -755,8 +806,8 @@ class SubtaskOutputTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, sto_test_data2)
def test_subtask_output_PATCH(self):
sto_test_data = test_data_creator.SubtaskOutput()
sto_test_data2 = test_data_creator.SubtaskOutput()
sto_test_data = test_data_creator.SubtaskOutput(subtask_url=self.subtask_url, subtask_connector_url=self.subtask_connector_url)
sto_test_data2 = test_data_creator.SubtaskOutput(subtask_url=self.subtask_url, subtask_connector_url=self.subtask_connector_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', sto_test_data, 201,
@@ -774,7 +825,7 @@ class SubtaskOutputTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_subtask_output_DELETE(self):
sto_test_data = test_data_creator.SubtaskOutput()
sto_test_data = test_data_creator.SubtaskOutput(subtask_url=self.subtask_url, subtask_connector_url=self.subtask_connector_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', sto_test_data, 201,
@@ -786,9 +837,9 @@ class SubtaskOutputTestCase(unittest.TestCase):
DELETE_and_assert_gone(self, url)
def test_subtask_output_CASCADE_behavior_on_subtask_deleted(self):
st_test_data = test_data_creator.Subtask()
subtask_url = test_data_creator.post_data_and_get_url(st_test_data, '/subtask/')
sto_test_data = test_data_creator.SubtaskOutput(subtask_url=subtask_url)
# 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, subtask_connector_url=self.subtask_connector_url)
# POST new item, verify
url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', sto_test_data, 201, sto_test_data)['url']
@@ -801,7 +852,7 @@ class SubtaskOutputTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 404, {})
def test_subtask_output_SET_NULL_behavior_on_connector_deleted(self):
sto_test_data = test_data_creator.SubtaskOutput()
sto_test_data = test_data_creator.SubtaskOutput(subtask_url=self.subtask_url, subtask_connector_url=self.subtask_connector_url)
# POST new item, verify
url = \
@@ -905,6 +956,13 @@ class AntennaSetTestCase(unittest.TestCase):
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)
@@ -914,7 +972,7 @@ class DataproductTransformTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/1234321/', 404, {})
def test_dataproduct_transform_POST_and_GET(self):
dpt_test_data = test_data_creator.DataproductTransform()
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)
@@ -922,13 +980,13 @@ class DataproductTransformTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, dpt_test_data)
def test_dataproduct_transform_PUT_invalid_raises_error(self):
dpt_test_data = test_data_creator.DataproductTransform()
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()
dpt_test_data2 = test_data_creator.DataproductTransform()
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)
@@ -940,14 +998,18 @@ class DataproductTransformTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, dpt_test_data2)
def test_dataproduct_transform_PATCH(self):
dpt_test_data = test_data_creator.DataproductTransform()
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_and_assert_expected_response(self, url, 200, dpt_test_data)
output_dataproduct_url = test_data_creator.post_data_and_get_url(test_data_creator.Dataproduct(), '/dataproduct/')
# 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 }
@@ -959,7 +1021,7 @@ class DataproductTransformTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_dataproduct_transform_DELETE(self):
dpt_test_data = test_data_creator.DataproductTransform()
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)
@@ -970,9 +1032,12 @@ class DataproductTransformTestCase(unittest.TestCase):
DELETE_and_assert_gone(self, url)
def test_dataproduct_transform_PROTECT_behavior_on_input_deleted(self):
input_dp_test_data = test_data_creator.Dataproduct()
# 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)
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']
@@ -986,9 +1051,12 @@ class DataproductTransformTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, input_dataproduct_url, 200, input_dp_test_data)
def test_dataproduct_transform_PROTECT_behavior_on_output_deleted(self):
output_dp_test_data = test_data_creator.Dataproduct()
# 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)
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']
@@ -1152,6 +1220,10 @@ class ClusterTestCase(unittest.TestCase):
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)
@@ -1161,7 +1233,7 @@ class DataproductHashTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/1234321/', 404, {})
def test_dataproduct_hash_POST_and_GET(self):
dph_test_data = test_data_creator.DataproductHash()
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,
@@ -1170,14 +1242,14 @@ class DataproductHashTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, dph_test_data)
def test_dataproduct_hash_PUT_invalid_raises_error(self):
dph_test_data = test_data_creator.DataproductHash()
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")
dph_test_data2 = test_data_creator.DataproductHash(hash="the other")
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,
@@ -1190,7 +1262,7 @@ class DataproductHashTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, dph_test_data2)
def test_dataproduct_hash_PATCH(self):
dph_test_data = test_data_creator.DataproductHash()
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,
@@ -1208,7 +1280,7 @@ class DataproductHashTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_dataproduct_hash_DELETE(self):
dph_test_data = test_data_creator.DataproductHash()
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,
@@ -1220,7 +1292,7 @@ class DataproductHashTestCase(unittest.TestCase):
DELETE_and_assert_gone(self, url)
def test_dataproduct_hash_PROTECT_behavior_on_dataproduct_deleted(self):
dph_test_data = test_data_creator.DataproductHash()
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,
@@ -1235,7 +1307,7 @@ class DataproductHashTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, dph_test_data['dataproduct'], 200, {})
def test_dataproduct_hash_PROTECT_behavior_on_algorithm_deleted(self):
dph_test_data = test_data_creator.DataproductHash()
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,
@@ -1251,6 +1323,10 @@ class DataproductHashTestCase(unittest.TestCase):
class DataproductArchiveInfoTestCase(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_archive_info_list_apiformat(self):
r = requests.get(BASE_URL + '/dataproduct_archive_info/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
@@ -1260,7 +1336,7 @@ class DataproductArchiveInfoTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/1234321/', 404, {})
def test_dataproduct_archive_info_POST_and_GET(self):
dpai_test_data = test_data_creator.DataproductArchiveInfo()
dpai_test_data = test_data_creator.DataproductArchiveInfo(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_archive_info/', dpai_test_data,
@@ -1269,14 +1345,14 @@ class DataproductArchiveInfoTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, dpai_test_data)
def test_dataproduct_archive_info_PUT_invalid_raises_error(self):
dpai_test_data = test_data_creator.DataproductArchiveInfo()
dpai_test_data = test_data_creator.DataproductArchiveInfo(dataproduct_url=self.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):
dpai_test_data = test_data_creator.DataproductArchiveInfo()
dpai_test_data2 = test_data_creator.DataproductArchiveInfo()
dpai_test_data = test_data_creator.DataproductArchiveInfo(dataproduct_url=self.dataproduct_url)
dpai_test_data2 = test_data_creator.DataproductArchiveInfo(dataproduct_url=self.dataproduct_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', dpai_test_data,
@@ -1289,7 +1365,7 @@ class DataproductArchiveInfoTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, dpai_test_data2)
def test_dataproduct_archive_info_PATCH(self):
dpai_test_data = test_data_creator.DataproductArchiveInfo()
dpai_test_data = test_data_creator.DataproductArchiveInfo(dataproduct_url=self.dataproduct_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', dpai_test_data,
@@ -1306,7 +1382,7 @@ class DataproductArchiveInfoTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_dataproduct_archive_info_DELETE(self):
dpai_test_data = test_data_creator.DataproductArchiveInfo()
dpai_test_data = test_data_creator.DataproductArchiveInfo(dataproduct_url=self.dataproduct_url)
# POST new item, verify
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', dpai_test_data,
@@ -1318,7 +1394,7 @@ class DataproductArchiveInfoTestCase(unittest.TestCase):
DELETE_and_assert_gone(self, url)
def test_dataproduct_archive_info_PROTECT_behavior_on_dataproduct_deleted(self):
dpai_test_data = test_data_creator.DataproductArchiveInfo()
dpai_test_data = test_data_creator.DataproductArchiveInfo(dataproduct_url=self.dataproduct_url)
# POST new item and verify
url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', dpai_test_data, 201,
@@ -1343,6 +1419,7 @@ class SubtaskQuery(unittest.TestCase):
- 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):
"""
@@ -1376,8 +1453,8 @@ class SubtaskQuery(unittest.TestCase):
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)
subtask_data = Subtask_test_data(start_time=start_time.strftime(DJANGO_TIMEFORMAT),
stop_time=stop_time.strftime(DJANGO_TIMEFORMAT),
subtask_data = Subtask_test_data(start_time=formatDatetime(start_time),
stop_time=formatDatetime(stop_time),
cluster_object=cluster_object)
models.Subtask.objects.create(**subtask_data)
@@ -1420,8 +1497,7 @@ class SubtaskQuery(unittest.TestCase):
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",
(start_time.strftime(DJANGO_TIMEFORMAT), stop_time.strftime(DJANGO_TIMEFORMAT), cluster_name))
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)
@@ -1435,7 +1511,7 @@ class SubtaskQuery(unittest.TestCase):
self.check_response_OK_and_result_count(response, 1)
logger.info("Check query in a period (%s until %s) for clusterNotExist" %
(start_time.strftime(DJANGO_TIMEFORMAT), stop_time.strftime(DJANGO_TIMEFORMAT)))
(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)
@@ -1450,7 +1526,7 @@ class SubtaskQuery(unittest.TestCase):
start_time = datetime.now()
stop_time = start_time + timedelta(days=period_length_in_days)
logger.info("Check query in a period (%s until %s)" %
(start_time.strftime(DJANGO_TIMEFORMAT), stop_time.strftime(DJANGO_TIMEFORMAT)))
(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)
@@ -1476,7 +1552,7 @@ class SubtaskQuery(unittest.TestCase):
start_time = datetime.now()
expected_count = period_length_in_days
logger.info("Check query greater than start_time (%s) for %s " %
(start_time.strftime(DJANGO_TIMEFORMAT), cluster_name))
(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)
@@ -1498,7 +1574,7 @@ class SubtaskQuery(unittest.TestCase):
for cluster_name, period_length_in_days in SubtaskQuery.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 " %
(stop_time.strftime(DJANGO_TIMEFORMAT), cluster_name))
(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)
@@ -1533,7 +1609,7 @@ class SubtaskQuery(unittest.TestCase):
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)" %
(start_time.strftime(DJANGO_TIMEFORMAT), stop_time.strftime(DJANGO_TIMEFORMAT)))
(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)
@@ -1541,7 +1617,7 @@ class SubtaskQuery(unittest.TestCase):
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)" %
(start_time.strftime(DJANGO_TIMEFORMAT), stop_time.strftime(DJANGO_TIMEFORMAT)))
(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)