From 74f93ec2edce0d7057bcdfaa5cb454b4db1aebcc Mon Sep 17 00:00:00 2001
From: Jorrit Schaap <schaap@astron.nl>
Date: Fri, 31 Jan 2020 11:50:44 +0100
Subject: [PATCH] TMSS-139: factored out test data creation of the unittest
classes. Fixed unittests
---
.../t_tmssapp_specification_functional.py | 1495 +++++++----------
1 file changed, 653 insertions(+), 842 deletions(-)
diff --git a/SAS/TMSS/test/t_tmssapp_specification_functional.py b/SAS/TMSS/test/t_tmssapp_specification_functional.py
index c27e4cf82db..2333d78c4dc 100755
--- a/SAS/TMSS/test/t_tmssapp_specification_functional.py
+++ b/SAS/TMSS/test/t_tmssapp_specification_functional.py
@@ -27,6 +27,16 @@
# todo: behavior in a controlled way.
# todo: We should probably also fully test behavior wrt mandatory and nullable fields.
+import logging
+logger = logging.getLogger(__name__)
+logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO)
+
+# use setup/teardown magic for tmss test database
+from lofar.sas.tmss.test.tmss_test_environment_unittest_setup import *
+
+from requests.auth import HTTPBasicAuth
+AUTH = HTTPBasicAuth(tmss_test_env.ldap_server.dbcreds.user, tmss_test_env.ldap_server.dbcreds.password)
+BASE_URL = tmss_test_env.django_server.url
import os
import sys
@@ -34,18 +44,13 @@ import unittest
import requests
import json
from datetime import datetime
-from requests.auth import HTTPBasicAuth
from lofar.common import dbcredentials
import logging
import uuid
-DJANGO_PORT=os.environ.get('DJANGO_TEST_PORT', 0)
-BASE_URL = 'http://localhost:%s' % (DJANGO_PORT,)
-
-# read ldap_credentials from ~/.lofar/dbcredentials/<ldap_creds_filename.ini>
-# such a file is created automagically in the test_funcs.sh and passed to this test via the TMSS_LDAPCREDENTIALS environment variable.
-ldap_credentials = dbcredentials.DBCredentials().get(os.environ.get('TMSS_LDAPCREDENTIALS', 'tmss_ldap_test'))
-print('Using ldap credentials', ldap_credentials.stringWithHiddenPassword())
+def post_data_and_get_url(data, url_postfix):
+ """POST the given data the BASE_URL+url_postfix, and return the response's url"""
+ return json.loads(requests.post(BASE_URL + url_postfix, json=data, auth=AUTH).content.decode('utf-8'))['url']
def _call_API_and_assert_expected_response(self, url, call, data, expected_code, expected_content):
"""
@@ -53,15 +58,15 @@ def _call_API_and_assert_expected_response(self, url, call, data, expected_code,
:return: response as dict. This either contains the data of an entry or error details. If JSON cannot be parsed, return string.
"""
if call == 'PUT':
- response = requests.put(url, json=data, auth=self.auth)
+ response = requests.put(url, json=data, auth=AUTH)
elif call == 'POST':
- response = requests.post(url, json=data, auth=self.auth)
+ response = requests.post(url, json=data, auth=AUTH)
elif call == 'GET':
- response = requests.get(url, auth=self.auth)
+ response = requests.get(url, auth=AUTH)
elif call == 'PATCH':
- response = requests.patch(url, json=data, auth=self.auth)
+ response = requests.patch(url, json=data, auth=AUTH)
elif call == 'DELETE':
- response = requests.delete(url, auth=self.auth)
+ response = requests.delete(url, auth=AUTH)
else:
raise ValueError("The provided call '%s' is not a valid API method choice" % call)
@@ -118,50 +123,37 @@ def DELETE_and_assert_gone(self, url):
DELETE item at provided url and assert that the request was accepted by the server
:return: url for new item
"""
- response = requests.delete(url, auth=self.auth)
+ response = requests.delete(url, auth=AUTH)
if response.status_code != 204:
print("!!! Unexpected: [%s] - %s %s: %s" % (self.id(), 'DELETE', url, response.content))
self.assertEqual(response.status_code, 204)
- response = requests.get(url, auth=self.auth)
+ response = requests.get(url, auth=AUTH)
if response.status_code != 404:
print("!!! Unexpected: [%s] - %s %s: %s" % (self.id(), 'GET', url, response.content))
self.assertEqual(response.status_code, 404)
class BasicFunctionTestCase(unittest.TestCase):
-
- def setUp(self):
- pass
-
# todo: test_welcome_page (once we have one :))
+ pass
-class GeneratorTemplateTestCase(unittest.TestCase):
-
- def setUp(self):
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- test_data_1 = {"name": "generatortemplate1",
- "description": 'My one observation',
- "version": 'v0.314159265359',
- "schema": {"mykey": "my value"},
- "create_function": 'Funky',
- "tags": ["TMSS", "TESTING"]}
-
- test_data_2 = {"name": "generatortemplate2",
- "description": 'My other observation',
- "version": 'v3.14159265359',
- "schema": {"mykey": "my other value"},
- "create_function": 'Trousers',
- "tags": []}
+def GeneratorTemplate_test_data(name="generatortemplate"):
+ return {"name": name,
+ "description": 'My one observation',
+ "version": 'v0.314159265359',
+ "schema": {"mykey": "my value"},
+ "create_function": 'Funky',
+ "tags": ["TMSS", "TESTING"]}
+class GeneratorTemplateTestCase(unittest.TestCase):
test_patch = {"version": 'v6.28318530718',
"schema": {"mykey": "my better value"},
}
def test_generator_template_list_apiformat(self):
- r = requests.get(BASE_URL + '/generator_template/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/generator_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Generator Template List" in r.content.decode('utf8'))
@@ -171,71 +163,62 @@ class GeneratorTemplateTestCase(unittest.TestCase):
def test_generator_template_POST_and_GET(self):
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', GeneratorTemplate_test_data(), 201, GeneratorTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, GeneratorTemplate_test_data())
def test_generator_template_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/generator_template/9876789876/', self.test_data_1, 404, {})
+ PUT_and_assert_expected_response(self, BASE_URL + '/generator_template/9876789876/', GeneratorTemplate_test_data(), 404, {})
def test_generator_template_PUT(self):
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', GeneratorTemplate_test_data(), 201, GeneratorTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, GeneratorTemplate_test_data())
# PUT new values, verify
- PUT_and_assert_expected_response(self, url, self.test_data_2, 200, self.test_data_2)
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ PUT_and_assert_expected_response(self, url, GeneratorTemplate_test_data("generatortemplate2"), 200, GeneratorTemplate_test_data("generatortemplate2"))
+ GET_and_assert_expected_response(self, url, 200, GeneratorTemplate_test_data("generatortemplate2"))
def test_generator_template_PATCH(self):
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', GeneratorTemplate_test_data(), 201, GeneratorTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, GeneratorTemplate_test_data())
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
+ expected_data = dict(GeneratorTemplate_test_data())
expected_data.update(self.test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_generator_template_DELETE(self):
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/', GeneratorTemplate_test_data(), 201, GeneratorTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, GeneratorTemplate_test_data())
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
+def SchedulingUnitTemplate_test_data(name="schedulingunittemplate1"):
+ return { "name": name,
+ "description": 'My description',
+ "version": 'v0.314159265359',
+ "schema": {"mykey": "my value"},
+ "tags": ["TMSS", "TESTING"]}
-class SchedulingUnitTemplateTestCase(unittest.TestCase):
-
- def setUp(self):
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- test_data_1 = {"name": "schedulingunittemplate1",
- "description": 'My one observation',
- "version": 'v0.314159265359',
- "schema": {"mykey": "my value"},
- "tags": ["TMSS", "TESTING"]}
-
- test_data_2 = {"name": "schedulingunittemplate2",
- "description": 'My other observation',
- "version": 'v3.14159265359',
- "schema": {"mykey": "my other value"},
- "tags": []}
+class SchedulingUnitTemplateTestCase(unittest.TestCase):
test_patch = {"version": 'v6.28318530718',
"schema": {"mykey": "my better value"},
}
def test_scheduling_unit_template_list_apiformat(self):
- r = requests.get(BASE_URL + '/scheduling_unit_template/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/scheduling_unit_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Scheduling Unit Template List" in r.content.decode('utf8'))
@@ -245,73 +228,62 @@ class SchedulingUnitTemplateTestCase(unittest.TestCase):
def test_scheduling_unit_template_POST_and_GET(self):
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', SchedulingUnitTemplate_test_data(), 201, SchedulingUnitTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url+'?format=json', 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url+'?format=json', 200, SchedulingUnitTemplate_test_data())
def test_scheduling_unit_template_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/9876789876/', self.test_data_1, 404, {})
+ PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/9876789876/', SchedulingUnitTemplate_test_data(), 404, {})
def test_scheduling_unit_template_PUT(self):
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', SchedulingUnitTemplate_test_data(), 201, SchedulingUnitTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, SchedulingUnitTemplate_test_data())
# PUT new values, verify
- PUT_and_assert_expected_response(self, url, self.test_data_2, 200, self.test_data_2)
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ PUT_and_assert_expected_response(self, url, SchedulingUnitTemplate_test_data("schedulingunittemplate2"), 200, SchedulingUnitTemplate_test_data("schedulingunittemplate2"))
+ GET_and_assert_expected_response(self, url, 200, SchedulingUnitTemplate_test_data("schedulingunittemplate2"))
def test_scheduling_unit_template_PATCH(self):
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', SchedulingUnitTemplate_test_data(), 201, SchedulingUnitTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, SchedulingUnitTemplate_test_data())
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
+ expected_data = dict(SchedulingUnitTemplate_test_data())
expected_data.update(self.test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_scheduling_unit_template_DELETE(self):
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/', SchedulingUnitTemplate_test_data(), 201, SchedulingUnitTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, SchedulingUnitTemplate_test_data())
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
+def TaskTemplate_test_data(name="tasktemplate1"):
+ return {"name": name,
+ "description": 'My one observation',
+ "version": 'v0.314159265359',
+ "schema": {"mykey": "my value"},
+ "tags": ["TMSS", "TESTING"],
+ "validation_code_js": "???"}
class TaskTemplateTestCase(unittest.TestCase):
-
- def setUp(self):
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- test_data_1 = {"name": "tasktemplate1",
- "description": 'My one observation',
- "version": 'v0.314159265359',
- "schema": {"mykey": "my value"},
- "tags": ["TMSS", "TESTING"],
- "validation_code_js": "???"}
-
- test_data_2 = {"name": "tasktemplate2",
- "description": 'My one observation',
- "version": 'v3.14159265359',
- "schema": {"mykey": "my other value"},
- "tags": [],
- "validation_code_js": "!!!"}
-
test_patch = {"version": 'v6.28318530718',
"schema": {"mykey": "my better value"},
}
def test_task_template_list_apiformat(self):
- r = requests.get(BASE_URL + '/task_template/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/task_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Template List" in r.content.decode('utf8'))
@@ -320,73 +292,63 @@ class TaskTemplateTestCase(unittest.TestCase):
def test_task_template_POST_and_GET(self):
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', self.test_data_1, 201,
- self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', TaskTemplate_test_data(), 201,
+ TaskTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url + '?format=json', 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url + '?format=json', 200, TaskTemplate_test_data())
def test_task_template_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/task_template/9876789876/', self.test_data_1, 404, {})
+ PUT_and_assert_expected_response(self, BASE_URL + '/task_template/9876789876/', TaskTemplate_test_data(), 404, {})
def test_task_template_PUT(self):
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', self.test_data_1, 201,
- self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', TaskTemplate_test_data(), 201,
+ TaskTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, TaskTemplate_test_data())
# PUT new values, verify
- PUT_and_assert_expected_response(self, url, self.test_data_2, 200, self.test_data_2)
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ PUT_and_assert_expected_response(self, url, TaskTemplate_test_data("tasktemplate2"), 200, TaskTemplate_test_data("tasktemplate2"))
+ GET_and_assert_expected_response(self, url, 200, TaskTemplate_test_data("tasktemplate2"))
def test_task_template_PATCH(self):
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', self.test_data_1, 201,
- self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', TaskTemplate_test_data(), 201,
+ TaskTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, TaskTemplate_test_data())
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
+ expected_data = dict(TaskTemplate_test_data())
expected_data.update(self.test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_task_template_DELETE(self):
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', self.test_data_1, 201,
- self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', TaskTemplate_test_data(), 201,
+ TaskTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, TaskTemplate_test_data())
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
-class WorkRelationSelectionTemplateTestCase(unittest.TestCase):
-
- def setUp(self):
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- # test data
- test_data_1 = {"name": "workrelationselectiontemplate1",
- "description": 'My one observation',
- "version": 'v0.314159265359',
- "schema": {"mykey": "my value"},
- "tags": ["TMSS", "TESTING"]}
-
- test_data_2 = {"name": "workrelationselectiontemplate2",
- "description": 'My other observation',
- "version": 'v3.14159265359',
- "schema": {"mykey": "my other value"},
- "tags": []}
+def WorkRelationSelectionTemplate_test_data(name="workrelationselectiontemplate1"):
+ return {"name": name,
+ "description": 'My one observation',
+ "version": 'v0.314159265359',
+ "schema": {"mykey": "my value"},
+ "tags": ["TMSS", "TESTING"]}
+class WorkRelationSelectionTemplateTestCase(unittest.TestCase):
test_patch = {"version": 'v6.28318530718',
"schema": {"mykey": "my better value"},
}
def test_work_relation_selection_template_list_apiformat(self):
- r = requests.get(BASE_URL + '/work_relation_selection_template/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/work_relation_selection_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Work Relation Selection Template List" in r.content.decode('utf8'))
@@ -396,84 +358,70 @@ class WorkRelationSelectionTemplateTestCase(unittest.TestCase):
def test_work_relation_selection_template_POST_and_GET(self):
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_relation_selection_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_relation_selection_template/', WorkRelationSelectionTemplate_test_data(), 201, WorkRelationSelectionTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url+'?format=json', 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url+'?format=json', 200, WorkRelationSelectionTemplate_test_data())
def test_work_relation_selection_template_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/work_relation_selection_template/9876789876/', self.test_data_1, 404, {})
+ PUT_and_assert_expected_response(self, BASE_URL + '/work_relation_selection_template/9876789876/', WorkRelationSelectionTemplate_test_data(), 404, {})
def test_work_relation_selection_template_PUT(self):
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_relation_selection_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_relation_selection_template/', WorkRelationSelectionTemplate_test_data(), 201, WorkRelationSelectionTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, WorkRelationSelectionTemplate_test_data())
# PUT new values, verify
- PUT_and_assert_expected_response(self, url, self.test_data_2, 200, self.test_data_2)
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ PUT_and_assert_expected_response(self, url, WorkRelationSelectionTemplate_test_data("workrelationselectiontemplate2"), 200, WorkRelationSelectionTemplate_test_data("workrelationselectiontemplate2"))
+ GET_and_assert_expected_response(self, url, 200, WorkRelationSelectionTemplate_test_data("workrelationselectiontemplate2"))
def test_work_relation_selection_template_PATCH(self):
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_relation_selection_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_relation_selection_template/', WorkRelationSelectionTemplate_test_data(), 201, WorkRelationSelectionTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, WorkRelationSelectionTemplate_test_data())
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
+ expected_data = dict(WorkRelationSelectionTemplate_test_data())
expected_data.update(self.test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_work_relation_selection_template_DELETE(self):
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_relation_selection_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_relation_selection_template/', WorkRelationSelectionTemplate_test_data(), 201, WorkRelationSelectionTemplate_test_data())
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, WorkRelationSelectionTemplate_test_data())
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
-class TaskConnectorsTestCase(unittest.TestCase):
-
- def setUp(self):
+def TaskConnectors_test_data(role="correlator", input_of_url=None, output_of_url=None):
+ if input_of_url is None:
+ input_of_url = post_data_and_get_url(TaskTemplate_test_data(), '/task_template/')
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
+ if output_of_url is None:
+ output_of_url = post_data_and_get_url(TaskTemplate_test_data(), '/task_template/')
- # related items
- output_of_url = POST_and_assert_expected_response(self, BASE_URL + '/task_template/',
- TaskTemplateTestCase.test_data_1, 201,
- TaskTemplateTestCase.test_data_1)['url']
+ return {"role": BASE_URL + '/role/%s/'%role,
+ "datatype": BASE_URL + '/datatype/image/',
+ "dataformats": [BASE_URL + '/dataformat/Beamformed/'],
+ "output_of": output_of_url,
+ "input_of": input_of_url,
+ "tags": []}
- input_of_url = POST_and_assert_expected_response(self, BASE_URL + '/task_template/',
- TaskTemplateTestCase.test_data_2, 201,
- TaskTemplateTestCase.test_data_2)['url']
+class TaskConnectorsTestCase(unittest.TestCase):
- # test data
- self.test_data_1 = {"role": BASE_URL + '/role/correlator/',
- "datatype": BASE_URL + '/datatype/image/',
- "dataformats": [BASE_URL + '/dataformat/Beamformed/'],
- "output_of": output_of_url,
- "input_of": input_of_url,
- "tags": []}
-
- self.test_data_2 = {"role": BASE_URL + '/role/target/',
- "datatype": BASE_URL + '/datatype/visibilities/',
- "dataformats": [BASE_URL + '/dataformat/MeasurementSet/'],
- "output_of": output_of_url,
- "input_of": input_of_url,
- "tags": []}
-
- self.test_patch = {"role": BASE_URL + '/role/calibrator/',
- "dataformats": [BASE_URL + '/dataformat/Beamformed/',
- BASE_URL + '/dataformat/MeasurementSet/']}
+ test_patch = {"role": BASE_URL + '/role/calibrator/',
+ "dataformats": [BASE_URL + '/dataformat/Beamformed/',
+ BASE_URL + '/dataformat/MeasurementSet/']}
def test_task_connectors_list_apiformat(self):
- r = requests.get(BASE_URL + '/task_connectors/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/task_connectors/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Connectors List" in r.content.decode('utf8'))
@@ -481,15 +429,16 @@ class TaskConnectorsTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/task_connectors/1234321/', 404, {})
def test_task_connectors_POST_and_GET(self):
+ tc_test_data = TaskConnectors_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', tc_test_data, 201, tc_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, tc_test_data)
def test_task_connectors_POST_invalid_role_raises_error(self):
# POST a new item with invalid choice
- test_data_invalid_role = dict(self.test_data_1)
+ test_data_invalid_role = dict(TaskConnectors_test_data())
test_data_invalid_role['role'] = BASE_URL + '/role/forbidden/'
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', test_data_invalid_role, 400, {})
self.assertTrue('Invalid hyperlink' in str(r_dict['role']))
@@ -497,7 +446,7 @@ class TaskConnectorsTestCase(unittest.TestCase):
def test_task_connectors_POST_invalid_datatype_raises_error(self):
# POST a new item with invalid choice
- test_data_invalid = dict(self.test_data_1)
+ test_data_invalid = dict(TaskConnectors_test_data())
test_data_invalid['datatype'] = BASE_URL + '/datatype/forbidden/'
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', test_data_invalid, 400, {})
self.assertTrue('Invalid hyperlink' in str(r_dict['datatype']))
@@ -505,7 +454,7 @@ class TaskConnectorsTestCase(unittest.TestCase):
def test_task_connectors_POST_invalid_dataformats_raises_error(self):
# POST a new item with invalid choice
- test_data_invalid = dict(self.test_data_1)
+ test_data_invalid = dict(TaskConnectors_test_data())
test_data_invalid['dataformats'] = [BASE_URL + '/dataformat/forbidden/']
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', test_data_invalid, 400, {})
self.assertTrue('Invalid hyperlink' in str(r_dict['dataformats']))
@@ -513,7 +462,7 @@ class TaskConnectorsTestCase(unittest.TestCase):
def test_task_connectors_POST_nonexistant_input_of_raises_error(self):
# POST a new item with wrong reference
- test_data_invalid = dict(self.test_data_1)
+ test_data_invalid = dict(TaskConnectors_test_data())
test_data_invalid['input_of'] = BASE_URL + "/task_template/6353748/"
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', test_data_invalid, 400, {})
self.assertTrue('Invalid hyperlink' in str(r_dict['input_of']))
@@ -521,7 +470,7 @@ class TaskConnectorsTestCase(unittest.TestCase):
def test_task_connectors_POST_nonexistant_output_of_raises_error(self):
# POST a new item with wrong reference
- test_data_invalid = dict(self.test_data_1)
+ test_data_invalid = dict(TaskConnectors_test_data())
test_data_invalid['output_of'] = BASE_URL + "/task_template/6353748/"
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', test_data_invalid, 400, {})
self.assertTrue('Invalid hyperlink' in str(r_dict['output_of']))
@@ -529,130 +478,132 @@ class TaskConnectorsTestCase(unittest.TestCase):
def test_task_connectors_POST_existing_outputs_works(self):
# First POST a new item to reference
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', TaskTemplateTestCase.test_data_1, 201, TaskTemplateTestCase.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/', TaskTemplate_test_data(), 201, TaskTemplate_test_data())
url = r_dict['url']
# POST a new item with correct reference
- test_data_valid = dict(self.test_data_1)
+ test_data_valid = dict(TaskConnectors_test_data())
test_data_valid['output_of'] = url
POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', test_data_valid, 201, test_data_valid)
def test_task_connectors_PUT_nonexistant_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/task_connectors/9876789876/', self.test_data_1, 404, {})
+ PUT_and_assert_expected_response(self, BASE_URL + '/task_connectors/9876789876/', TaskConnectors_test_data(), 404, {})
def test_task_connectors_PUT(self):
+ tc_test_data1 = TaskConnectors_test_data(role="correlator")
+ tc_test_data2 = TaskConnectors_test_data(role="beamformer")
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', tc_test_data1, 201, tc_test_data1)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, tc_test_data1)
# PUT new values, verify
- PUT_and_assert_expected_response(self, url, self.test_data_2, 200, self.test_data_2)
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ PUT_and_assert_expected_response(self, url, tc_test_data2, 200, tc_test_data2)
+ GET_and_assert_expected_response(self, url, 200, tc_test_data2)
def test_task_connectors_PATCH(self):
+ tc_test_data = TaskConnectors_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', tc_test_data, 201, tc_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, tc_test_data)
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
+ expected_data = dict(tc_test_data)
expected_data.update(self.test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_task_connectors_DELETE(self):
+ tc_test_data = TaskConnectors_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', tc_test_data, 201, tc_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, tc_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_task_relation_blueprint_CASCADE_behavior_on_inputs_template_deleted(self):
+ tc_test_data = TaskConnectors_test_data()
# POST new item
- url = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', self.test_data_2, 201, self.test_data_2)['url']
+ url = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', tc_test_data, 201, tc_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ GET_and_assert_expected_response(self, url, 200, tc_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.test_data_2['input_of'])
+ DELETE_and_assert_gone(self, tc_test_data['input_of'])
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_task_relation_blueprint_CASCADE_behavior_on_outputs_template_deleted(self):
+ tc_test_data = TaskConnectors_test_data()
# POST new item
- url = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', self.test_data_2, 201, self.test_data_2)['url']
+ url = POST_and_assert_expected_response(self, BASE_URL + '/task_connectors/', tc_test_data, 201, tc_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ GET_and_assert_expected_response(self, url, 200, tc_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.test_data_2['output_of'])
+ DELETE_and_assert_gone(self, tc_test_data['output_of'])
# assert
GET_and_assert_expected_response(self, url, 404, {})
-class DefaultTemplates(unittest.TestCase):
-
- def setUp(self):
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- test_data_1 = {"name": "unique_name_post",
- "template": None,
- "tags": []}
-
- test_data_2 = {"name": "unique_name_protect",
- "template": None,
- "tags": []}
+def DefaultTemplates_test_data(name="defaulttemplate"):
+ return {"name": name,
+ "template": None,
+ "tags": []}
+class DefaultTemplates(unittest.TestCase):
def test_default_generator_template_POST(self):
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/',
- GeneratorTemplateTestCase.test_data_1, 201,
- GeneratorTemplateTestCase.test_data_1)
+ GeneratorTemplate_test_data(), 201,
+ GeneratorTemplate_test_data())
url = r_dict['url']
- test_data_1 = dict(self.test_data_1)
+ test_data_1 = dict(DefaultTemplates_test_data())
test_data_1['template'] = url
POST_and_assert_expected_response(self, BASE_URL + '/default_generator_template/', test_data_1, 201, test_data_1)
def test_default_scheduling_unit_template_POST(self):
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/',
- SchedulingUnitTemplateTestCase.test_data_1, 201,
- SchedulingUnitTemplateTestCase.test_data_1)
+ SchedulingUnitTemplate_test_data(), 201,
+ SchedulingUnitTemplate_test_data())
url = r_dict['url']
- test_data_1 = dict(self.test_data_1)
+ test_data_1 = dict(DefaultTemplates_test_data())
test_data_1['template'] = url
POST_and_assert_expected_response(self, BASE_URL + '/default_scheduling_unit_template/', test_data_1, 201, test_data_1)
def test_default_task_template_POST(self):
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_template/',
- TaskTemplateTestCase.test_data_1, 201,
- TaskTemplateTestCase.test_data_1)
+ TaskTemplate_test_data(), 201,
+ TaskTemplate_test_data())
url = r_dict['url']
- test_data_1 = dict(self.test_data_1)
+ test_data_1 = dict(DefaultTemplates_test_data())
test_data_1['template'] = url
POST_and_assert_expected_response(self, BASE_URL + '/default_task_template/', test_data_1, 201, test_data_1)
def test_default_work_relation_selection_template_POST(self):
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_relation_selection_template/',
- WorkRelationSelectionTemplateTestCase.test_data_1, 201,
- WorkRelationSelectionTemplateTestCase.test_data_1)
+ WorkRelationSelectionTemplate_test_data(), 201,
+ WorkRelationSelectionTemplate_test_data())
url = r_dict['url']
- test_data_1 = dict(self.test_data_1)
+ test_data_1 = dict(DefaultTemplates_test_data())
test_data_1['template'] = url
POST_and_assert_expected_response(self, BASE_URL + '/default_work_relation_selection_template/', test_data_1, 201, test_data_1)
@@ -660,108 +611,93 @@ class DefaultTemplates(unittest.TestCase):
# POST with dependency
template_url = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/',
- GeneratorTemplateTestCase.test_data_1, 201,
- GeneratorTemplateTestCase.test_data_1)['url']
- test_data = dict(self.test_data_2)
+ GeneratorTemplate_test_data(), 201,
+ GeneratorTemplate_test_data())['url']
+ test_data = dict(DefaultTemplates_test_data("defaulttemplate2"))
test_data['template'] = template_url
POST_and_assert_expected_response(self, BASE_URL + '/default_generator_template/',
test_data, 201, 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=self.auth)
+ response = requests.delete(template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
- GET_and_assert_expected_response(self, template_url, 200, GeneratorTemplateTestCase.test_data_1)
+ GET_and_assert_expected_response(self, template_url, 200, GeneratorTemplate_test_data())
def test_default_scheduling_unit_template_PROTECT_behavior_on_template_deleted(self):
# POST with dependency
template_url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_template/',
- SchedulingUnitTemplateTestCase.test_data_1, 201,
- SchedulingUnitTemplateTestCase.test_data_1)['url']
- test_data = dict(self.test_data_2)
+ SchedulingUnitTemplate_test_data(), 201,
+ SchedulingUnitTemplate_test_data())['url']
+ test_data = dict(DefaultTemplates_test_data("defaulttemplate2"))
test_data['template'] = template_url
POST_and_assert_expected_response(self, BASE_URL + '/default_scheduling_unit_template/',
test_data, 201, 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=self.auth)
+ response = requests.delete(template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
- GET_and_assert_expected_response(self, template_url, 200, SchedulingUnitTemplateTestCase.test_data_1)
+ GET_and_assert_expected_response(self, template_url, 200, SchedulingUnitTemplate_test_data())
def test_default_task_template_PROTECT_behavior_on_template_deleted(self):
# POST with dependency
template_url = POST_and_assert_expected_response(self, BASE_URL + '/task_template/',
- TaskTemplateTestCase.test_data_1, 201,
- TaskTemplateTestCase.test_data_1)['url']
- test_data = dict(self.test_data_2)
+ TaskTemplate_test_data(), 201,
+ TaskTemplate_test_data())['url']
+ test_data = dict(DefaultTemplates_test_data("defaulttemplate2"))
test_data['template'] = template_url
POST_and_assert_expected_response(self, BASE_URL + '/default_task_template/',
test_data, 201, 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=self.auth)
+ response = requests.delete(template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
- GET_and_assert_expected_response(self, template_url, 200, TaskTemplateTestCase.test_data_1)
+ GET_and_assert_expected_response(self, template_url, 200, TaskTemplate_test_data())
def test_default_work_relation_selection_template_PROTECT_behavior_on_template_deleted(self):
# POST with dependency
template_url = POST_and_assert_expected_response(self, BASE_URL + '/work_relation_selection_template/',
- WorkRelationSelectionTemplateTestCase.test_data_1, 201,
- WorkRelationSelectionTemplateTestCase.test_data_1)['url']
- test_data = dict(self.test_data_2)
+ WorkRelationSelectionTemplate_test_data(), 201,
+ WorkRelationSelectionTemplate_test_data())['url']
+ test_data = dict(DefaultTemplates_test_data("defaulttemplate2"))
test_data['template'] = template_url
POST_and_assert_expected_response(self, BASE_URL + '/default_work_relation_selection_template/',
test_data, 201, 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=self.auth)
+ response = requests.delete(template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
- GET_and_assert_expected_response(self, template_url, 200, WorkRelationSelectionTemplateTestCase.test_data_1)
-
+ GET_and_assert_expected_response(self, template_url, 200, WorkRelationSelectionTemplate_test_data())
-class CycleTestCase(unittest.TestCase):
- def setUp(self):
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
+def Cycle_test_data(description="my cycle description"):
+ return {"name": 'my_cycle_' + str(uuid.uuid4()),
+ "description": description,
+ "tags": [],
+ "start": datetime.utcnow().isoformat(),
+ "stop": datetime.utcnow().isoformat(),
+ "number": 1,
+ "standard_hours": 2,
+ "expert_hours": 3,
+ "filler_hours": 4,
+ "projects": []}
- # test data
- self.test_data_1 = {"name": 'my_cycle_' + str(uuid.uuid4()),
- "description": "my one cycle",
- "tags": [],
- "start": datetime.utcnow().isoformat(),
- "stop": datetime.utcnow().isoformat(),
- "number": 1,
- "standard_hours": 2,
- "expert_hours": 3,
- "filler_hours": 4,
- "projects": []}
-
- self.test_data_2 = {"name": 'my_cycle_' + str(uuid.uuid4()),
- "description": "This is my other cycle",
- "tags": ['othercycle'],
- "start": datetime.utcnow().isoformat(),
- "stop": datetime.utcnow().isoformat(),
- "number": 4,
- "standard_hours": 3,
- "expert_hours": 2,
- "filler_hours": 1,
- "projects": []}
-
- self.test_patch = {"start": datetime(year=2015, month=10, day=21).isoformat()}
+class CycleTestCase(unittest.TestCase):
+ test_patch = {"start": datetime(year=2015, month=10, day=21).isoformat()}
def test_cycle_list_apiformat(self):
- r = requests.get(BASE_URL + '/cycle/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/cycle/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Cycle List" in r.content.decode('utf8'))
@@ -771,79 +707,69 @@ class CycleTestCase(unittest.TestCase):
def test_cycle_POST_and_GET(self):
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', self.test_data_1, 201, self.test_data_1)
+ cycle_test_data = Cycle_test_data()
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', cycle_test_data, 201, cycle_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, cycle_test_data)
def test_cycle_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/cycle/9876789876/', self.test_data_1, 404, {})
+ PUT_and_assert_expected_response(self, BASE_URL + '/cycle/9876789876/', Cycle_test_data(), 404, {})
def test_cycle_PUT(self):
+ cycle_test_data = Cycle_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', cycle_test_data, 201, cycle_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, cycle_test_data)
# PUT new values, verify
- test_data = dict(self.test_data_2)
- test_data['name'] = self.test_data_1['name'] # since name is PK, need to keep that unchanged
+ test_data = dict(Cycle_test_data("other description"))
+ test_data['name'] = cycle_test_data['name'] # since name is PK, need to keep that unchanged
PUT_and_assert_expected_response(self, url, test_data, 200, test_data)
GET_and_assert_expected_response(self, url, 200, test_data)
def test_cycle_PATCH(self):
+ cycle_test_data = Cycle_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', cycle_test_data, 201, cycle_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, cycle_test_data)
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
+ expected_data = dict(cycle_test_data)
expected_data.update(self.test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_cycle_DELETE(self):
+ cycle_test_data = Cycle_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', cycle_test_data, 201, cycle_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, cycle_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
-class ProjectTestCase(unittest.TestCase):
-
- def setUp(self):
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
+def Project_test_data(description="my project description"):
+ return {"name": 'my_project_' + str(uuid.uuid4()),
+ "description": description,
+ "tags": [],
+ "priority": 1,
+ "can_trigger": False,
+ "private_data": True}
- # test data
- self.test_data_1 = {"name": 'my_project_' + str(uuid.uuid4()),
- "description": "This is my project",
- "tags": [],
- "priority": 1,
- "can_trigger": False,
- "private_data": True}
-
- self.test_data_2 = {"name": 'my_project_' + str(uuid.uuid4()),
- "description": "This is my other project",
- "tags": ['otherproject'],
- "priority": 42,
- "can_trigger": True,
- "private_data": False}
-
- self.test_patch = {"priority": 500,
- "tags": ["SUPERIMPORTANT"]}
-
- self.ctc = CycleTestCase()
- self.ctc.setUp()
+class ProjectTestCase(unittest.TestCase):
+ test_patch = {"priority": 500,
+ "tags": ["SUPERIMPORTANT"]}
def test_project_list_apiformat(self):
- r = requests.get(BASE_URL + '/project/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/project/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Project List" in r.content.decode('utf8'))
@@ -851,47 +777,51 @@ class ProjectTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/project/1234321/', 404, {})
def test_project_POST_and_GET(self):
+ project_test_data = Project_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', project_test_data, 201, project_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, project_test_data)
def test_project_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/project/9876789876/', self.test_data_1, 404, {})
+ PUT_and_assert_expected_response(self, BASE_URL + '/project/9876789876/', Project_test_data(), 404, {})
def test_project_PUT(self):
+ project_test_data = Project_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', project_test_data, 201, project_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, project_test_data)
# PUT new values, verify
- test_data = dict(self.test_data_2)
- test_data['name'] = self.test_data_1['name'] # since name is PK, need to keep that unchanged
+ test_data = dict(Project_test_data("other description"))
+ test_data['name'] = project_test_data['name'] # since name is PK, need to keep that unchanged
PUT_and_assert_expected_response(self, url, test_data, 200, test_data)
GET_and_assert_expected_response(self, url, 200, test_data)
def test_project_PATCH(self):
+ project_test_data = Project_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', project_test_data, 201, project_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, project_test_data)
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
+ expected_data = dict(project_test_data)
expected_data.update(self.test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_project_DELETE(self):
+ project_test_data = Project_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', project_test_data, 201, project_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, project_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
@@ -899,8 +829,9 @@ class ProjectTestCase(unittest.TestCase):
def test_project_PROTECT_behavior_on_cycle_deleted(self):
# POST new item with dependencies
- cycle_url = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', self.ctc.test_data_1, 201, self.ctc.test_data_1)['url']
- test_data = dict(self.test_data_1)
+ cycle_test_data = Cycle_test_data()
+ cycle_url = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', cycle_test_data, 201, cycle_test_data)['url']
+ test_data = dict(Project_test_data())
test_data['cycle'] = cycle_url
url = POST_and_assert_expected_response(self, BASE_URL + '/project/', test_data, 201, test_data)['url']
@@ -908,63 +839,37 @@ class ProjectTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, test_data)
# add project reference to cycle test data (we make Django add that to the cycle in serializer)
- cycle_test_data = dict(self.ctc.test_data_1)
cycle_test_data['projects'] = [url] # add the
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
- response = requests.delete(cycle_url, auth=self.auth)
+ response = requests.delete(cycle_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_and_assert_expected_response(self, cycle_url, 200, cycle_test_data)
+def SchedulingSet_test_data(name="my_scheduling_set", project_url=None, generator_template_url=None):
+ if project_url is None:
+ project_url = post_data_and_get_url(Project_test_data(), '/project/')
-class SchedulingSetTestCase(unittest.TestCase):
+ if generator_template_url is None:
+ generator_template_url = post_data_and_get_url(GeneratorTemplate_test_data(), '/generator_template/')
- def setUp(self):
+ return {"name": name,
+ "description": "This is my scheduling set",
+ "tags": [],
+ "generator_doc": "{}",
+ "project": project_url,
+ "generator_template": generator_template_url,
+ "generator_source": None,
+ "scheduling_unit_drafts": []}
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- # related items
-
- self.ptc = ProjectTestCase()
- self.ptc.setUp()
-
- self.project_url = POST_and_assert_expected_response(self,
- BASE_URL + '/project/',
- self.ptc.test_data_1, 201,
- self.ptc.test_data_1)['url']
-
- self.template_url = POST_and_assert_expected_response(self,
- BASE_URL + '/generator_template/',
- GeneratorTemplateTestCase.test_data_1, 201,
- GeneratorTemplateTestCase.test_data_1)['url']
-
- # test data
- self.test_data_1 = {"name": 'my_scheduling_set',
- "description": "This is my scheduling set",
- "tags": [],
- "generator_doc": "{}",
- "project": self.project_url,
- "generator_template": None,
- "generator_source": None,
- "scheduling_unit_drafts": []}
-
- # test data
- self.test_data_2 = {"name": 'my_other_scheduling_set',
- "description": "This is my other scheduling set",
- "tags": ['otherschedulingset'],
- "generator_doc": "{}",
- "project": self.project_url,
- "generator_template": self.template_url,
- "generator_source": None,
- "scheduling_unit_drafts": []}
-
- self.test_patch = {"description": "This is a new and improved description",
+class SchedulingSetTestCase(unittest.TestCase):
+ test_patch = {"description": "This is a new and improved description",
"generator_doc": "{'para': 'meter'}"}
def test_scheduling_set_list_apiformat(self):
- r = requests.get(BASE_URL + '/scheduling_set/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/scheduling_set/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Scheduling Set List" in r.content.decode('utf8'))
@@ -972,134 +877,117 @@ class SchedulingSetTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/scheduling_set/1234321/', 404, {})
def test_scheduling_set_POST_and_GET(self):
+ schedulingset_test_data = SchedulingSet_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', schedulingset_test_data, 201, schedulingset_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, schedulingset_test_data)
def test_scheduling_set_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_set/9876789876/', self.test_data_1, 404, {})
+ schedulingset_test_data = SchedulingSet_test_data()
+ PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_set/9876789876/', schedulingset_test_data, 404, {})
def test_scheduling_set_PUT(self):
+ project_url = post_data_and_get_url(Project_test_data(), '/project/')
+ schedulingset_test_data = SchedulingSet_test_data(project_url=project_url)
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', schedulingset_test_data, 201, schedulingset_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, schedulingset_test_data)
+ schedulingset_test_data2 = SchedulingSet_test_data("schedulingset2", project_url=project_url)
# PUT new values, verify
- PUT_and_assert_expected_response(self, url, self.test_data_2, 200, self.test_data_2)
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ PUT_and_assert_expected_response(self, url, schedulingset_test_data2, 200, schedulingset_test_data2)
+ GET_and_assert_expected_response(self, url, 200, schedulingset_test_data2)
def test_scheduling_set_PATCH(self):
+ schedulingset_test_data = SchedulingSet_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', schedulingset_test_data, 201, schedulingset_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, schedulingset_test_data)
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
+ expected_data = dict(schedulingset_test_data)
expected_data.update(self.test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_scheduling_set_DELETE(self):
+ schedulingset_test_data = SchedulingSet_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', schedulingset_test_data, 201, schedulingset_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, schedulingset_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_scheduling_set_PROTECT_behavior_on_project_deleted(self):
+ project_url = post_data_and_get_url(Project_test_data(), '/project/')
+ project_test_data = GET_and_assert_expected_response(self, project_url, 200, {})
+ schedulingset_test_data = SchedulingSet_test_data(project_url=project_url)
# POST new item
- url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', self.test_data_1, 201, self.test_data_1)['url']
+ url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', schedulingset_test_data, 201, schedulingset_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, schedulingset_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(self.project_url, auth=self.auth)
+ response = requests.delete(project_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
- GET_and_assert_expected_response(self, self.project_url, 200, self.ptc.test_data_1)
+ GET_and_assert_expected_response(self, project_url, 200, project_test_data)
def test_scheduling_set_SET_NULL_behavior_on_generator_template_deleted(self):
+ generator_template_url = post_data_and_get_url(GeneratorTemplate_test_data(), '/generator_template/')
+ schedulingset_test_data = SchedulingSet_test_data(generator_template_url=generator_template_url)
# POST new item
- test_data = dict(self.test_data_2)
+ test_data = dict(schedulingset_test_data)
url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_set/', test_data, 201, test_data)['url']
# verify
GET_and_assert_expected_response(self, url, 200, test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.template_url)
+ DELETE_and_assert_gone(self, generator_template_url)
# assert
test_data['generator_template'] = None
GET_and_assert_expected_response(self, url, 200, test_data)
+def SchedulingUnitDraft_test_data(name="my_scheduling_unit_draft", scheduling_set_url=None, template_url=None):
+ if scheduling_set_url is None:
+ scheduling_set_url = post_data_and_get_url(SchedulingSet_test_data(), '/scheduling_set/')
+
+ if template_url is None:
+ template_url = post_data_and_get_url(SchedulingUnitTemplate_test_data(), '/scheduling_unit_template/')
+
+ return {"name": name,
+ "description": "This is my run draft",
+ "tags": [],
+ "requirements_doc": "{}",
+ "copy_reason": BASE_URL + '/copy_reason/template/',
+ "generator_instance_doc": "{}",
+ "copies": None,
+ "scheduling_set": scheduling_set_url,
+ "requirements_template": template_url,
+ "related_scheduling_unit_blueprint": [],
+ "task_drafts": []}
class SchedulingUnitDraftTestCase(unittest.TestCase):
-
- def setUp(self):
-
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- # related items
-
- rstc = SchedulingSetTestCase()
- rstc.setUp()
-
- self.scheduling_set_url = POST_and_assert_expected_response(self,
- BASE_URL + '/scheduling_set/',
- rstc.test_data_1, 201,
- rstc.test_data_1)['url']
-
- self.template_url = POST_and_assert_expected_response(self,
- BASE_URL + '/scheduling_unit_template/',
- SchedulingUnitTemplateTestCase.test_data_1, 201,
- SchedulingUnitTemplateTestCase.test_data_1)['url']
-
- # test data
-
- self.test_data_1 = {"name": 'my_scheduling_unit_draft',
- "description": "This is my run draft",
- "tags": [],
- "requirements_doc": "{}",
- "copy_reason": BASE_URL + '/copy_reason/template/',
- "generator_instance_doc": "{}",
- "copies": None,
- "scheduling_set": self.scheduling_set_url,
- "requirements_template": self.template_url,
- "related_scheduling_unit_blueprint": [],
- "task_drafts": []}
-
- # test data
- self.test_data_2 = {"name": 'my_other_scheduling_unit_draft',
- "description": "This is my other run draft",
- "tags": [],
- "requirements_doc": "{}",
- "copy_reason": BASE_URL + '/copy_reason/repeated/',
- "generator_instance_doc": "{}",
- "copies": None,
- "scheduling_set": self.scheduling_set_url,
- "requirements_template": self.template_url,
- "related_scheduling_unit_blueprint": [],
- "task_drafts": []}
-
- self.test_patch = {"description": "This is a new and improved description",
- "requirements_doc": "{'para': 'meter'}"}
+ test_patch = {"description": "This is a new and improved description",
+ "requirements_doc": "{'para': 'meter'}"}
def test_scheduling_unit_draft_list_apiformat(self):
- r = requests.get(BASE_URL + '/scheduling_unit_draft/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/scheduling_unit_draft/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Scheduling Unit Draft List" in r.content.decode('utf8'))
@@ -1107,82 +995,94 @@ class SchedulingUnitDraftTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/1234321/', 404, {})
def test_scheduling_unit_draft_POST_and_GET(self):
+ schedulingunitdraft_test_data = SchedulingUnitDraft_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, schedulingunitdraft_test_data)
def test_scheduling_unit_draft_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/9876789876/', self.test_data_1, 404, {})
+ schedulingunitdraft_test_data = SchedulingUnitDraft_test_data()
+ PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/9876789876/', schedulingunitdraft_test_data, 404, {})
def test_scheduling_unit_draft_PUT(self):
+ schedulingunitdraft_test_data = SchedulingUnitDraft_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, schedulingunitdraft_test_data)
+
+ schedulingunitdraft_test_data2 = SchedulingUnitDraft_test_data("my_scheduling_unit_draft2")
# PUT new values, verify
- PUT_and_assert_expected_response(self, url, self.test_data_2, 200, self.test_data_2)
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ PUT_and_assert_expected_response(self, url, schedulingunitdraft_test_data2, 200, schedulingunitdraft_test_data2)
+ GET_and_assert_expected_response(self, url, 200, schedulingunitdraft_test_data2)
def test_scheduling_unit_draft_PATCH(self):
+ schedulingunitdraft_test_data = SchedulingUnitDraft_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, schedulingunitdraft_test_data)
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
+ expected_data = dict(schedulingunitdraft_test_data)
expected_data.update(self.test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_scheduling_unit_draft_DELETE(self):
+ schedulingunitdraft_test_data = SchedulingUnitDraft_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, schedulingunitdraft_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_scheduling_unit_draft_CASCADE_behavior_on_scheduling_unit_template_deleted(self):
+ template_url = post_data_and_get_url(SchedulingUnitTemplate_test_data(), '/scheduling_unit_template/')
+ schedulingunitdraft_test_data = SchedulingUnitDraft_test_data(template_url=template_url)
# POST new item
- url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', self.test_data_2, 201, self.test_data_2)['url']
+ url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ GET_and_assert_expected_response(self, url, 200, schedulingunitdraft_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.template_url)
+ DELETE_and_assert_gone(self, template_url)
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_scheduling_unit_draft_CASCADE_behavior_on_scheduling_set_deleted(self):
+ scheduling_set_url = post_data_and_get_url(SchedulingSet_test_data(), '/scheduling_set/')
+ schedulingunitdraft_test_data = SchedulingUnitDraft_test_data(scheduling_set_url=scheduling_set_url)
# POST new item
- url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', self.test_data_2, 201, self.test_data_2)['url']
+ url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ GET_and_assert_expected_response(self, url, 200, schedulingunitdraft_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.scheduling_set_url)
+ DELETE_and_assert_gone(self, scheduling_set_url)
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_scheduling_unit_draft_SET_NULL_behavior_on_copies_deleted(self):
+ schedulingunitdraft_test_data = SchedulingUnitDraft_test_data()
# POST new item with dependency
- copy_url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', self.test_data_2, 201, self.test_data_2)['url']
- test_data = dict(self.test_data_1)
+ copy_url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', schedulingunitdraft_test_data, 201, schedulingunitdraft_test_data)['url']
+ test_data = dict(schedulingunitdraft_test_data)
test_data['copies'] = copy_url
url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_draft/', test_data, 201, test_data)['url']
@@ -1197,59 +1097,32 @@ class SchedulingUnitDraftTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, test_data)
-class TaskDraftTestCase(unittest.TestCase):
-
- def setUp(self):
-
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- # related items
-
- rdtc = SchedulingUnitDraftTestCase()
- rdtc.setUp()
-
- self.scheduling_unit_draft_url = POST_and_assert_expected_response(self,
- BASE_URL + '/scheduling_unit_draft/',
- rdtc.test_data_1, 201,
- rdtc.test_data_1)['url']
+def TaskDraft_test_data(name='my_task_draft', scheduling_unit_draft_url=None, template_url=None):
+ if scheduling_unit_draft_url is None:
+ scheduling_unit_draft_url = post_data_and_get_url(SchedulingUnitDraft_test_data(), '/scheduling_unit_draft/')
- self.template_url = POST_and_assert_expected_response(self,
- BASE_URL + '/task_template/',
- TaskTemplateTestCase.test_data_1, 201,
- TaskTemplateTestCase.test_data_1)['url']
+ if template_url is None:
+ template_url = post_data_and_get_url(TaskTemplate_test_data(), '/task_template/')
- # test data
+ return {"name": name,
+ "description": "This is my task draft",
+ "tags": [],
+ "specifications_doc": "{}",
+ "copy_reason": BASE_URL + '/copy_reason/template/',
+ "copies": None,
+ "scheduling_unit_draft": scheduling_unit_draft_url,
+ "specifications_template": template_url,
+ 'related_task_blueprint': [],
+ 'produced_by': [],
+ 'consumed_by': []}
- self.test_data_1 = {"name": 'my_task_draft',
- "description": "This is my work request draft",
- "tags": [],
- "specifications_doc": "{}",
- "copy_reason": BASE_URL + '/copy_reason/template/',
- "copies": None,
- "scheduling_unit_draft": self.scheduling_unit_draft_url,
- "specifications_template": self.template_url,
- 'related_task_blueprint': [],
- 'produced_by': [],
- 'consumed_by': []}
- # test data
- self.test_data_2 = {"name": 'my_other_task_draft',
- "description": "This is my other work request draft",
- "tags": [],
- "specifications_doc": "{}",
- "copy_reason": BASE_URL + '/copy_reason/template/',
- "copies": None,
- "scheduling_unit_draft": self.scheduling_unit_draft_url,
- "specifications_template": self.template_url,
- 'related_task_blueprint': [],
- 'produced_by': [],
- 'consumed_by': []}
-
- self.test_patch = {"description": "This is a new and improved description",
- "specifications_doc": "{'para': 'meter'}"}
+class TaskDraftTestCase(unittest.TestCase):
+ test_patch = {"description": "This is a new and improved description",
+ "specifications_doc": "{'para': 'meter'}"}
def test_task_draft_list_apiformat(self):
- r = requests.get(BASE_URL + '/task_draft/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/task_draft/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Draft List" in r.content.decode('utf8'))
@@ -1257,82 +1130,94 @@ class TaskDraftTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/task_draft/1234321/', 404, {})
def test_task_draft_POST_and_GET(self):
+ taskdraft_test_data = TaskDraft_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data, 201, taskdraft_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, taskdraft_test_data)
def test_task_draft_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/task_draft/9876789876/', self.test_data_1, 404, {})
+ taskdraft_test_data = TaskDraft_test_data()
+ PUT_and_assert_expected_response(self, BASE_URL + '/task_draft/9876789876/', taskdraft_test_data, 404, {})
def test_task_draft_PUT(self):
+ taskdraft_test_data1 = TaskDraft_test_data(name="the one")
+ taskdraft_test_data2 = TaskDraft_test_data(name="the other")
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data1, 201, taskdraft_test_data1)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, taskdraft_test_data1)
# PUT new values, verify
- PUT_and_assert_expected_response(self, url, self.test_data_2, 200, self.test_data_2)
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ PUT_and_assert_expected_response(self, url, taskdraft_test_data2, 200, taskdraft_test_data2)
+ GET_and_assert_expected_response(self, url, 200, taskdraft_test_data2)
def test_task_draft_PATCH(self):
+ taskdraft_test_data = TaskDraft_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data, 201, taskdraft_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, taskdraft_test_data)
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
+ expected_data = dict(taskdraft_test_data)
expected_data.update(self.test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_task_draft_DELETE(self):
+ taskdraft_test_data = TaskDraft_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data, 201, taskdraft_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, taskdraft_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_task_draft_CASCADE_behavior_on_task_template_deleted(self):
+ template_url = post_data_and_get_url(TaskTemplate_test_data(), '/task_template/')
+ taskdraft_test_data = TaskDraft_test_data(name="task draft 2", template_url=template_url)
# POST new item
- url = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', self.test_data_2, 201, self.test_data_2)['url']
+ url = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data, 201, taskdraft_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ GET_and_assert_expected_response(self, url, 200, taskdraft_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.template_url)
+ DELETE_and_assert_gone(self, template_url)
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_task_draft_CASCADE_behavior_on_scheduling_unit_draft_deleted(self):
+ scheduling_unit_draft_url = post_data_and_get_url(SchedulingUnitDraft_test_data(), '/scheduling_unit_draft/')
+ taskdraft_test_data = TaskDraft_test_data(name="task draft 2", scheduling_unit_draft_url=scheduling_unit_draft_url)
# POST new item
- url = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', self.test_data_2, 201, self.test_data_2)['url']
+ url = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data, 201, taskdraft_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ GET_and_assert_expected_response(self, url, 200, taskdraft_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.scheduling_unit_draft_url)
+ DELETE_and_assert_gone(self, scheduling_unit_draft_url)
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_task_draft_SET_NULL_behavior_on_copies_deleted(self):
+ taskdraft_test_data1 = TaskDraft_test_data(name="the one")
+ taskdraft_test_data2 = TaskDraft_test_data(name="the other")
# POST new item with dependency
- copy_url = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', self.test_data_2, 201, self.test_data_2)['url']
- test_data = dict(self.test_data_1)
+ copy_url = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', taskdraft_test_data2, 201, taskdraft_test_data2)['url']
+ test_data = dict(taskdraft_test_data1)
test_data['copies'] = copy_url
url = POST_and_assert_expected_response(self, BASE_URL + '/task_draft/', test_data, 201, test_data)['url']
@@ -1346,74 +1231,37 @@ class TaskDraftTestCase(unittest.TestCase):
test_data['copies'] = None
GET_and_assert_expected_response(self, url, 200, test_data)
+def TaskRelationDraft_test_data(name="myTaskRelationDraft", producer_url=None, consumer_url=None, template_url=None, input_url=None, output_url=None):
+ if producer_url is None:
+ producer_url = post_data_and_get_url(TaskDraft_test_data(), '/task_draft/')
-class TaskRelationDraftTestCase(unittest.TestCase):
+ if consumer_url is None:
+ consumer_url = post_data_and_get_url(TaskDraft_test_data(),'/task_draft/')
- def setUp(self):
+ if template_url is None:
+ template_url = post_data_and_get_url(WorkRelationSelectionTemplate_test_data(), '/work_relation_selection_template/')
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
+ if input_url is None:
+ input_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
- # related items
+ if output_url is None:
+ output_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
- wrdtc = TaskDraftTestCase()
- wrdtc.setUp()
+ return {"tags": [],
+ "selection_doc": "{}",
+ "dataformat": BASE_URL + "/dataformat/Beamformed/",
+ "producer": producer_url,
+ "consumer": consumer_url,
+ "input": input_url,
+ "output": output_url,
+ "selection_template": template_url,
+ 'related_task_relation_blueprint': []}
- wiortc = TaskConnectorsTestCase()
- wiortc.setUp()
-
- self.producer_url = POST_and_assert_expected_response(self,
- BASE_URL + '/task_draft/',
- wrdtc.test_data_1, 201,
- wrdtc.test_data_1)['url']
-
- self.consumer_url = POST_and_assert_expected_response(self,
- BASE_URL + '/task_draft/',
- wrdtc.test_data_2, 201,
- wrdtc.test_data_2)['url']
-
- self.template_url = POST_and_assert_expected_response(self,
- BASE_URL + '/work_relation_selection_template/',
- WorkRelationSelectionTemplateTestCase.test_data_1, 201,
- WorkRelationSelectionTemplateTestCase.test_data_1)['url']
-
- self.input_url = POST_and_assert_expected_response(self,
- BASE_URL + '/task_connectors/',
- wiortc.test_data_1, 201,
- wiortc.test_data_1)['url']
-
- self.output_url = POST_and_assert_expected_response(self,
- BASE_URL + '/task_connectors/',
- wiortc.test_data_1, 201,
- wiortc.test_data_1)['url']
-
-
- # test data
-
- self.test_data_1 = {"tags": [],
- "selection_doc": "{}",
- "dataformat": BASE_URL + "/dataformat/Beamformed/",
- "producer": self.producer_url,
- "consumer": self.consumer_url,
- "input": self.input_url,
- "output": self.output_url,
- "selection_template": self.template_url,
- 'related_task_relation_blueprint': []}
-
- # test data
- self.test_data_2 = {"tags": [],
- "selection_doc": "{}",
- "dataformat": BASE_URL + "/dataformat/Beamformed/",
- "producer": self.producer_url,
- "consumer": self.consumer_url,
- "input": self.input_url,
- "output": self.output_url,
- "selection_template": self.template_url,
- 'related_task_relation_blueprint': []}
-
- self.test_patch = {"selection_doc": "{'para': 'meter'}"}
+class TaskRelationDraftTestCase(unittest.TestCase):
+ test_patch = {"selection_doc": "{'para': 'meter'}"}
def test_task_relation_draft_list_apiformat(self):
- r = requests.get(BASE_URL + '/task_relation_draft/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/task_relation_draft/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Relation Draft List" in r.content.decode('utf8'))
@@ -1421,171 +1269,164 @@ class TaskRelationDraftTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/1234321/', 404, {})
def test_task_relation_draft_POST_and_GET(self):
+ trd_test_data = TaskRelationDraft_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/', trd_test_data, 201, trd_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trd_test_data)
def test_task_relation_draft_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/9876789876/', self.test_data_1, 404, {})
+ trd_test_data = TaskRelationDraft_test_data()
+ PUT_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/9876789876/', trd_test_data, 404, {})
def test_task_relation_draft_PUT(self):
+ trd_test_data1 = TaskRelationDraft_test_data()
+ trd_test_data2 = TaskRelationDraft_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/', trd_test_data1, 201, trd_test_data1)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trd_test_data1)
# PUT new values, verify
- PUT_and_assert_expected_response(self, url, self.test_data_2, 200, self.test_data_2)
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ PUT_and_assert_expected_response(self, url, trd_test_data2, 200, trd_test_data2)
+ GET_and_assert_expected_response(self, url, 200, trd_test_data2)
def test_task_relation_draft_PATCH(self):
+ trd_test_data = TaskRelationDraft_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/', trd_test_data, 201, trd_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trd_test_data)
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
+ expected_data = dict(trd_test_data)
expected_data.update(self.test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_task_relation_draft_DELETE(self):
+ trd_test_data = TaskRelationDraft_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/', trd_test_data, 201, trd_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trd_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_task_relation_draft_CASCADE_behavior_on_work_relation_selection_template_deleted(self):
+ template_url = post_data_and_get_url(WorkRelationSelectionTemplate_test_data(), '/work_relation_selection_template/')
+ trd_test_data = TaskRelationDraft_test_data(template_url=template_url)
# POST new item
- url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/', self.test_data_2, 201, self.test_data_2)['url']
+ url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/', trd_test_data, 201, trd_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ GET_and_assert_expected_response(self, url, 200, trd_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.template_url)
+ DELETE_and_assert_gone(self, template_url)
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_task_relation_draft_CASCADE_behavior_on_producer_deleted(self):
+ producer_url = post_data_and_get_url(TaskDraft_test_data(), '/task_draft/')
+ trd_test_data = TaskRelationDraft_test_data(producer_url=producer_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/',
- self.test_data_1, 201, self.test_data_1)['url']
+ trd_test_data, 201, trd_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trd_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.producer_url)
+ DELETE_and_assert_gone(self, producer_url)
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_task_relation_draft_CASCADE_behavior_on_consumer_deleted(self):
+ consumer_url = post_data_and_get_url(TaskDraft_test_data(), '/task_draft/')
+ trd_test_data = TaskRelationDraft_test_data(consumer_url=consumer_url)
# POST new item with dependency
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/',
- self.test_data_1, 201, self.test_data_1)['url']
+ trd_test_data, 201, trd_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trd_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.consumer_url)
+ DELETE_and_assert_gone(self, consumer_url)
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_task_relation_draft_CASCADE_behavior_on_input_deleted(self):
+ input_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
+ trd_test_data = TaskRelationDraft_test_data(input_url=input_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/',
- self.test_data_1, 201, self.test_data_1)['url']
+ trd_test_data, 201, trd_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trd_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.input_url)
+ DELETE_and_assert_gone(self, input_url)
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_task_relation_draft_CASCADE_behavior_on_output_deleted(self):
+ output_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
+ trd_test_data = TaskRelationDraft_test_data(output_url=output_url)
# POST new item with dependency
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_draft/',
- self.test_data_1, 201, self.test_data_1)['url']
+ trd_test_data, 201, trd_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trd_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.output_url)
+ DELETE_and_assert_gone(self, output_url)
# assert
GET_and_assert_expected_response(self, url, 404, {})
-class SchedulingUnitBlueprintTestCase(unittest.TestCase):
-
- def setUp(self):
-
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
+def SchedulingUnitBlueprint_test_data(name="my_scheduling_unit_blueprint", scheduling_unit_draft_url=None, template_url=None):
+ if scheduling_unit_draft_url is None:
+ scheduling_unit_draft_url = post_data_and_get_url(SchedulingUnitDraft_test_data(), '/scheduling_unit_draft/')
- # related items
+ if template_url is None:
+ template_url = post_data_and_get_url(SchedulingUnitTemplate_test_data(), '/scheduling_unit_template/')
- rdtc = SchedulingUnitDraftTestCase()
- rdtc.setUp()
+ return {"name": name,
+ "description": "This is my run blueprint",
+ "tags": [],
+ "requirements_doc": "{}",
+ "do_cancel": False,
+ "draft": scheduling_unit_draft_url,
+ "requirements_template": template_url}
- scheduling_unit_draft_url = POST_and_assert_expected_response(self,
- BASE_URL + '/scheduling_unit_draft/',
- rdtc.test_data_1, 201,
- rdtc.test_data_1)['url']
-
- template_url = POST_and_assert_expected_response(self,
- BASE_URL + '/scheduling_unit_template/',
- SchedulingUnitTemplateTestCase.test_data_1, 201,
- SchedulingUnitTemplateTestCase.test_data_1)['url']
-
- # test data
- self.test_data_1 = {"name": 'my_scheduling_unit_blueprint',
- "description": "This is my run blueprint",
- "tags": [],
- "requirements_doc": "{}",
- "do_cancel": False,
- "draft": scheduling_unit_draft_url,
- "requirements_template": template_url}
-
- # test data
- self.test_data_2 = {"name": 'my_other_scheduling_unit_blueprint',
- "description": "This is my other run blueprint",
- "tags": [],
- "requirements_doc": "{}",
- "do_cancel": True,
- "draft": scheduling_unit_draft_url,
- "requirements_template": template_url}
-
- self.test_patch = {"description": "This is an updated description",
- "do_cancel": True}
+class SchedulingUnitBlueprintTestCase(unittest.TestCase):
+ test_patch = {"description": "This is an updated description",
+ "do_cancel": True}
def test_scheduling_unit_blueprint_list_apiformat(self):
- r = requests.get(BASE_URL + '/scheduling_unit_blueprint/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/scheduling_unit_blueprint/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Scheduling Unit Blueprint List" in r.content.decode('utf8'))
@@ -1593,139 +1434,113 @@ class SchedulingUnitBlueprintTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/1234321/', 404, {})
def test_scheduling_unit_blueprint_POST_and_GET(self):
+ sub_test_data = SchedulingUnitBlueprint_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', sub_test_data, 201, sub_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, sub_test_data)
def test_scheduling_unit_blueprint_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/9876789876/', self.test_data_1, 404, {})
+ sub_test_data = SchedulingUnitBlueprint_test_data()
+ PUT_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/9876789876/', sub_test_data, 404, {})
def test_scheduling_unit_blueprint_PUT(self):
+ sub_test_data1 = SchedulingUnitBlueprint_test_data(name="the one")
+ sub_test_data2 = SchedulingUnitBlueprint_test_data(name="the other")
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', sub_test_data1, 201, sub_test_data1)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, sub_test_data1)
# PUT new values, verify
- PUT_and_assert_expected_response(self, url, self.test_data_2, 200, self.test_data_2)
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ PUT_and_assert_expected_response(self, url, sub_test_data2, 200, sub_test_data2)
+ GET_and_assert_expected_response(self, url, 200, sub_test_data2)
def test_scheduling_unit_blueprint_PATCH(self):
+ sub_test_data = SchedulingUnitBlueprint_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', sub_test_data, 201, sub_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, sub_test_data)
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
+ expected_data = dict(sub_test_data)
expected_data.update(self.test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_scheduling_unit_blueprint_DELETE(self):
+ sub_test_data = SchedulingUnitBlueprint_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', sub_test_data, 201, sub_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, sub_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_scheduling_unit_blueprint_CASCADE_behavior_on_scheduling_unit_template_deleted(self):
+ sub_test_data = SchedulingUnitBlueprint_test_data()
# POST new item
- url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', self.test_data_2, 201, self.test_data_2)['url']
+ url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', sub_test_data, 201, sub_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ GET_and_assert_expected_response(self, url, 200, sub_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.test_data_2['requirements_template'])
+ DELETE_and_assert_gone(self, sub_test_data['requirements_template'])
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_scheduling_unit_blueprint_CASCADE_behavior_on_scheduling_unit_draft_deleted(self):
+ sub_test_data = SchedulingUnitBlueprint_test_data()
# POST new item
- url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', self.test_data_2, 201, self.test_data_2)['url']
+ url = POST_and_assert_expected_response(self, BASE_URL + '/scheduling_unit_blueprint/', sub_test_data, 201, sub_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ GET_and_assert_expected_response(self, url, 200, sub_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.test_data_2['draft'])
+ DELETE_and_assert_gone(self, sub_test_data['draft'])
# assert
GET_and_assert_expected_response(self, url, 404, {})
+def TaskBlueprint_test_data(name="my_TaskBlueprint", draft_url=None, template_url=None, scheduling_unit_blueprint_url=None):
+ if draft_url is None:
+ draft_url = post_data_and_get_url(TaskDraft_test_data(), '/task_draft/')
-class TaskBlueprintTestCase(unittest.TestCase):
-
- def setUp(self):
-
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- # related items
-
- wrdtc = TaskDraftTestCase()
- wrdtc.setUp()
+ if template_url is None:
+ template_url = post_data_and_get_url(TaskTemplate_test_data(), '/task_template/')
- rbtc = SchedulingUnitBlueprintTestCase()
- rbtc.setUp()
+ if scheduling_unit_blueprint_url is None:
+ scheduling_unit_blueprint_url = post_data_and_get_url(SchedulingUnitBlueprint_test_data(), '/scheduling_unit_blueprint/')
+ return {"name": name,
+ "description": "This is my work request blueprint",
+ "tags": [],
+ "specifications_doc": "{}",
+ "do_cancel": False,
+ "draft": draft_url,
+ "specifications_template": template_url,
+ "scheduling_unit_blueprint": scheduling_unit_blueprint_url,
+ "subtasks": [],
+ "produced_by": [],
+ "consumed_by": []}
- draft_url = POST_and_assert_expected_response(self,
- BASE_URL + '/task_draft/',
- wrdtc.test_data_1, 201,
- wrdtc.test_data_1)['url']
-
- template_url = POST_and_assert_expected_response(self,
- BASE_URL + '/task_template/',
- TaskTemplateTestCase.test_data_1, 201,
- TaskTemplateTestCase.test_data_1)['url']
-
- scheduling_unit_blueprint_url = POST_and_assert_expected_response(self,
- BASE_URL + '/scheduling_unit_blueprint/',
- rbtc.test_data_1, 201,
- rbtc.test_data_1)['url']
-
- # test data
- self.test_data_1 = {"name": 'my_task_blueprint',
- "description": "This is my work request blueprint",
- "tags": [],
- "specifications_doc": "{}",
- "do_cancel": False,
- "draft": draft_url,
- "specifications_template": template_url,
- "scheduling_unit_blueprint": scheduling_unit_blueprint_url,
- "subtasks": [],
- "produced_by": [],
- "consumed_by": []}
-
- # test data
- self.test_data_2 = {"name": 'my_other_task_blueprin', # <- Missing a 't'? Well, varchar(30) is shorter than you'd think...
- "description": "This is my other work request blueprint",
- "tags": [],
- "specifications_doc": "{}",
- "do_cancel": True,
- "draft": draft_url,
- "specifications_template": template_url,
- "scheduling_unit_blueprint": scheduling_unit_blueprint_url,
- "subtasks": [],
- "produced_by": [],
- "consumed_by": []}
-
- self.test_patch = {"description": "This is an updated description",
- "do_cancel": True}
+class TaskBlueprintTestCase(unittest.TestCase):
+ test_patch = {"description": "This is an updated description",
+ "do_cancel": True}
def test_task_blueprint_list_apiformat(self):
- r = requests.get(BASE_URL + '/task_blueprint/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/task_blueprint/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Blueprint List" in r.content.decode('utf8'))
@@ -1733,53 +1548,60 @@ class TaskBlueprintTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/task_blueprint/1234321/', 404, {})
def test_task_blueprint_POST_and_GET(self):
+ tb_test_data = TaskBlueprint_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, tb_test_data)
def test_task_blueprint_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/task_blueprint/9876789876/', self.test_data_1, 404, {})
+ tb_test_data = TaskBlueprint_test_data()
+ PUT_and_assert_expected_response(self, BASE_URL + '/task_blueprint/9876789876/', tb_test_data, 404, {})
def test_task_blueprint_PUT(self):
+ tb_test_data1 = TaskBlueprint_test_data(name="the one")
+ tb_test_data2 = TaskBlueprint_test_data(name="the other")
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data1, 201, tb_test_data1)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, tb_test_data1)
# PUT new values, verify
- PUT_and_assert_expected_response(self, url, self.test_data_2, 200, self.test_data_2)
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ PUT_and_assert_expected_response(self, url, tb_test_data2, 200, tb_test_data2)
+ GET_and_assert_expected_response(self, url, 200, tb_test_data2)
def test_task_blueprint_PATCH(self):
+ tb_test_data = TaskBlueprint_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, tb_test_data)
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
+ expected_data = dict(tb_test_data)
expected_data.update(self.test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_task_blueprint_DELETE(self):
+ tb_test_data = TaskBlueprint_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, tb_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_task_blueprint_prevents_missing_specification_template(self):
+ tb_test_data = TaskBlueprint_test_data()
# test data
- test_data = dict(self.test_data_1)
+ test_data = dict(tb_test_data)
test_data['specifications_template'] = None
# POST invalid data and assert response
@@ -1787,9 +1609,10 @@ class TaskBlueprintTestCase(unittest.TestCase):
self.assertTrue('This field may not be null' in str(r_dict['specifications_template']))
def test_task_blueprint_prevents_missing_draft(self):
+ tb_test_data = TaskBlueprint_test_data()
# test data
- test_data = dict(self.test_data_1)
+ test_data = dict(tb_test_data)
test_data['draft'] = None
# POST invalid data and assert response
@@ -1797,9 +1620,10 @@ class TaskBlueprintTestCase(unittest.TestCase):
self.assertTrue('This field may not be null' in str(r_dict['draft']))
def test_task_blueprint_prevents_missing_scheduling_unit_blueprint(self):
+ tb_test_data = TaskBlueprint_test_data()
# test data
- test_data = dict(self.test_data_1)
+ test_data = dict(tb_test_data)
test_data['scheduling_unit_blueprint'] = None
# POST invalid data and assert response
@@ -1807,121 +1631,86 @@ class TaskBlueprintTestCase(unittest.TestCase):
self.assertTrue('This field may not be null' in str(r_dict['scheduling_unit_blueprint']))
def test_task_blueprint_CASCADE_behavior_on_task_template_deleted(self):
+ tb_test_data = TaskBlueprint_test_data()
# POST new item
- url = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', self.test_data_2, 201, self.test_data_2)['url']
+ url = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ GET_and_assert_expected_response(self, url, 200, tb_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.test_data_2['specifications_template'])
+ DELETE_and_assert_gone(self, tb_test_data['specifications_template'])
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_task_blueprint_CASCADE_behavior_on_task_draft_deleted(self):
+ tb_test_data = TaskBlueprint_test_data()
# POST new item
- url = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', self.test_data_2, 201, self.test_data_2)['url']
+ url = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ GET_and_assert_expected_response(self, url, 200, tb_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.test_data_2['draft'])
+ DELETE_and_assert_gone(self, tb_test_data['draft'])
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_task_blueprint_CASCADE_behavior_on_scheduling_unit_blueprint_deleted(self):
+ tb_test_data = TaskBlueprint_test_data()
# POST new item
- url = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', self.test_data_2, 201, self.test_data_2)['url']
+ url = POST_and_assert_expected_response(self, BASE_URL + '/task_blueprint/', tb_test_data, 201, tb_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ GET_and_assert_expected_response(self, url, 200, tb_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.test_data_2['scheduling_unit_blueprint'])
+ DELETE_and_assert_gone(self, tb_test_data['scheduling_unit_blueprint'])
# assert
GET_and_assert_expected_response(self, url, 404, {})
+def TaskRelationBlueprint_test_data(draft_url=None, template_url=None, input_url=None, output_url=None, consumer_url=None, producer_url=None):
+ if draft_url is None:
+ draft_url = post_data_and_get_url(TaskRelationDraft_test_data(), '/task_relation_draft/')
-class TaskRelationBlueprintTestCase(unittest.TestCase):
-
- def setUp(self):
-
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- # related items
-
- wrrdtc = TaskRelationDraftTestCase()
- wrrdtc.setUp()
+ if producer_url is None:
+ producer_url = post_data_and_get_url(TaskBlueprint_test_data(), '/task_blueprint/')
- wiortc = TaskConnectorsTestCase()
- wiortc.setUp()
+ if consumer_url is None:
+ consumer_url = post_data_and_get_url(TaskBlueprint_test_data(),'/task_blueprint/')
- wrbtc = TaskBlueprintTestCase()
- wrbtc.setUp()
+ if template_url is None:
+ template_url = post_data_and_get_url(WorkRelationSelectionTemplate_test_data(), '/work_relation_selection_template/')
- self.draft_url = POST_and_assert_expected_response(self,
- BASE_URL + '/task_relation_draft/',
- wrrdtc.test_data_1, 201,
- wrrdtc.test_data_1)['url']
+ if input_url is None:
+ input_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
- self.template_url = POST_and_assert_expected_response(self,
- BASE_URL + '/work_relation_selection_template/',
- WorkRelationSelectionTemplateTestCase.test_data_1, 201,
- WorkRelationSelectionTemplateTestCase.test_data_1)['url']
+ if output_url is None:
+ output_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
- self.input_url = POST_and_assert_expected_response(self,
- BASE_URL + '/task_connectors/',
- wiortc.test_data_1, 201,
- wiortc.test_data_1)['url']
-
- self.output_url = POST_and_assert_expected_response(self,
- BASE_URL + '/task_connectors/',
- wiortc.test_data_2, 201,
- wiortc.test_data_2)['url']
-
- self.consumer_url = POST_and_assert_expected_response(self,
- BASE_URL + '/task_blueprint/',
- wrbtc.test_data_1, 201,
- wrbtc.test_data_1)['url']
+ # test data
+ return {"tags": [],
+ "selection_doc": "{}",
+ "dataformat": BASE_URL + '/dataformat/MeasurementSet/',
+ "input": input_url,
+ "output": output_url,
+ "draft": draft_url,
+ "selection_template": template_url,
+ "producer": producer_url,
+ "consumer": consumer_url}
- self.producer_url = POST_and_assert_expected_response(self,
- BASE_URL + '/task_blueprint/',
- wrbtc.test_data_2, 201,
- wrbtc.test_data_2)['url']
- # test data
- self.test_data_1 = {"tags": [],
- "selection_doc": "{}",
- "dataformat": BASE_URL + '/dataformat/MeasurementSet/',
- "input": self.input_url,
- "output": self.output_url,
- "draft": self.draft_url,
- "selection_template": self.template_url,
- "producer": self.producer_url,
- "consumer": self.consumer_url}
-
- # test data
- self.test_data_2 = {"tags": ['GUTEN', 'TAG'],
- "selection_doc": "{}",
- "dataformat": BASE_URL + '/dataformat/Beamformed/',
- "input": self.input_url,
- "output": self.output_url,
- "draft": self.draft_url,
- "selection_template": self.template_url,
- "producer": self.producer_url,
- "consumer": self.consumer_url}
-
- self.test_patch = {"selection_doc": "{'new': 'doc'}"}
+class TaskRelationBlueprintTestCase(unittest.TestCase):
+ test_patch = {"selection_doc": "{'new': 'doc'}"}
def test_task_relation_blueprint_list_apiformat(self):
- r = requests.get(BASE_URL + '/task_relation_blueprint/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/task_relation_blueprint/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Task Relation Blueprint List" in r.content.decode('utf8'))
@@ -1929,53 +1718,60 @@ class TaskRelationBlueprintTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/1234321/', 404, {})
def test_task_relation_blueprint_POST_and_GET(self):
+ trb_test_data = TaskRelationBlueprint_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', trb_test_data, 201, trb_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trb_test_data)
def test_task_relation_blueprint_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/9876789876/', self.test_data_1, 404, {})
+ trb_test_data = TaskRelationBlueprint_test_data()
+ PUT_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/9876789876/', trb_test_data, 404, {})
def test_task_relation_blueprint_PUT(self):
+ trb_test_data1 = TaskRelationBlueprint_test_data()
+ trb_test_data2 = TaskRelationBlueprint_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', trb_test_data1, 201, trb_test_data1)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trb_test_data1)
# PUT new values, verify
- PUT_and_assert_expected_response(self, url, self.test_data_2, 200, self.test_data_2)
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ PUT_and_assert_expected_response(self, url, trb_test_data2, 200, trb_test_data2)
+ GET_and_assert_expected_response(self, url, 200, trb_test_data2)
def test_task_relation_blueprint_PATCH(self):
+ trb_test_data = TaskRelationBlueprint_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', trb_test_data, 201, trb_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trb_test_data)
# PATCH item and verify
PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
+ expected_data = dict(trb_test_data)
expected_data.update(self.test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_task_relation_blueprint_DELETE(self):
+ trb_test_data = TaskRelationBlueprint_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', trb_test_data, 201, trb_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trb_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_task_relation_blueprint_prevents_missing_selection_template(self):
+ trb_test_data = TaskRelationBlueprint_test_data()
# test data
- test_data = dict(self.test_data_1)
+ test_data = dict(trb_test_data)
test_data['selection_template'] = None
# POST invalid data and assert response
@@ -1983,9 +1779,10 @@ class TaskRelationBlueprintTestCase(unittest.TestCase):
self.assertTrue('This field may not be null' in str(r_dict['selection_template']))
def test_task_relation_blueprint_prevents_missing_draft(self):
+ trb_test_data = TaskRelationBlueprint_test_data()
# test data
- test_data = dict(self.test_data_1)
+ test_data = dict(trb_test_data)
test_data['draft'] = None
# POST invalid data and assert response
@@ -1993,9 +1790,10 @@ class TaskRelationBlueprintTestCase(unittest.TestCase):
self.assertTrue('This field may not be null' in str(r_dict['draft']))
def test_task_relation_blueprint_prevents_missing_producer(self):
+ trb_test_data = TaskRelationBlueprint_test_data()
# test data
- test_data = dict(self.test_data_1)
+ test_data = dict(trb_test_data)
test_data['producer'] = None
# POST invalid data and assert response
@@ -2003,9 +1801,10 @@ class TaskRelationBlueprintTestCase(unittest.TestCase):
self.assertTrue('This field may not be null' in str(r_dict['producer']))
def test_task_relation_blueprint_prevents_missing_consumer(self):
+ trb_test_data = TaskRelationBlueprint_test_data()
# test data
- test_data = dict(self.test_data_1)
+ test_data = dict(trb_test_data)
test_data['consumer'] = None
# POST invalid data and assert response
@@ -2013,9 +1812,10 @@ class TaskRelationBlueprintTestCase(unittest.TestCase):
self.assertTrue('This field may not be null' in str(r_dict['consumer']))
def test_task_relation_blueprint_prevents_missing_input(self):
+ trb_test_data = TaskRelationBlueprint_test_data()
# test data
- test_data = dict(self.test_data_1)
+ test_data = dict(trb_test_data)
test_data['input'] = None
# POST invalid data and assert response
@@ -2023,9 +1823,10 @@ class TaskRelationBlueprintTestCase(unittest.TestCase):
self.assertTrue('This field may not be null' in str(r_dict['input']))
def test_task_relation_blueprint_prevents_missing_output(self):
+ trb_test_data = TaskRelationBlueprint_test_data()
# test data
- test_data = dict(self.test_data_1)
+ test_data = dict(trb_test_data)
test_data['output'] = None
# POST invalid data and assert response
@@ -2033,78 +1834,88 @@ class TaskRelationBlueprintTestCase(unittest.TestCase):
self.assertTrue('This field may not be null' in str(r_dict['output']))
def test_task_relation_blueprint_CASCADE_behavior_on_work_relation_selection_template_deleted(self):
+ template_url = post_data_and_get_url(WorkRelationSelectionTemplate_test_data(), '/work_relation_selection_template/')
+ trb_test_data = TaskRelationBlueprint_test_data(template_url=template_url)
# POST new item
- url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', self.test_data_2, 201, self.test_data_2)['url']
+ url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/', trb_test_data, 201, trb_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ GET_and_assert_expected_response(self, url, 200, trb_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.template_url)
+ DELETE_and_assert_gone(self, template_url)
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_task_relation_blueprint_CASCADE_behavior_on_producer_deleted(self):
+ producer_url = post_data_and_get_url(TaskBlueprint_test_data(), '/task_blueprint/')
+ trb_test_data = TaskRelationBlueprint_test_data(producer_url=producer_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/',
- self.test_data_1, 201, self.test_data_1)['url']
+ trb_test_data, 201, trb_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trb_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.producer_url)
+ DELETE_and_assert_gone(self, producer_url)
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_task_relation_blueprint_CASCADE_behavior_on_consumer_deleted(self):
+ consumer_url = post_data_and_get_url(TaskBlueprint_test_data(), '/task_blueprint/')
+ trb_test_data = TaskRelationBlueprint_test_data(consumer_url=consumer_url)
# POST new item with dependency
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/',
- self.test_data_1, 201, self.test_data_1)['url']
+ trb_test_data, 201, trb_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trb_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.consumer_url)
+ DELETE_and_assert_gone(self, consumer_url)
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_task_relation_blueprint_CASCADE_behavior_on_input_deleted(self):
+ input_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
+ trb_test_data = TaskRelationBlueprint_test_data(input_url=input_url)
# POST new item
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/',
- self.test_data_1, 201, self.test_data_1)['url']
+ trb_test_data, 201, trb_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trb_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.input_url)
+ DELETE_and_assert_gone(self, input_url)
# assert
GET_and_assert_expected_response(self, url, 404, {})
def test_task_relation_blueprint_CASCADE_behavior_on_output_deleted(self):
+ output_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
+ trb_test_data = TaskRelationBlueprint_test_data(output_url=output_url)
# POST new item with dependency
url = POST_and_assert_expected_response(self, BASE_URL + '/task_relation_blueprint/',
- self.test_data_1, 201, self.test_data_1)['url']
+ trb_test_data, 201, trb_test_data)['url']
# verify
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, trb_test_data)
# DELETE dependency
- DELETE_and_assert_gone(self, self.output_url)
+ DELETE_and_assert_gone(self, output_url)
# assert
GET_and_assert_expected_response(self, url, 404, {})
--
GitLab