From 0177169dfd49dcf2eb1ce4548ec03d4bfca43b12 Mon Sep 17 00:00:00 2001
From: Jorrit Schaap <schaap@astron.nl>
Date: Mon, 3 Feb 2020 12:46:28 +0100
Subject: [PATCH] TMSS-139: factored out the generation of test_data into the
common tmss_test_environment_unittest_setup.py module. Fixed
t_tmssapp_scheduling_functional and t_tmssapp_specification_functional.
---
SAS/TMSS/test/CMakeLists.txt | 2 -
.../test/t_tmssapp_scheduling_functional.py | 1441 ++++++-----------
.../t_tmssapp_specification_functional.py | 146 +-
.../tmss_test_environment_unittest_setup.py | 205 ++-
4 files changed, 805 insertions(+), 989 deletions(-)
diff --git a/SAS/TMSS/test/CMakeLists.txt b/SAS/TMSS/test/CMakeLists.txt
index 7e431cde57e..e45b88a19bd 100644
--- a/SAS/TMSS/test/CMakeLists.txt
+++ b/SAS/TMSS/test/CMakeLists.txt
@@ -17,8 +17,6 @@ if(BUILD_TESTING)
tmss_test_environment_unittest_setup.py
t_tmssapp_specification_django.py
t_tmssapp_scheduling_django.py
- t_tmssapp_specification_functional.py
- t_tmssapp_scheduling_functional.py
DESTINATION lofar/sas/tmss/test)
lofar_add_test(t_tmss_test_database)
diff --git a/SAS/TMSS/test/t_tmssapp_scheduling_functional.py b/SAS/TMSS/test/t_tmssapp_scheduling_functional.py
index 3361e9eb7f6..492a70decf9 100755
--- a/SAS/TMSS/test/t_tmssapp_scheduling_functional.py
+++ b/SAS/TMSS/test/t_tmssapp_scheduling_functional.py
@@ -27,55 +27,20 @@
# todo: behavior in a controlled way.
# todo: We should probably also fully test behavior wrt mandatory and nullable fields.
-import os
import unittest
-import requests
-import json
-from datetime import datetime
-from t_tmssapp_specification_functional import BASE_URL, _call_API_and_assert_expected_response, \
- PUT_and_assert_expected_response, POST_and_assert_expected_response, DELETE_and_assert_gone, \
- GET_and_assert_expected_response, PATCH_and_assert_expected_response, \
- TaskBlueprintTestCase # Note: Importing TestCases here will cause them to be re-run as part of this test module!
-from requests.auth import HTTPBasicAuth
-from lofar.common import dbcredentials
-
-# 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())
-
-class SubtaskTemplateTestCase(unittest.TestCase):
+import logging
+logger = logging.getLogger(__name__)
+logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO)
+# Do Mandatory setup step:
+# use setup/teardown magic for tmss test database, ldap server and django server
+# (ignore pycharm unused import statement, python unittests does use at RunTime the tmss_test_environment_unittest_setup module)
+from lofar.sas.tmss.test.tmss_test_environment_unittest_setup import *
- def setUp(self):
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- # test data
- test_data_1 = {"type": BASE_URL + '/subtask_type/copy/',
- "name": "subtask1",
- "description": 'My one observation',
- "version": 'v0.314159265359',
- "schema": {"mykey": "my value"},
- "realtime": True,
- "queue": False,
- "tags": ["TMSS", "TESTING"]}
-
- test_data_2 = {"type": BASE_URL + '/subtask_type/pipeline/',
- "name": "subtask2",
- "description": 'My other observation',
- "version": 'v0.314159265359',
- "schema": {"myotherkey": "my other value"},
- "realtime": False,
- "queue": True,
- "tags": []}
-
- test_patch = {"type": BASE_URL + '/subtask_type/inspection/',
- "version": 'v6.28318530718',
- "schema": {"mykey": "my better value"},
- }
+class SubtaskTemplateTestCase(unittest.TestCase):
def test_subtask_template_list_apiformat(self):
- r = requests.get(BASE_URL + '/subtask_template/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/subtask_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Subtask Template List" in r.content.decode('utf8'))
@@ -83,46 +48,62 @@ class SubtaskTemplateTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/subtask_template/1234321/', 404, {})
def test_subtask_template_POST_and_GET(self):
+ st_test_data = SubtaskTemplate_test_data()
+
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/', st_test_data, 201, st_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, st_test_data)
def test_subtask_template_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/subtask_template/9876789876/', self.test_data_1, 404, {})
+ st_test_data = SubtaskTemplate_test_data()
+ PUT_and_assert_expected_response(self, BASE_URL + '/subtask_template/9876789876/', st_test_data, 404, {})
def test_subtask_template_PUT(self):
+ st_test_data = SubtaskTemplate_test_data(name="the one")
+ st_test_data2 = SubtaskTemplate_test_data(name="the other")
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/', st_test_data, 201, st_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, st_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, st_test_data2, 200, st_test_data2)
+ GET_and_assert_expected_response(self, url, 200, st_test_data2)
def test_subtask_template_PATCH(self):
+ st_test_data = SubtaskTemplate_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/', st_test_data, 201, st_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, st_test_data)
+
+ test_patch = {"type": BASE_URL + '/subtask_type/inspection/',
+ "version": 'v6.28318530718',
+ "schema": {"mykey": "my better value"},
+ }
# 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.update(self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
+ expected_data = dict(st_test_data)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_subtask_template_DELETE(self):
+ st_test_data = SubtaskTemplate_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/', st_test_data, 201, st_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, st_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_subtask_template_PROTECT_behavior_on_type_choice_deleted(self):
+ st_test_data = SubtaskTemplate_test_data()
# create dependency that is safe to delete (enums are not populated / re-established between tests)
type_data = {'value': 'kickme'}
@@ -130,44 +111,22 @@ class SubtaskTemplateTestCase(unittest.TestCase):
type_url = BASE_URL + '/subtask_type/kickme/'
# POST new item and verify
- test_data = dict(self.test_data_1)
+ test_data = dict(st_test_data)
test_data['type'] = type_url
url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/', test_data, 201, test_data)['url']
GET_and_assert_expected_response(self, url, 200, 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(type_url, auth=self.auth)
+ response = requests.delete(type_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_and_assert_expected_response(self, type_url, 200, type_data)
-
class DataproductSpecificationsTemplateTestCase(unittest.TestCase):
-
- def setUp(self):
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- # test data
- test_data_1 = {"name": "data",
- "description": 'My one date',
- "version": 'v0.314159265359',
- "schema": {"mykey": "my value"},
- "tags": ["TMSS", "TESTING"]}
-
- test_data_2 = {"name": "data",
- "description": 'My other data',
- "version": 'v0.314159265359',
- "schema": {"myotherkey": "my other value"},
- "tags": []}
-
- test_patch = {"version": 'v6.28318530718',
- "schema": {"mykey": "my better value"},
- }
-
def test_dataproduct_specifications_template_list_apiformat(self):
- r = requests.get(BASE_URL + '/dataproduct_specifications_template/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/dataproduct_specifications_template/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Dataproduct Specifications Template List" in r.content.decode('utf8'))
@@ -175,228 +134,116 @@ class DataproductSpecificationsTemplateTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/1234321/', 404, {})
def test_dataproduct_specifications_template_POST_and_GET(self):
+ dst_test_data = DataproductSpecificationsTemplate_test_data()
+
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/', dst_test_data, 201, dst_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, dst_test_data)
def test_dataproduct_specifications_template_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/9876789876/', self.test_data_1, 404, {})
+ dst_test_data = DataproductSpecificationsTemplate_test_data()
+
+ PUT_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/9876789876/', dst_test_data, 404, {})
def test_dataproduct_specifications_template_PUT(self):
+ dst_test_data = DataproductSpecificationsTemplate_test_data(name="the one")
+ dst_test_data2 = DataproductSpecificationsTemplate_test_data(name="the other")
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/', dst_test_data, 201, dst_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, dst_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, dst_test_data2, 200, dst_test_data2)
+ GET_and_assert_expected_response(self, url, 200, dst_test_data2)
def test_dataproduct_specifications_template_PATCH(self):
+ dst_test_data = DataproductSpecificationsTemplate_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/', dst_test_data, 201, dst_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, dst_test_data)
+
+ test_patch = {"version": 'v6.28318530718',
+ "schema": {"mykey": "my better value"},
+ }
# 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.update(self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
+ expected_data = dict(dst_test_data)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_dataproduct_specifications_template_DELETE(self):
+ dst_test_data = DataproductSpecificationsTemplate_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/', dst_test_data, 201, dst_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, dst_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
class DataproductFeedbackTemplateTestCase(unittest.TestCase):
-
- def setUp(self):
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- # test data
- test_data_1 = {"name": "data",
- "description": 'My one date',
- "version": 'v0.314159265359',
- "schema": {"mykey": "my value"},
- "tags": ["TMSS", "TESTING"]}
-
- test_data_2 = {"name": "data",
- "description": 'My other data',
- "version": 'v0.314159265359',
- "schema": {"myotherkey": "my other value"},
- "tags": []}
-
# This currently adds nothing on top of the template base class, so nothing new to test here.
+ pass
class SubtaskInputSelectionTemplateTestCase(unittest.TestCase):
-
- def setUp(self):
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- # test data
- test_data_1 = {"name": "data",
- "description": 'My one date',
- "version": 'v0.314159265359',
- "schema": {"mykey": "my value"},
- "tags": ["TMSS", "TESTING"]}
-
- test_data_2 = {"name": "data",
- "description": 'My other data',
- "version": 'v0.314159265359',
- "schema": {"myotherkey": "my other value"},
- "tags": []}
-
# This currently adds nothing on top of the template base class, so nothing new to test here.
+ pass
-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": []}
-
+class DefaultSubtaskTemplatesTestCase(unittest.TestCase):
def test_default_subtask_template_POST(self):
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/',
- SubtaskTemplateTestCase.test_data_1, 201,
- SubtaskTemplateTestCase.test_data_1)
- url = r_dict['url']
-
- test_data_1 = dict(self.test_data_1)
- test_data_1['template'] = url
- POST_and_assert_expected_response(self, BASE_URL + '/default_subtask_template/', test_data_1, 201, test_data_1)
+ template_url = post_data_and_get_url(SubtaskTemplate_test_data(), '/subtask_template/')
+ dst_test_data = DefaultSubtaskTemplates_test_data(template_url=template_url)
+ POST_and_assert_expected_response(self, BASE_URL + '/default_subtask_template/', dst_test_data, 201, dst_test_data)
def test_default_dataproduct_specifications_template_POST(self):
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/',
- DataproductSpecificationsTemplateTestCase.test_data_1, 201,
- DataproductSpecificationsTemplateTestCase.test_data_1)
- url = r_dict['url']
-
- test_data_1 = dict(self.test_data_1)
- test_data_1['template'] = url
- POST_and_assert_expected_response(self, BASE_URL + '/default_dataproduct_specifications_template/', test_data_1, 201, test_data_1)
+ template_url = post_data_and_get_url(DataproductSpecificationsTemplate_test_data(), '/dataproduct_specifications_template/')
+ dst_test_data = DefaultSubtaskTemplates_test_data(template_url=template_url)
+ POST_and_assert_expected_response(self, BASE_URL + '/default_dataproduct_specifications_template/', dst_test_data, 201, dst_test_data)
def test_default_subtask_template_PROTECT_behavior_on_template_deleted(self):
+ st_test_data = SubtaskTemplate_test_data()
+ template_url = post_data_and_get_url(st_test_data, '/subtask_template/')
+ dst_test_data = DefaultSubtaskTemplates_test_data(template_url=template_url)
# POST with dependency
- template_url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_template/',
- SubtaskTemplateTestCase.test_data_1, 201,
- SubtaskTemplateTestCase.test_data_1)['url']
- test_data = dict(self.test_data_2)
- test_data['template'] = template_url
- POST_and_assert_expected_response(self, BASE_URL + '/default_subtask_template/', test_data, 201, test_data)
+ POST_and_assert_expected_response(self, BASE_URL + '/default_subtask_template/', dst_test_data, 201, dst_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
- response = requests.delete(template_url, auth=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, SubtaskTemplateTestCase.test_data_1)
+ GET_and_assert_expected_response(self, template_url, 200, st_test_data)
def test_default_dataproduct_specifications_template_PROTECT_behavior_on_template_deleted(self):
+ dpst_test_data = DataproductSpecificationsTemplate_test_data()
+ template_url = post_data_and_get_url(dpst_test_data, '/dataproduct_specifications_template/')
+ dst_test_data = DefaultSubtaskTemplates_test_data(template_url=template_url)
# POST with dependency
- template_url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_specifications_template/',
- DataproductSpecificationsTemplateTestCase.test_data_1, 201,
- DataproductSpecificationsTemplateTestCase.test_data_1)['url']
- test_data = dict(self.test_data_2)
- test_data['template'] = template_url
- POST_and_assert_expected_response(self, BASE_URL + '/default_dataproduct_specifications_template/', test_data, 201, test_data)
+ POST_and_assert_expected_response(self, BASE_URL + '/default_dataproduct_specifications_template/', dst_test_data, 201, dst_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
- response = requests.delete(template_url, auth=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, DataproductSpecificationsTemplateTestCase.test_data_1)
-
-
+ GET_and_assert_expected_response(self, template_url, 200, dpst_test_data)
class SubtaskTestCase(unittest.TestCase):
-
- def setUp(self):
-
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- # related items
-
- tttc = SubtaskTemplateTestCase()
- tttc.setUp()
-
- self.task_template_url = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask_template/',
- tttc.test_data_1, 201,
- tttc.test_data_1)['url']
-
- wrbtc = TaskBlueprintTestCase()
- wrbtc.setUp()
-
- self.task_blueprint_url = POST_and_assert_expected_response(self,
- BASE_URL + '/task_blueprint/',
- wrbtc.test_data_1, 201,
- wrbtc.test_data_1)['url']
-
- self.cluster_url_1 = POST_and_assert_expected_response(self,
- BASE_URL + '/cluster/',
- ClusterTestCase.test_data_1, 201,
- ClusterTestCase.test_data_1)['url']
-
- self.cluster_url_2 = POST_and_assert_expected_response(self,
- BASE_URL + '/cluster/',
- ClusterTestCase.test_data_2, 201,
- ClusterTestCase.test_data_2)['url']
-
- # test data
- self.test_data_1 = {"start_time": datetime.utcnow().isoformat(),
- "stop_time": datetime.utcnow().isoformat(),
- "state": BASE_URL + '/subtask_state/scheduling/',
- "specifications_doc": "{}",
- "task_blueprint": self.task_blueprint_url,
- "specifications_template": self.task_template_url,
- "tags": ["TMSS", "TESTING"],
- "do_cancel": datetime.utcnow().isoformat(),
- "priority": 1,
- "schedule_method": BASE_URL + '/schedule_method/manual/',
- "cluster": self.cluster_url_1,
- "scheduler_input_doc": "{}"
- }
-
- self.test_data_2 = {"start_time": datetime.utcnow().isoformat(),
- "stop_time": datetime.utcnow().isoformat(),
- "state": BASE_URL + '/subtask_state/queueing/',
- "specifications_doc": "{'some': 'spec'}",
- "task_blueprint": self.task_blueprint_url,
- "specifications_template": self.task_template_url,
- "tags": [],
- "do_cancel": datetime.utcnow().isoformat(),
- "priority": 2,
- "schedule_method": BASE_URL + '/schedule_method/batch/',
- "cluster": self.cluster_url_2,
- "scheduler_input_doc": "{}"
- }
-
- self.test_patch = {
- "specifications_doc": {"somespec": "somevalue"},
- }
-
def test_subtask_list_apiformat(self):
- r = requests.get(BASE_URL + '/subtask/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/subtask/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Subtask List" in r.content.decode('utf8'))
@@ -404,50 +251,60 @@ class SubtaskTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/subtask/1234321/', 404, {})
def test_subtask_POST_and_GET(self):
+ st_test_data = Subtask_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', st_test_data, 201, st_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, st_test_data)
def test_subtask_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/subtask/9876789876/', self.test_data_1, 404, {})
+ st_test_data = Subtask_test_data()
+
+ PUT_and_assert_expected_response(self, BASE_URL + '/subtask/9876789876/', st_test_data, 404, {})
def test_subtask_PUT(self):
+ st_test_data = Subtask_test_data()
+ st_test_data2 = Subtask_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', st_test_data, 201, st_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, st_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, st_test_data2, 200, st_test_data2)
+ GET_and_assert_expected_response(self, url, 200, st_test_data2)
def test_subtask_PATCH(self):
+ st_test_data = Subtask_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', st_test_data, 201, st_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, st_test_data)
+
+ test_patch = {"specifications_doc": {"somespec": "somevalue"}}
# 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.update(self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
+ expected_data = dict(st_test_data)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_subtask_DELETE(self):
+ st_test_data = Subtask_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', st_test_data, 201, st_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, st_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_subtask_PROTECT_behavior_on_state_choice_deleted(self):
+ st_test_data = Subtask_test_data()
# create dependency that is safe to delete (enums are not populated / re-established between tests)
state_data = {'value': 'kickme'}
@@ -455,112 +312,55 @@ class SubtaskTestCase(unittest.TestCase):
state_url = BASE_URL + '/subtask_state/kickme/'
# POST new item and verify
- test_data = dict(self.test_data_1)
+ test_data = dict(st_test_data)
test_data['state'] = state_url
url = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', test_data, 201, test_data)['url']
GET_and_assert_expected_response(self, url, 200, test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
- response = requests.delete(state_url, auth=self.auth)
+ response = requests.delete(state_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_and_assert_expected_response(self, state_url, 200, state_data)
def test_subtask_SET_NULL_behavior_on_task_blueprint_deleted(self):
+ tbp_test_data = TaskBlueprint_test_data()
+ task_blueprint_url = post_data_and_get_url(tbp_test_data, '/task_blueprint/')
+ st_test_data = Subtask_test_data(task_blueprint_url=task_blueprint_url)
# POST new item and verify
- url = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', self.test_data_2, 201, self.test_data_2)['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ url = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', st_test_data, 201, st_test_data)['url']
+ GET_and_assert_expected_response(self, url, 200, st_test_data)
# DELETE dependency and check it's gone
- DELETE_and_assert_gone(self, self.task_blueprint_url)
+ DELETE_and_assert_gone(self, task_blueprint_url)
# assert item reference is set null
- expected_data = dict(self.test_data_2)
+ expected_data = dict(st_test_data)
expected_data['task_blueprint'] = None
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_subtask_PROTECT_behavior_on_template_deleted(self):
+ stt_test_data = SubtaskTemplate_test_data()
+ specifications_template_url = post_data_and_get_url(stt_test_data, '/subtask_template/')
+ st_test_data = Subtask_test_data(specifications_template_url=specifications_template_url)
# POST new item and verify
- url = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', self.test_data_2, 201, self.test_data_2)['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ url = POST_and_assert_expected_response(self, BASE_URL + '/subtask/', st_test_data, 201, st_test_data)['url']
+ GET_and_assert_expected_response(self, url, 200, st_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
- response = requests.delete(self.task_template_url, auth=self.auth)
+ response = requests.delete(specifications_template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
- GET_and_assert_expected_response(self, self.task_template_url, 200, SubtaskTemplateTestCase.test_data_1)
+ GET_and_assert_expected_response(self, specifications_template_url, 200, stt_test_data)
class DataproductTestCase(unittest.TestCase):
-
- def setUp(self):
-
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- # related items
-
- self.specifications_template_url = POST_and_assert_expected_response(self,
- BASE_URL + '/dataproduct_specifications_template/',
- DataproductSpecificationsTemplateTestCase.test_data_1, 201,
- DataproductSpecificationsTemplateTestCase.test_data_1)['url']
-
- sotc = SubtaskOutputTestCase()
- sotc.setUp()
-
- self.subtask_output_url = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask_output/',
- sotc.test_data_1, 201,
- sotc.test_data_1)['url']
-
- self.dataproduct_feedback_template_url = POST_and_assert_expected_response(self,
- BASE_URL + '/dataproduct_feedback_template/',
- DataproductFeedbackTemplateTestCase.test_data_1, 201,
- DataproductFeedbackTemplateTestCase.test_data_1)['url']
-
-
- # test data
- self.test_data_1 = {"filename": "my.file",
- "directory": "/home/boskabouter/",
- "dataformat": BASE_URL + '/dataformat/Beamformed/',
- "deleted_since": None,
- "pinned_since": None,
- "specifications_doc": "{}",
- "specifications_template": self.specifications_template_url,
- "tags": ["TMSS", "TESTING"],
- "producer": self.subtask_output_url,
- "do_cancel": datetime.utcnow().isoformat(),
- "expected_size": 1234,
- "size": 123,
- "feedback_doc": "{}",
- "feedback_template": self.dataproduct_feedback_template_url
- }
-
- self.test_data_2 = {"filename": "my_other.file",
- "directory": "/home/boskabouter/",
- "dataformat": BASE_URL + '/dataformat/MeasurementSet/',
- "deleted_since": None,
- "pinned_since": None,
- "specifications_doc": "{}",
- "specifications_template": self.specifications_template_url,
- "tags": [],
- "producer": self.subtask_output_url,
- "do_cancel": datetime.utcnow().isoformat(),
- "expected_size": 1234,
- "size": 123,
- "feedback_doc": "{}",
- "feedback_template": self.dataproduct_feedback_template_url
- }
-
- self.test_patch = {"filename": 'my_better.filename',
- "deleted_since": datetime.utcnow().isoformat()
- }
-
def test_dataproduct_list_apiformat(self):
- r = requests.get(BASE_URL + '/dataproduct/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/dataproduct/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Dataproduct List" in r.content.decode('utf8'))
@@ -568,50 +368,61 @@ class DataproductTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/dataproduct/1234321/', 404, {})
def test_dataproduct_POST_and_GET(self):
+ dp_test_data = Dataproduct_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', dp_test_data, 201, dp_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, dp_test_data)
def test_dataproduct_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/dataproduct/9876789876/', self.test_data_1, 404, {})
+ dp_test_data = Dataproduct_test_data()
+
+ PUT_and_assert_expected_response(self, BASE_URL + '/dataproduct/9876789876/', dp_test_data, 404, {})
def test_dataproduct_PUT(self):
+ dp_test_data = Dataproduct_test_data()
+ dp_test_data2 = Dataproduct_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', dp_test_data, 201, dp_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, dp_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, dp_test_data2, 200, dp_test_data2)
+ GET_and_assert_expected_response(self, url, 200, dp_test_data2)
def test_dataproduct_PATCH(self):
+ dp_test_data = Dataproduct_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', dp_test_data, 201, dp_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, dp_test_data)
+
+ test_patch = {"filename": 'my_better.filename',
+ "deleted_since": datetime.utcnow().isoformat()}
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
- expected_data.update(self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
+ expected_data = dict(dp_test_data)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_dataproduct_DELETE(self):
+ dp_test_data = Dataproduct_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', dp_test_data, 201, dp_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, dp_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_dataproduct_PROTECT_behavior_on_dataformat_deleted(self):
+ dp_test_data = Dataproduct_test_data()
# create dependency that is safe to delete (enums are not populated / re-established between tests)
dataformat_data = {'value': 'kickme'}
@@ -619,68 +430,36 @@ class DataproductTestCase(unittest.TestCase):
dataformat_url = BASE_URL + '/dataformat/kickme/'
# POST new item and verify
- test_data = dict(self.test_data_1)
+ test_data = dict(dp_test_data)
test_data['dataformat'] = dataformat_url
url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', test_data, 201, test_data)['url']
GET_and_assert_expected_response(self, url, 200, test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
- response = requests.delete(dataformat_url, auth=self.auth)
+ response = requests.delete(dataformat_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_and_assert_expected_response(self, dataformat_url, 200, dataformat_data)
def test_dataproduct_CASCADE_behavior_on_specifications_template_deleted(self):
+ specifications_template_url = post_data_and_get_url(SubtaskTemplate_test_data(), '/dataproduct_specifications_template/')
+ dp_test_data = Dataproduct_test_data(specifications_template_url=specifications_template_url)
# POST new item, verify
- url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', self.test_data_1, 201, self.test_data_1)['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct/', dp_test_data, 201, dp_test_data)['url']
+ GET_and_assert_expected_response(self, url, 200, dp_test_data)
# DELETE dependency and check it's gone
- DELETE_and_assert_gone(self, self.specifications_template_url)
+ DELETE_and_assert_gone(self, specifications_template_url)
# assert item gone
GET_and_assert_expected_response(self, url, 404, {})
class SubtaskConnectorTestCase(unittest.TestCase):
-
- def setUp(self):
-
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- self.subtask_template_url_1 = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask_template/',
- SubtaskTemplateTestCase.test_data_1, 201,
- SubtaskTemplateTestCase.test_data_1)['url']
-
- self.subtask_template_url_2 = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask_template/',
- SubtaskTemplateTestCase.test_data_2, 201,
- SubtaskTemplateTestCase.test_data_2)['url']
-
- # test data
- self.test_data_1 = {"role": BASE_URL + '/role/correlator/',
- "datatype": BASE_URL + '/datatype/image/',
- "dataformats": [BASE_URL + '/dataformat/Beamformed/'],
- "output_of": self.subtask_template_url_1,
- "input_of": self.subtask_template_url_2,
- "tags": []}
-
- self.test_data_2 = {"role": BASE_URL + '/role/target/',
- "datatype": BASE_URL + '/datatype/image/',
- "dataformats": [BASE_URL + '/dataformat/MeasurementSet/'],
- "output_of": self.subtask_template_url_2,
- "input_of": self.subtask_template_url_1,
- "tags": []}
-
- self.test_patch = {"role": BASE_URL + '/role/calibrator/',
- "datatype": BASE_URL + '/datatype/quality/',
- }
-
def test_subtask_connector_list_apiformat(self):
- r = requests.get(BASE_URL + '/subtask_connector/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/subtask_connector/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Subtask Connector List" in r.content.decode('utf8'))
@@ -688,50 +467,61 @@ class SubtaskConnectorTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/subtask_connector/1234321/', 404, {})
def test_subtask_connector_POST_and_GET(self):
+ stc_test_data = SubtaskConnector_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_connector/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_connector/', stc_test_data, 201, stc_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, stc_test_data)
def test_subtask_connector_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/subtask_connector/9876789876/', self.test_data_1, 404, {})
+ stc_test_data = SubtaskConnector_test_data()
+
+ PUT_and_assert_expected_response(self, BASE_URL + '/subtask_connector/9876789876/', stc_test_data, 404, {})
def test_subtask_connector_PUT(self):
+ stc_test_data = SubtaskConnector_test_data()
+ stc_test_data2 = SubtaskConnector_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_connector/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_connector/', stc_test_data, 201, stc_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, stc_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, stc_test_data2, 200, stc_test_data2)
+ GET_and_assert_expected_response(self, url, 200, stc_test_data2)
def test_subtask_connector_PATCH(self):
+ stc_test_data = SubtaskConnector_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_connector/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_connector/', stc_test_data, 201, stc_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, stc_test_data)
+
+ test_patch = {"role": BASE_URL + '/role/calibrator/',
+ "datatype": BASE_URL + '/datatype/quality/', }
# 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.update(self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
+ expected_data = dict(stc_test_data)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_subtask_connector_DELETE(self):
+ stc_test_data = SubtaskConnector_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_connector/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_connector/', stc_test_data, 201, stc_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, stc_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_subtask_connector_PROTECT_behavior_on_role_deleted(self):
+ stc_test_data = SubtaskConnector_test_data()
# create dependency that is safe to delete (enums are not populated / re-established between tests)
role_data = {'value': 'kickme'}
@@ -740,20 +530,21 @@ class SubtaskConnectorTestCase(unittest.TestCase):
# POST new item and verify
- test_data = dict(self.test_data_1)
+ test_data = dict(stc_test_data)
test_data['role'] = role_url
url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_connector/', test_data, 201, test_data)['url']
GET_and_assert_expected_response(self, url, 200, 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(role_url, auth=self.auth)
+ response = requests.delete(role_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_and_assert_expected_response(self, role_url, 200, role_data)
def test_subtask_connector_PROTECT_behavior_on_datatype_deleted(self):
+ stc_test_data = SubtaskConnector_test_data()
# create new dependency that is safe to delete (enums are not populated / re-established between tests)
datatype_data = {'value': 'kickme'}
@@ -761,117 +552,22 @@ class SubtaskConnectorTestCase(unittest.TestCase):
datatype_url = BASE_URL + '/datatype/kickme/'
# POST new item and verify
- test_data = dict(self.test_data_1)
+ test_data = dict(stc_test_data)
test_data['datatype'] = datatype_url
url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_connector/', test_data, 201, test_data)['url']
GET_and_assert_expected_response(self, url, 200, 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(datatype_url, auth=self.auth)
+ response = requests.delete(datatype_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_and_assert_expected_response(self, datatype_url, 200, datatype_data)
class SubtaskInputTestCase(unittest.TestCase):
-
- def setUp(self):
-
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- stc = SubtaskTestCase()
- stc.setUp()
-
- self.subtask_url_1 = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask/',
- stc.test_data_1, 201,
- stc.test_data_1)['url']
-
-
- self.subtask_url_2 = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask/',
- stc.test_data_2, 201,
- stc.test_data_2)['url']
-
- dtc = DataproductTestCase()
- dtc.setUp()
-
- self.dataproduct_url_1 = POST_and_assert_expected_response(self,
- BASE_URL + '/dataproduct/',
- dtc.test_data_1, 201,
- dtc.test_data_1)['url']
-
- self.dataproduct_url_2 = POST_and_assert_expected_response(self,
- BASE_URL + '/dataproduct/',
- dtc.test_data_2, 201,
- dtc.test_data_2)['url']
-
- sctc = SubtaskConnectorTestCase()
- sctc.setUp()
-
- self.subtask_connector_url_1 = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask_connector/',
- sctc.test_data_1, 201,
- sctc.test_data_1)['url']
-
- self.subtask_connector_url_2 = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask_connector/',
- sctc.test_data_2, 201,
- sctc.test_data_2)['url']
-
- sotc = SubtaskOutputTestCase()
- sotc.setUp()
-
- self.subtask_output_url_1 = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask_output/',
- sotc.test_data_1, 201,
- sotc.test_data_1)['url']
-
- self.subtask_output_url_2 = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask_output/',
- sotc.test_data_2, 201,
- sotc.test_data_2)['url']
-
- self.selection_template_url_1 = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask_input_selection_template/',
- SubtaskInputSelectionTemplateTestCase.test_data_1, 201,
- SubtaskInputSelectionTemplateTestCase.test_data_1)['url']
-
- self.selection_template_url_2 = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask_input_selection_template/',
- SubtaskInputSelectionTemplateTestCase.test_data_2, 201,
- SubtaskInputSelectionTemplateTestCase.test_data_2)['url']
-
-
-
- # test data
- self.test_data_1 = {"subtask": self.subtask_url_1,
- "task_relation_blueprint": None,
- # **TaskRelationBlueprintTest.test_data_1), # todo: after specification model update
- "connector": self.subtask_connector_url_1,
- "producer": self.subtask_output_url_1,
- "dataproducts": [self.dataproduct_url_1, self.dataproduct_url_2],
- "selection_doc": "{}",
- "selection_template": self.selection_template_url_1,
- "tags": []}
-
- self.test_data_2 = {"subtask": self.subtask_url_2,
- "task_relation_blueprint": None,
- # **TaskRelationBlueprintTest.test_data_2), # todo: after specification model update
- "connector": self.subtask_connector_url_2,
- "producer": self.subtask_output_url_2,
- "dataproducts": [self.dataproduct_url_1, self.dataproduct_url_2],
- "selection_doc": "{}",
- "selection_template": self.selection_template_url_2,
- "tags": []}
-
- self.test_patch = {"subtask": self.subtask_url_2,
- "tags": ['FANCYTAG'],
- }
-
def test_subtask_input_list_apiformat(self):
- r = requests.get(BASE_URL + '/subtask_input/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/subtask_input/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Subtask Input List" in r.content.decode('utf8'))
@@ -879,161 +575,141 @@ class SubtaskInputTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/subtask_input/1234321/', 404, {})
def test_subtask_input_POST_and_GET(self):
+ sti_test_data = SubtaskInput_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, sti_test_data)
def test_subtask_input_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/subtask_input/9876789876/', self.test_data_1, 404, {})
+ sti_test_data = SubtaskInput_test_data()
+
+ PUT_and_assert_expected_response(self, BASE_URL + '/subtask_input/9876789876/', sti_test_data, 404, {})
def test_subtask_input_PUT(self):
+ sti_test_data = SubtaskInput_test_data()
+ sti_test_data2 = SubtaskInput_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, sti_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, sti_test_data2, 200, sti_test_data2)
+ GET_and_assert_expected_response(self, url, 200, sti_test_data2)
def test_subtask_input_PATCH(self):
+ sti_test_data = SubtaskInput_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, sti_test_data)
+
+ subtask_url = post_data_and_get_url(Subtask_test_data(), '/subtask/')
+ test_patch = {"subtask": subtask_url,
+ "tags": ['FANCYTAG'],
+ }
# 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.update(self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
+ expected_data = dict(sti_test_data)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_subtask_input_DELETE(self):
+ sti_test_data = SubtaskInput_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, sti_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_subtask_input_CASCADE_behavior_on_subtask_deleted(self):
+ subtask_url = post_data_and_get_url(Subtask_test_data(), '/subtask/')
+ sti_test_data = SubtaskInput_test_data(subtask_url=subtask_url)
# POST new item, verify
- url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', self.test_data_1, 201, self.test_data_1)['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)['url']
+ GET_and_assert_expected_response(self, url, 200, sti_test_data)
# DELETE dependency and check it's gone
- DELETE_and_assert_gone(self, self.subtask_url_1)
+ DELETE_and_assert_gone(self, subtask_url)
# assert item gone
GET_and_assert_expected_response(self, url, 404, {})
def test_subtask_input_SET_NULL_behavior_on_connector_deleted(self):
+ subtask_connector_url = post_data_and_get_url(SubtaskConnector_test_data(), '/subtask_connector/')
+ sti_test_data = SubtaskInput_test_data(subtask_connector_url=subtask_connector_url)
# POST new item, verify
- url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', self.test_data_2, 201, self.test_data_2)['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)['url']
+ GET_and_assert_expected_response(self, url, 200, sti_test_data)
# DELETE dependency and check it's gone
- DELETE_and_assert_gone(self, self.test_data_2['connector'])
+ DELETE_and_assert_gone(self, subtask_connector_url)
# assert item reference is set null
- expected_data = dict(self.test_data_2)
+ expected_data = dict(sti_test_data)
expected_data['connector'] = None
GET_and_assert_expected_response(self, url, 200, expected_data)
- # todo: uncomment once the blueprint is there...
- # def test_subtask_input_SET_NULL_behavior_on_task_relation_blueprint_deleted(self):
- #
- # # POST new item, verify
- # url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', self.test_data_2, 201, self.test_data_2)['url']
- # GET_and_assert_expected_response(self, url, 200, self.test_data_2)
- #
- # # DELETE dependency and check it's gone
- # DELETE_and_assert_gone(self, self.test_data_2['task_relation_blueprint'])
- #
- # # assert item reference is set null
- # expected_data = dict(self.test_data_2)
- # expected_data['task_relation_blueprint'] = None
- # GET_and_assert_expected_response(self, url, 200, expected_data)
+ def test_subtask_input_SET_NULL_behavior_on_task_relation_blueprint_deleted(self):
+ task_relation_blueprint_url = post_data_and_get_url(TaskRelationBlueprint_test_data(), '/task_relation_blueprint/')
+ sti_test_data = SubtaskInput_test_data(task_relation_blueprint_url=task_relation_blueprint_url)
+
+ # POST new item, verify
+ url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)['url']
+ GET_and_assert_expected_response(self, url, 200, sti_test_data)
+
+ # DELETE dependency and check it's gone
+ DELETE_and_assert_gone(self, task_relation_blueprint_url)
+
+ # assert item reference is set null
+ expected_data = dict(sti_test_data)
+ expected_data['task_relation_blueprint'] = None
+ GET_and_assert_expected_response(self, url, 200, expected_data)
def test_subtask_input_PROTECT_behavior_on_producer_deleted(self):
+ subtask_output_url = post_data_and_get_url(SubtaskOutput_test_data(), '/subtask_output/')
+ sti_test_data = SubtaskInput_test_data(subtask_output_url=subtask_output_url)
# POST with dependency
- url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', self.test_data_2, 201, self.test_data_2)['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)['url']
+ GET_and_assert_expected_response(self, url, 200, sti_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
- response = requests.delete(self.test_data_2['producer'], auth=self.auth)
+ response = requests.delete(subtask_output_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
- GET_and_assert_expected_response(self, self.test_data_2['producer'], 200, {})
+ GET_and_assert_expected_response(self, subtask_output_url, 200, {})
def test_subtask_input_PROTECT_behavior_on_selection_template_deleted(self):
+ subtask_input_selection_template_url = post_data_and_get_url(SubtaskInputSelectionTemplate_test_data(), '/subtask_input_selection_template/')
+ sti_test_data = SubtaskInput_test_data(subtask_input_selection_template_url=subtask_input_selection_template_url)
+
# POST with dependency
- url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', self.test_data_2, 201, self.test_data_2)['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_input/', sti_test_data, 201, sti_test_data)['url']
+ GET_and_assert_expected_response(self, url, 200, sti_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
- response = requests.delete(self.test_data_2['selection_template'], auth=self.auth)
+ response = requests.delete(subtask_input_selection_template_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
- GET_and_assert_expected_response(self, self.test_data_2['selection_template'], 200, {})
+ GET_and_assert_expected_response(self, subtask_input_selection_template_url, 200, {})
class SubtaskOutputTestCase(unittest.TestCase):
- def setUp(self):
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- stc = SubtaskTestCase()
- stc.setUp()
-
- self.subtask_url_1 = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask/',
- stc.test_data_1, 201,
- stc.test_data_1)['url']
-
- self.subtask_url_2 = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask/',
- stc.test_data_2, 201,
- stc.test_data_2)['url']
-
- sctc = SubtaskConnectorTestCase()
- sctc.setUp()
-
- self.subtask_connector_url_1 = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask_connector/',
- sctc.test_data_1, 201,
- sctc.test_data_1)['url']
-
- self.subtask_connector_url_2 = POST_and_assert_expected_response(self,
- BASE_URL + '/subtask_connector/',
- sctc.test_data_2, 201,
- sctc.test_data_2)['url']
-
- # test data
- self.test_data_1 = {"subtask": self.subtask_url_1,
- "connector": self.subtask_connector_url_1,
- "tags": []}
-
- self.test_data_2 = {"subtask": self.subtask_url_2,
- "connector": self.subtask_connector_url_2,
- "tags": []}
-
- self.test_patch = {"subtask": self.subtask_url_2,
- "tags": ['FANCYTAG'],
- }
-
def test_subtask_output_list_apiformat(self):
- r = requests.get(BASE_URL + '/subtask_output/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/subtask_output/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Subtask Output List" in r.content.decode('utf8'))
@@ -1041,106 +717,99 @@ class SubtaskOutputTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/subtask_output/1234321/', 404, {})
def test_subtask_output_POST_and_GET(self):
+ sto_test_data = SubtaskOutput_test_data()
+
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', self.test_data_1, 201,
- self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', sto_test_data, 201,
+ sto_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, sto_test_data)
def test_subtask_output_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/subtask_output/9876789876/', self.test_data_1, 404, {})
+ sto_test_data = SubtaskOutput_test_data()
+ PUT_and_assert_expected_response(self, BASE_URL + '/subtask_output/9876789876/', sto_test_data, 404, {})
def test_subtask_output_PUT(self):
+ sto_test_data = SubtaskOutput_test_data()
+ sto_test_data2 = SubtaskOutput_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', self.test_data_1, 201,
- self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', sto_test_data, 201,sto_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, sto_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, sto_test_data2, 200, sto_test_data2)
+ GET_and_assert_expected_response(self, url, 200, sto_test_data2)
def test_subtask_output_PATCH(self):
+ sto_test_data = SubtaskOutput_test_data()
+ sto_test_data2 = SubtaskOutput_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', self.test_data_1, 201,
- self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', sto_test_data, 201,
+ sto_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, sto_test_data)
+
+ test_patch = {"subtask": sto_test_data2["subtask"],
+ "tags": ['FANCYTAG'], }
# 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.update(self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
+ expected_data = dict(sto_test_data)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_subtask_output_DELETE(self):
+ sto_test_data = SubtaskOutput_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', self.test_data_1, 201,
- self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', sto_test_data, 201,
+ sto_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, sto_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_subtask_output_CASCADE_behavior_on_subtask_deleted(self):
+ st_test_data = Subtask_test_data()
+ subtask_url = post_data_and_get_url(st_test_data, '/subtask/')
+ sto_test_data = SubtaskOutput_test_data(subtask_url=subtask_url)
+
# POST new item, verify
- url = \
- POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', self.test_data_1, 201, self.test_data_1)[
- 'url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ url = POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', sto_test_data, 201, sto_test_data)['url']
+ GET_and_assert_expected_response(self, url, 200, sto_test_data)
# DELETE dependency and check it's gone
- DELETE_and_assert_gone(self, self.subtask_url_1)
+ DELETE_and_assert_gone(self, subtask_url)
# assert item gone
GET_and_assert_expected_response(self, url, 404, {})
def test_subtask_output_SET_NULL_behavior_on_connector_deleted(self):
+ sto_test_data = SubtaskOutput_test_data()
+
# POST new item, verify
url = \
- POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', self.test_data_2, 201, self.test_data_2)[
+ POST_and_assert_expected_response(self, BASE_URL + '/subtask_output/', sto_test_data, 201, sto_test_data)[
'url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_2)
+ GET_and_assert_expected_response(self, url, 200, sto_test_data)
# DELETE dependency and check it's gone
- DELETE_and_assert_gone(self, self.test_data_2['connector'])
+ DELETE_and_assert_gone(self, sto_test_data['connector'])
# assert item reference is set null
- expected_data = dict(self.test_data_2)
+ expected_data = dict(sto_test_data)
expected_data['connector'] = None
GET_and_assert_expected_response(self, url, 200, expected_data)
class AntennaSetTestCase(unittest.TestCase):
-
- def setUp(self):
-
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- # test data
- self.test_data_1 = {"name": "antennaset1",
- "description": 'My one observation',
- "station_type": BASE_URL + '/station_type/core/',
- "rcus": [1,2,3,4,5],
- "inputs": ['input1', 'input2'],
- "tags": ['tmss', 'testing']}
-
- self.test_data_2 = {"name": "antennaset2",
- "description": 'My one observation',
- "station_type": BASE_URL + '/station_type/international/',
- "rcus": [],
- "inputs": [],
- "tags": []}
-
- self.test_patch = {"rcus": [11,12,13,14,15],
- "station_type": BASE_URL + '/station_type/remote/',
- }
-
def test_antenna_set_list_apiformat(self):
- r = requests.get(BASE_URL + '/antenna_set/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/antenna_set/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Antenna Set List" in r.content.decode('utf8'))
@@ -1148,50 +817,61 @@ class AntennaSetTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/antenna_set/1234321/', 404, {})
def test_antenna_set_POST_and_GET(self):
+ antennaset_test_data = AntennaSet_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/antenna_set/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/antenna_set/', antennaset_test_data, 201, antennaset_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, antennaset_test_data)
def test_antenna_set_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/antenna_set/9876789876/', self.test_data_1, 404, {})
+ antennaset_test_data = AntennaSet_test_data()
+
+ PUT_and_assert_expected_response(self, BASE_URL + '/antenna_set/9876789876/', antennaset_test_data, 404, {})
def test_antenna_set_PUT(self):
+ antennaset_test_data = AntennaSet_test_data(name="the one")
+ antennaset_test_data2 = AntennaSet_test_data(name="the other")
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/antenna_set/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/antenna_set/', antennaset_test_data, 201, antennaset_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, antennaset_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, antennaset_test_data2, 200, antennaset_test_data2)
+ GET_and_assert_expected_response(self, url, 200, antennaset_test_data2)
def test_antenna_set_PATCH(self):
+ antennaset_test_data = AntennaSet_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/antenna_set/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/antenna_set/', antennaset_test_data, 201, antennaset_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, antennaset_test_data)
+
+ test_patch = {"rcus": [11, 12, 13, 14, 15],
+ "station_type": BASE_URL + '/station_type/remote/'}
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
- expected_data.update(self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
+ expected_data = dict(antennaset_test_data)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_antenna_set_DELETE(self):
+ antennaset_test_data = AntennaSet_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/antenna_set/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/antenna_set/', antennaset_test_data, 201, antennaset_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, antennaset_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_antenna_set_PROTECT_behavior_on_station_type_deleted(self):
+ antennaset_test_data = AntennaSet_test_data()
# create dependency that is safe to delete (enums are not populated / re-established between tests)
dataformat_data = {'value': 'kickme'}
@@ -1199,55 +879,22 @@ class AntennaSetTestCase(unittest.TestCase):
dataformat_url = BASE_URL + '/station_type/kickme/'
# POST new item and verify
- test_data = dict(self.test_data_1)
+ test_data = dict(antennaset_test_data)
test_data['station_type'] = dataformat_url
url = POST_and_assert_expected_response(self, BASE_URL + '/antenna_set/', test_data, 201, test_data)['url']
GET_and_assert_expected_response(self, url, 200, test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
- response = requests.delete(dataformat_url, auth=self.auth)
+ response = requests.delete(dataformat_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
GET_and_assert_expected_response(self, dataformat_url, 200, dataformat_data)
class DataproductTransformTestCase(unittest.TestCase):
-
- def setUp(self):
-
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- dtc = DataproductTestCase()
- dtc.setUp()
-
- self.dataproduct_url_1 = POST_and_assert_expected_response(self,
- BASE_URL + '/dataproduct/',
- dtc.test_data_1, 201,
- dtc.test_data_1)['url']
-
- self.dataproduct_url_2 = POST_and_assert_expected_response(self,
- BASE_URL + '/dataproduct/',
- dtc.test_data_2, 201,
- dtc.test_data_2)['url']
-
- # test data
- self.test_data_1 = {"input": self.dataproduct_url_1,
- "output": self.dataproduct_url_2,
- "identity": True,
- "tags": ['tmss', 'testing']}
-
- self.test_data_2 = {"input": self.dataproduct_url_2,
- "output": self.dataproduct_url_1,
- "identity": False,
- "tags": []}
-
- self.test_patch = {"output": self.dataproduct_url_1,
- "identity": False
- }
-
def test_dataproduct_transform_list_apiformat(self):
- r = requests.get(BASE_URL + '/dataproduct_transform/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/dataproduct_transform/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Dataproduct Transform List" in r.content.decode('utf8'))
@@ -1255,109 +902,97 @@ class DataproductTransformTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/1234321/', 404, {})
def test_dataproduct_transform_POST_and_GET(self):
+ dpt_test_data = DataproductTransform_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', dpt_test_data, 201, dpt_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, dpt_test_data)
def test_dataproduct_transform_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/9876789876/', self.test_data_1, 404, {})
+ dpt_test_data = DataproductTransform_test_data()
+
+ PUT_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/9876789876/', dpt_test_data, 404, {})
def test_dataproduct_transform_PUT(self):
+ dpt_test_data = DataproductTransform_test_data()
+ dpt_test_data2 = DataproductTransform_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', dpt_test_data, 201, dpt_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, dpt_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, dpt_test_data2, 200, dpt_test_data2)
+ GET_and_assert_expected_response(self, url, 200, dpt_test_data2)
def test_dataproduct_transform_PATCH(self):
+ dpt_test_data = DataproductTransform_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', dpt_test_data, 201, dpt_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, dpt_test_data)
+
+ output_dataproduct_url = post_data_and_get_url(Dataproduct_test_data(), '/dataproduct/')
+
+ test_patch = {"output": output_dataproduct_url,
+ "identity": False }
# 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.update(self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
+ expected_data = dict(dpt_test_data)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_dataproduct_transform_DELETE(self):
+ dpt_test_data = DataproductTransform_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', dpt_test_data, 201, dpt_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, dpt_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_dataproduct_transform_PROTECT_behavior_on_input_deleted(self):
+ input_dp_test_data = Dataproduct_test_data()
+ input_dataproduct_url = post_data_and_get_url(input_dp_test_data, '/dataproduct/')
+ dpt_test_data = DataproductTransform_test_data(input_dataproduct_url=input_dataproduct_url)
# POST new item and verify
- url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', self.test_data_1, 201, self.test_data_1)['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', dpt_test_data, 201, dpt_test_data)['url']
+ GET_and_assert_expected_response(self, url, 200, dpt_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
- response = requests.delete(self.test_data_1['input'], auth=self.auth)
+ response = requests.delete(input_dataproduct_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
- GET_and_assert_expected_response(self, self.test_data_1['input'], 200, {})
+ GET_and_assert_expected_response(self, input_dataproduct_url, 200, input_dp_test_data)
def test_dataproduct_transform_PROTECT_behavior_on_output_deleted(self):
+ output_dp_test_data = Dataproduct_test_data()
+ output_dataproduct_url = post_data_and_get_url(output_dp_test_data, '/dataproduct/')
+ dpt_test_data = DataproductTransform_test_data(output_dataproduct_url=output_dataproduct_url)
# POST new item and verify
- url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', self.test_data_1, 201, self.test_data_1)['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_transform/', dpt_test_data, 201, dpt_test_data)['url']
+ GET_and_assert_expected_response(self, url, 200, dpt_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
- response = requests.delete(self.test_data_1['output'], auth=self.auth)
+ response = requests.delete(output_dataproduct_url, auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
- GET_and_assert_expected_response(self, self.test_data_1['output'], 200, {})
+ GET_and_assert_expected_response(self, output_dataproduct_url, 200, output_dp_test_data)
class FilesystemTestCase(unittest.TestCase):
-
- def setUp(self):
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- self.cluster_url_1 = POST_and_assert_expected_response(self,
- BASE_URL + '/cluster/',
- ClusterTestCase.test_data_1, 201,
- ClusterTestCase.test_data_1)['url']
-
- self.cluster_url_2 = POST_and_assert_expected_response(self,
- BASE_URL + '/cluster/',
- ClusterTestCase.test_data_2, 201,
- ClusterTestCase.test_data_2)['url']
-
- # test data
- self.test_data_1 = {"name": "Filesystem 1",
- "description": 'My one filesystem',
- "capacity": 1111111111,
- "cluster": self.cluster_url_1,
- "tags": ['tmss', 'testing']}
-
- self.test_data_2 = {"name": "Filesystem 2",
- "description": 'My other filesystem',
- "capacity": 2222222222,
- "cluster": self.cluster_url_2,
- "tags": []}
-
- self.test_patch = {"cluster": self.cluster_url_2,
- "capacity": 3333333333}
-
def test_filesystem_list_apiformat(self):
- r = requests.get(BASE_URL + '/filesystem/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/filesystem/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Filesystem List" in r.content.decode('utf8'))
@@ -1365,83 +1000,85 @@ class FilesystemTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/filesystem/1234321/', 404, {})
def test_filesystem_POST_and_GET(self):
+ fs_test_data = Filesystem_test_data()
+
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/filesystem/', self.test_data_1,
- 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/filesystem/', fs_test_data, 201, fs_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, fs_test_data)
def test_filesystem_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/filesystem/9876789876/', self.test_data_1,
+ fs_test_data = Filesystem_test_data()
+
+ PUT_and_assert_expected_response(self, BASE_URL + '/filesystem/9876789876/', fs_test_data,
404, {})
def test_filesystem_PUT(self):
+ fs_test_data = Filesystem_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/filesystem/', self.test_data_1,
- 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/filesystem/', fs_test_data,
+ 201, fs_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, fs_test_data)
+
+ fs_test_data2 = Filesystem_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, fs_test_data2, 200, fs_test_data2)
+ GET_and_assert_expected_response(self, url, 200, fs_test_data2)
def test_filesystem_PATCH(self):
+ cluster_url = post_data_and_get_url(Cluster_test_data(), '/cluster/')
+ fs_test_data = Filesystem_test_data(cluster_url=cluster_url)
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/filesystem/', self.test_data_1,
- 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/filesystem/', fs_test_data,
+ 201, fs_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, fs_test_data)
+
+ cluster_url2 = post_data_and_get_url(Cluster_test_data(), '/cluster/')
+ test_patch = {"cluster": cluster_url2,
+ "capacity": 3333333333}
# 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.update(self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
+ expected_data = dict(fs_test_data)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_filesystem_DELETE(self):
+ fs_test_data = Filesystem_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/filesystem/', self.test_data_1,
- 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/filesystem/', fs_test_data,
+ 201, fs_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, fs_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_filesystem_PROTECT_behavior_on_cluster_deleted(self):
+ fs_test_data = Filesystem_test_data()
+
# POST new item and verify
- url = POST_and_assert_expected_response(self, BASE_URL + '/filesystem/', self.test_data_1, 201,
- self.test_data_1)['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ url = POST_and_assert_expected_response(self, BASE_URL + '/filesystem/', fs_test_data, 201,
+ fs_test_data)['url']
+ GET_and_assert_expected_response(self, url, 200, fs_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
- response = requests.delete(self.test_data_1['cluster'], auth=self.auth)
+ response = requests.delete(fs_test_data['cluster'], auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
- GET_and_assert_expected_response(self, self.test_data_1['cluster'], 200, {})
+ GET_and_assert_expected_response(self, fs_test_data['cluster'], 200, {})
class ClusterTestCase(unittest.TestCase):
-
- auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- # test data
- test_data_1 = {"name": "Cluster 1",
- "description": 'My one cluster',
- "location": "upstairs",
- "tags": ['tmss', 'testing']}
-
- test_data_2 = {"name": "Cluster 2",
- "description": 'My other cluster',
- "location": "downstairs",
- "tags": []}
-
- test_patch = {"location": 'at the other end of the universe'}
-
def test_cluster_list_apiformat(self):
- r = requests.get(BASE_URL + '/cluster/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/cluster/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Cluster List" in r.content.decode('utf8'))
@@ -1449,84 +1086,62 @@ class ClusterTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/cluster/1234321/', 404, {})
def test_cluster_POST_and_GET(self):
+ c_test_data = Cluster_test_data()
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cluster/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cluster/', c_test_data, 201, c_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, c_test_data)
def test_cluster_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/cluster/9876789876/', self.test_data_1, 404, {})
+ c_test_data = Cluster_test_data()
+ PUT_and_assert_expected_response(self, BASE_URL + '/cluster/9876789876/', c_test_data, 404, {})
def test_cluster_PUT(self):
+ c_test_data = Cluster_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cluster/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cluster/', c_test_data, 201, c_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, c_test_data)
+
+ c_test_data2 = Cluster_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, c_test_data2, 200, c_test_data2)
+ GET_and_assert_expected_response(self, url, 200, c_test_data2)
def test_cluster_PATCH(self):
+ c_test_data = Cluster_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cluster/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cluster/', c_test_data, 201, c_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, c_test_data)
+
+ test_patch = {"location": 'at the other end of the universe'}
# 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.update(self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
+ expected_data = dict(c_test_data)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_cluster_DELETE(self):
+ c_test_data = Cluster_test_data()
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cluster/', self.test_data_1, 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cluster/', c_test_data, 201, c_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, c_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
class DataproductHashTestCase(unittest.TestCase):
-
- def setUp(self):
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- dtc = DataproductTestCase()
- dtc.setUp()
-
- self.dataproduct_url_1 = POST_and_assert_expected_response(self,
- BASE_URL + '/dataproduct/',
- dtc.test_data_1, 201,
- dtc.test_data_1)['url']
-
- self.dataproduct_url_2 = POST_and_assert_expected_response(self,
- BASE_URL + '/dataproduct/',
- dtc.test_data_2, 201,
- dtc.test_data_2)['url']
-
- # test data
- self.test_data_1 = {"dataproduct": self.dataproduct_url_1,
- "algorithm": BASE_URL + '/algorithm/md5/',
- "hash": "myhash_1",
- "tags": ['tmss', 'testing']}
-
- self.test_data_2 = {"dataproduct": self.dataproduct_url_2,
- "algorithm": BASE_URL + '/algorithm/aes256/',
- "hash": "myhash_2",
- "tags": []}
-
- self.test_patch = {"algorithm": BASE_URL + '/algorithm/aes256/',
- "hash": 'bender-was-here'}
-
def test_dataproduct_hash_list_apiformat(self):
- r = requests.get(BASE_URL + '/dataproduct_hash/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/dataproduct_hash/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Dataproduct Hash List" in r.content.decode('utf8'))
@@ -1534,111 +1149,98 @@ class DataproductHashTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/1234321/', 404, {})
def test_dataproduct_hash_POST_and_GET(self):
+ dph_test_data = DataproductHash_test_data()
+
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', self.test_data_1,
- 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', dph_test_data,
+ 201, dph_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, dph_test_data)
def test_dataproduct_hash_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/9876789876/', self.test_data_1,
+ dph_test_data = DataproductHash_test_data()
+
+ PUT_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/9876789876/', dph_test_data,
404, {})
def test_dataproduct_hash_PUT(self):
+ dph_test_data = DataproductHash_test_data(hash="the one")
+ dph_test_data2 = DataproductHash_test_data(hash="the other")
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', self.test_data_1,
- 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', dph_test_data,
+ 201, dph_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, dph_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, dph_test_data2, 200, dph_test_data2)
+ GET_and_assert_expected_response(self, url, 200, dph_test_data2)
def test_dataproduct_hash_PATCH(self):
+ dph_test_data = DataproductHash_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', self.test_data_1,
- 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', dph_test_data,
+ 201, dph_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, dph_test_data)
+
+ test_patch = {"algorithm": BASE_URL + '/algorithm/aes256/',
+ "hash": 'bender-was-here'}
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
- expected_data = dict(self.test_data_1)
- expected_data.update(self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
+ expected_data = dict(dph_test_data)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_dataproduct_hash_DELETE(self):
+ dph_test_data = DataproductHash_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', self.test_data_1,
- 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', dph_test_data,
+ 201, dph_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, dph_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_dataproduct_hash_PROTECT_behavior_on_dataproduct_deleted(self):
+ dph_test_data = DataproductHash_test_data()
+
# POST new item and verify
- url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', self.test_data_1, 201,
- self.test_data_1)['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', dph_test_data, 201,
+ dph_test_data)['url']
+ GET_and_assert_expected_response(self, url, 200, dph_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
- response = requests.delete(self.test_data_1['dataproduct'], auth=self.auth)
+ response = requests.delete(dph_test_data['dataproduct'], auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
- GET_and_assert_expected_response(self, self.test_data_1['dataproduct'], 200, {})
+ GET_and_assert_expected_response(self, dph_test_data['dataproduct'], 200, {})
def test_dataproduct_hash_PROTECT_behavior_on_algorithm_deleted(self):
+ dph_test_data = DataproductHash_test_data()
+
# POST new item and verify
- url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', self.test_data_1, 201,
- self.test_data_1)['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_hash/', dph_test_data, 201,
+ dph_test_data)['url']
+ GET_and_assert_expected_response(self, url, 200, dph_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
- response = requests.delete(self.test_data_1['algorithm'], auth=self.auth)
+ response = requests.delete(dph_test_data['algorithm'], auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
- GET_and_assert_expected_response(self, self.test_data_1['algorithm'], 200, {})
+ GET_and_assert_expected_response(self, dph_test_data['algorithm'], 200, {})
class DataproductArchiveInfoTestCase(unittest.TestCase):
-
- def setUp(self):
- self.auth = HTTPBasicAuth(username=ldap_credentials.user, password=ldap_credentials.password)
-
- dtc = DataproductTestCase()
- dtc.setUp()
-
- self.dataproduct_url_1 = POST_and_assert_expected_response(self,
- BASE_URL + '/dataproduct/',
- dtc.test_data_1, 201,
- dtc.test_data_1)['url']
-
- self.dataproduct_url_2 = POST_and_assert_expected_response(self,
- BASE_URL + '/dataproduct/',
- dtc.test_data_2, 201,
- dtc.test_data_2)['url']
- # test data
- self.test_data_1 = {"dataproduct": self.dataproduct_url_1,
- "storage_ticket": "myticket_1",
- "public_since": datetime.utcnow().isoformat(),
- "corrupted_since": datetime.utcnow().isoformat(),
- "tags": ['tmss', 'testing']}
-
- self.test_data_2 = {"dataproduct": self.dataproduct_url_2,
- "storage_ticket": "myticket_2",
- "public_since": datetime.utcnow().isoformat(),
- "corrupted_since": datetime.utcnow().isoformat(),
- "tags": []}
-
- self.test_patch = {"storage_ticket": "mygoldenticket"}
-
def test_dataproduct_archive_info_list_apiformat(self):
- r = requests.get(BASE_URL + '/dataproduct_archive_info/?format=api', auth=self.auth)
+ r = requests.get(BASE_URL + '/dataproduct_archive_info/?format=api', auth=AUTH)
self.assertEqual(r.status_code, 200)
self.assertTrue("Dataproduct Archive Info List" in r.content.decode('utf8'))
@@ -1646,62 +1248,77 @@ class DataproductArchiveInfoTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/1234321/', 404, {})
def test_dataproduct_archive_info_POST_and_GET(self):
+ dpai_test_data = DataproductArchiveInfo_test_data()
+
# POST and GET a new item and assert correctness
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', self.test_data_1,
- 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', dpai_test_data,
+ 201, dpai_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, dpai_test_data)
def test_dataproduct_archive_info_PUT_invalid_raises_error(self):
- PUT_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/9876789876/', self.test_data_1,
+ dpai_test_data = DataproductArchiveInfo_test_data()
+
+ PUT_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/9876789876/', dpai_test_data,
404, {})
def test_dataproduct_archive_info_PUT(self):
+ dpai_test_data = DataproductArchiveInfo_test_data()
+ dpai_test_data2 = DataproductArchiveInfo_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', self.test_data_1,
- 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', dpai_test_data,
+ 201, dpai_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, dpai_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, dpai_test_data2, 200, dpai_test_data2)
+ GET_and_assert_expected_response(self, url, 200, dpai_test_data2)
def test_dataproduct_archive_info_PATCH(self):
+ dpai_test_data = DataproductArchiveInfo_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', self.test_data_1,
- 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', dpai_test_data,
+ 201, dpai_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, dpai_test_data)
+
+ test_patch = {"storage_ticket": "mygoldenticket"}
# 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.update(self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
+ expected_data = dict(dpai_test_data)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_dataproduct_archive_info_DELETE(self):
+ dpai_test_data = DataproductArchiveInfo_test_data()
+
# POST new item, verify
- r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', self.test_data_1,
- 201, self.test_data_1)
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', dpai_test_data,
+ 201, dpai_test_data)
url = r_dict['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ GET_and_assert_expected_response(self, url, 200, dpai_test_data)
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
def test_dataproduct_archive_info_PROTECT_behavior_on_dataproduct_deleted(self):
+ dpai_test_data = DataproductArchiveInfo_test_data()
+
# POST new item and verify
- url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', self.test_data_1, 201,
- self.test_data_1)['url']
- GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+ url = POST_and_assert_expected_response(self, BASE_URL + '/dataproduct_archive_info/', dpai_test_data, 201,
+ dpai_test_data)['url']
+ GET_and_assert_expected_response(self, url, 200, dpai_test_data)
# Try to DELETE dependency, verify that was not successful
# Unfortunately we don't get a nice error in json, but a Django debug page on error 500...
- response = requests.delete(self.test_data_1['dataproduct'], auth=self.auth)
+ response = requests.delete(dpai_test_data['dataproduct'], auth=AUTH)
self.assertEqual(500, response.status_code)
self.assertTrue("ProtectedError" in str(response.content))
- GET_and_assert_expected_response(self, self.test_data_1['dataproduct'], 200, {})
+ GET_and_assert_expected_response(self, dpai_test_data['dataproduct'], 200, {})
if __name__ == "__main__":
diff --git a/SAS/TMSS/test/t_tmssapp_specification_functional.py b/SAS/TMSS/test/t_tmssapp_specification_functional.py
index 48c38de1c0b..e995cbbea84 100755
--- a/SAS/TMSS/test/t_tmssapp_specification_functional.py
+++ b/SAS/TMSS/test/t_tmssapp_specification_functional.py
@@ -27,14 +27,16 @@
# todo: behavior in a controlled way.
# todo: We should probably also fully test behavior wrt mandatory and nullable fields.
+import unittest
import logging
logger = logging.getLogger(__name__)
logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO)
-# use setup/teardown magic for tmss test database
+# Do Mandatory setup step:
+# use setup/teardown magic for tmss test database, ldap server and django server
+# (ignore pycharm unused import statement, python unittests does use at RunTime the tmss_test_environment_unittest_setup module)
from lofar.sas.tmss.test.tmss_test_environment_unittest_setup import *
-import unittest
class BasicFunctionTestCase(unittest.TestCase):
# todo: test_welcome_page (once we have one :))
@@ -42,10 +44,6 @@ class BasicFunctionTestCase(unittest.TestCase):
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=AUTH)
self.assertEqual(r.status_code, 200)
@@ -82,10 +80,13 @@ class GeneratorTemplateTestCase(unittest.TestCase):
url = r_dict['url']
GET_and_assert_expected_response(self, url, 200, GeneratorTemplate_test_data())
+ test_patch = {"version": 'v6.28318530718',
+ "schema": {"mykey": "my better value"}}
+
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(GeneratorTemplate_test_data())
- expected_data.update(self.test_patch)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_generator_template_DELETE(self):
@@ -100,10 +101,6 @@ class GeneratorTemplateTestCase(unittest.TestCase):
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=AUTH)
self.assertEqual(r.status_code, 200)
@@ -140,10 +137,13 @@ class SchedulingUnitTemplateTestCase(unittest.TestCase):
url = r_dict['url']
GET_and_assert_expected_response(self, url, 200, SchedulingUnitTemplate_test_data())
+ test_patch = {"version": 'v6.28318530718',
+ "schema": {"mykey": "my better value"}}
+
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(SchedulingUnitTemplate_test_data())
- expected_data.update(self.test_patch)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_scheduling_unit_template_DELETE(self):
@@ -157,9 +157,6 @@ class SchedulingUnitTemplateTestCase(unittest.TestCase):
DELETE_and_assert_gone(self, url)
class TaskTemplateTestCase(unittest.TestCase):
- 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=AUTH)
@@ -197,10 +194,13 @@ class TaskTemplateTestCase(unittest.TestCase):
url = r_dict['url']
GET_and_assert_expected_response(self, url, 200, TaskTemplate_test_data())
+ test_patch = {"version": 'v6.28318530718',
+ "schema": {"mykey": "my better value"},
+ }
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(TaskTemplate_test_data())
- expected_data.update(self.test_patch)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_task_template_DELETE(self):
@@ -215,10 +215,6 @@ class TaskTemplateTestCase(unittest.TestCase):
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=AUTH)
self.assertEqual(r.status_code, 200)
@@ -255,10 +251,14 @@ class WorkRelationSelectionTemplateTestCase(unittest.TestCase):
url = r_dict['url']
GET_and_assert_expected_response(self, url, 200, WorkRelationSelectionTemplate_test_data())
+ test_patch = {"version": 'v6.28318530718',
+ "schema": {"mykey": "my better value"},
+ }
+
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(WorkRelationSelectionTemplate_test_data())
- expected_data.update(self.test_patch)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_work_relation_selection_template_DELETE(self):
@@ -274,10 +274,6 @@ class WorkRelationSelectionTemplateTestCase(unittest.TestCase):
class TaskConnectorsTestCase(unittest.TestCase):
- 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=AUTH)
self.assertEqual(r.status_code, 200)
@@ -368,10 +364,14 @@ class TaskConnectorsTestCase(unittest.TestCase):
url = r_dict['url']
GET_and_assert_expected_response(self, url, 200, tc_test_data)
+ test_patch = {"role": BASE_URL + '/role/calibrator/',
+ "dataformats": [BASE_URL + '/dataformat/Beamformed/',
+ BASE_URL + '/dataformat/MeasurementSet/']}
+
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(tc_test_data)
- expected_data.update(self.test_patch)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_task_connectors_DELETE(self):
@@ -535,8 +535,6 @@ class DefaultTemplates(unittest.TestCase):
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=AUTH)
self.assertEqual(r.status_code, 200)
@@ -579,10 +577,12 @@ class CycleTestCase(unittest.TestCase):
url = r_dict['url']
GET_and_assert_expected_response(self, url, 200, cycle_test_data)
+ test_patch = {"start": datetime(year=2015, month=10, day=21).isoformat()}
+
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(cycle_test_data)
- expected_data.update(self.test_patch)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_cycle_DELETE(self):
@@ -598,9 +598,6 @@ class CycleTestCase(unittest.TestCase):
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=AUTH)
self.assertEqual(r.status_code, 200)
@@ -642,10 +639,13 @@ class ProjectTestCase(unittest.TestCase):
url = r_dict['url']
GET_and_assert_expected_response(self, url, 200, project_test_data)
+ test_patch = {"priority": 500,
+ "tags": ["SUPERIMPORTANT"]}
+
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(project_test_data)
- expected_data.update(self.test_patch)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_project_DELETE(self):
@@ -683,9 +683,6 @@ class ProjectTestCase(unittest.TestCase):
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=AUTH)
self.assertEqual(r.status_code, 200)
@@ -728,10 +725,13 @@ class SchedulingSetTestCase(unittest.TestCase):
url = r_dict['url']
GET_and_assert_expected_response(self, url, 200, schedulingset_test_data)
+ test_patch = {"description": "This is a new and improved description",
+ "generator_doc": "{'para': 'meter'}"}
+
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(schedulingset_test_data)
- expected_data.update(self.test_patch)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_scheduling_set_DELETE(self):
@@ -783,9 +783,6 @@ class SchedulingSetTestCase(unittest.TestCase):
class SchedulingUnitDraftTestCase(unittest.TestCase):
- 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=AUTH)
self.assertEqual(r.status_code, 200)
@@ -828,10 +825,13 @@ class SchedulingUnitDraftTestCase(unittest.TestCase):
url = r_dict['url']
GET_and_assert_expected_response(self, url, 200, schedulingunitdraft_test_data)
+ test_patch = {"description": "This is a new and improved description",
+ "requirements_doc": "{'para': 'meter'}"}
+
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(schedulingunitdraft_test_data)
- expected_data.update(self.test_patch)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_scheduling_unit_draft_DELETE(self):
@@ -898,9 +898,6 @@ class SchedulingUnitDraftTestCase(unittest.TestCase):
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=AUTH)
self.assertEqual(r.status_code, 200)
@@ -942,10 +939,13 @@ class TaskDraftTestCase(unittest.TestCase):
url = r_dict['url']
GET_and_assert_expected_response(self, url, 200, taskdraft_test_data)
+ test_patch = {"description": "This is a new and improved description",
+ "specifications_doc": "{'para': 'meter'}"}
+
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(taskdraft_test_data)
- expected_data.update(self.test_patch)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_task_draft_DELETE(self):
@@ -1013,8 +1013,6 @@ class TaskDraftTestCase(unittest.TestCase):
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=AUTH)
self.assertEqual(r.status_code, 200)
@@ -1056,10 +1054,12 @@ class TaskRelationDraftTestCase(unittest.TestCase):
url = r_dict['url']
GET_and_assert_expected_response(self, url, 200, trd_test_data)
+ test_patch = {"selection_doc": "{'para': 'meter'}"}
+
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(trd_test_data)
- expected_data.update(self.test_patch)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_task_relation_draft_DELETE(self):
@@ -1162,9 +1162,6 @@ class TaskRelationDraftTestCase(unittest.TestCase):
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=AUTH)
self.assertEqual(r.status_code, 200)
@@ -1206,10 +1203,13 @@ class SchedulingUnitBlueprintTestCase(unittest.TestCase):
url = r_dict['url']
GET_and_assert_expected_response(self, url, 200, sub_test_data)
+ test_patch = {"description": "This is an updated description",
+ "do_cancel": True}
+
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(sub_test_data)
- expected_data.update(self.test_patch)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_scheduling_unit_blueprint_DELETE(self):
@@ -1254,9 +1254,6 @@ class SchedulingUnitBlueprintTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 404, {})
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=AUTH)
self.assertEqual(r.status_code, 200)
@@ -1298,10 +1295,13 @@ class TaskBlueprintTestCase(unittest.TestCase):
url = r_dict['url']
GET_and_assert_expected_response(self, url, 200, tb_test_data)
+ test_patch = {"description": "This is an updated description",
+ "do_cancel": True}
+
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(tb_test_data)
- expected_data.update(self.test_patch)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_task_blueprint_DELETE(self):
@@ -1395,8 +1395,6 @@ class TaskBlueprintTestCase(unittest.TestCase):
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=AUTH)
self.assertEqual(r.status_code, 200)
@@ -1438,10 +1436,12 @@ class TaskRelationBlueprintTestCase(unittest.TestCase):
url = r_dict['url']
GET_and_assert_expected_response(self, url, 200, trb_test_data)
+ test_patch = {"selection_doc": "{'new': 'doc'}"}
+
# PATCH item and verify
- PATCH_and_assert_expected_response(self, url, self.test_patch, 200, self.test_patch)
+ PATCH_and_assert_expected_response(self, url, test_patch, 200, test_patch)
expected_data = dict(trb_test_data)
- expected_data.update(self.test_patch)
+ expected_data.update(test_patch)
GET_and_assert_expected_response(self, url, 200, expected_data)
def test_task_relation_blueprint_DELETE(self):
diff --git a/SAS/TMSS/test/tmss_test_environment_unittest_setup.py b/SAS/TMSS/test/tmss_test_environment_unittest_setup.py
index 6b3c7a7ffbb..c8dc5e904cc 100644
--- a/SAS/TMSS/test/tmss_test_environment_unittest_setup.py
+++ b/SAS/TMSS/test/tmss_test_environment_unittest_setup.py
@@ -133,9 +133,10 @@ def DELETE_and_assert_gone(test_instance, url):
-###################################################
+#######################################################
# the methods below can be used to create test data
-###################################################
+# naming convention is: <django_model_name>_test_data()
+#######################################################
from datetime import datetime
import uuid
@@ -357,4 +358,204 @@ def TaskRelationBlueprint_test_data(draft_url=None, template_url=None, input_url
"producer": producer_url,
"consumer": consumer_url}
+def SubtaskTemplate_test_data(name="subtask1"):
+ return {"type": BASE_URL + '/subtask_type/copy/',
+ "name": name,
+ "description": 'My one observation',
+ "version": 'v0.314159265359',
+ "schema": {"mykey": "my value"},
+ "realtime": True,
+ "queue": False,
+ "tags": ["TMSS", "TESTING"]}
+
+def DataproductSpecificationsTemplate_test_data(name="my_DataproductSpecificationsTemplate"):
+ return {"name": name,
+ "description": 'My one date',
+ "version": 'v0.314159265359',
+ "schema": {"mykey": "my value"},
+ "tags": ["TMSS", "TESTING"]}
+
+def DataproductFeedbackTemplate_test_data(name="my_DataproductFeedbackTemplate"):
+ return {"name": name,
+ "description": 'My one date',
+ "version": 'v0.314159265359',
+ "schema": {"mykey": "my value"},
+ "tags": ["TMSS", "TESTING"]}
+
+def DefaultSubtaskTemplates_test_data(name=None, template_url=None):
+ if template_url is None:
+ template_url = post_data_and_get_url(SubtaskTemplate_test_data(), '/subtask_template/')
+
+ return {"name": name if name else "default_template_%s" % uuid.uuid4(),
+ "template": template_url,
+ "tags": []}
+
+def Cluster_test_data(name=None):
+ return {"name": name if name else "Cluster %s" % uuid.uuid4(),
+ "description": 'My one cluster',
+ "location": "upstairs",
+ "tags": ['tmss', 'testing']}
+
+def Subtask_test_data(cluster_url=None, task_blueprint_url=None, specifications_template_url=None):
+ if cluster_url is None:
+ cluster_url = post_data_and_get_url(Cluster_test_data(), '/cluster/')
+
+ if task_blueprint_url is None:
+ task_blueprint_url = post_data_and_get_url(TaskBlueprint_test_data(), '/task_blueprint/')
+
+ if specifications_template_url is None:
+ specifications_template_url = post_data_and_get_url(SubtaskTemplate_test_data(), '/subtask_template/')
+
+ return {"start_time": datetime.utcnow().isoformat(),
+ "stop_time": datetime.utcnow().isoformat(),
+ "state": BASE_URL + '/subtask_state/scheduling/',
+ "specifications_doc": "{}",
+ "task_blueprint": task_blueprint_url,
+ "specifications_template": specifications_template_url,
+ "tags": ["TMSS", "TESTING"],
+ "do_cancel": datetime.utcnow().isoformat(),
+ "priority": 1,
+ "schedule_method": BASE_URL + '/schedule_method/manual/',
+ "cluster": cluster_url,
+ "scheduler_input_doc": "{}" }
+
+def SubtaskOutput_test_data(subtask_url=None, subtask_connector_url=None):
+ if subtask_url is None:
+ subtask_url = post_data_and_get_url(Subtask_test_data(), '/subtask/')
+
+ if subtask_connector_url is None:
+ subtask_connector_url = post_data_and_get_url(SubtaskConnector_test_data(), '/subtask_connector/')
+
+ return {"subtask": subtask_url,
+ "connector": subtask_connector_url,
+ "tags": []}
+
+def SubtaskConnector_test_data(input_of_url=None, output_of_url=None):
+ if input_of_url is None:
+ input_of_url = post_data_and_get_url(SubtaskTemplate_test_data(), '/subtask_template/')
+
+ if output_of_url is None:
+ output_of_url = post_data_and_get_url(SubtaskTemplate_test_data(), '/subtask_template/')
+
+ return {"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": []}
+
+def Dataproduct_test_data(filename="my_filename", specifications_template_url=None, subtask_output_url=None, dataproduct_feedback_template_url=None):
+ if specifications_template_url is None:
+ specifications_template_url = post_data_and_get_url(SubtaskTemplate_test_data(), '/dataproduct_specifications_template/')
+
+ if subtask_output_url is None:
+ subtask_output_url = post_data_and_get_url(SubtaskOutput_test_data(), '/subtask_output/')
+
+ if dataproduct_feedback_template_url is None:
+ dataproduct_feedback_template_url = post_data_and_get_url(DataproductFeedbackTemplate_test_data(), '/dataproduct_feedback_template/')
+
+ return {"filename": "my.file",
+ "directory": "/home/boskabouter/",
+ "dataformat": BASE_URL + '/dataformat/Beamformed/',
+ "deleted_since": None,
+ "pinned_since": None,
+ "specifications_doc": "{}",
+ "specifications_template": specifications_template_url,
+ "tags": ["TMSS", "TESTING"],
+ "producer": subtask_output_url,
+ "do_cancel": datetime.utcnow().isoformat(),
+ "expected_size": 1234,
+ "size": 123,
+ "feedback_doc": "{}",
+ "feedback_template": dataproduct_feedback_template_url
+ }
+
+def AntennaSet_test_data(name="antennaset1"):
+ return {"name": name,
+ "description": 'My one observation',
+ "station_type": BASE_URL + '/station_type/core/',
+ "rcus": [1,2,3,4,5],
+ "inputs": ['input1', 'input2'],
+ "tags": ['tmss', 'testing']}
+
+def DataproductTransform_test_data(input_dataproduct_url=None, output_dataproduct_url=None):
+ if input_dataproduct_url is None:
+ input_dataproduct_url = post_data_and_get_url(Dataproduct_test_data(), '/dataproduct/')
+
+ if output_dataproduct_url is None:
+ output_dataproduct_url = post_data_and_get_url(Dataproduct_test_data(), '/dataproduct/')
+
+ return {"input": input_dataproduct_url,
+ "output": output_dataproduct_url,
+ "identity": True,
+ "tags": ['tmss', 'testing']}
+
+def DataproductHash_test_data(algorithm_url=None, hash="my_hash", dataproduct_url=None):
+ if algorithm_url is None:
+ algorithm_url = BASE_URL + '/algorithm/md5/'
+
+ if dataproduct_url is None:
+ dataproduct_url = post_data_and_get_url(Dataproduct_test_data(), '/dataproduct/')
+
+ return {"dataproduct": dataproduct_url,
+ "algorithm": algorithm_url,
+ "hash": hash,
+ "tags": ['tmss', 'testing']}
+
+
+def DataproductArchiveInfo_test_data(storage_ticket="my_storage_ticket", dataproduct_url=None):
+ if dataproduct_url is None:
+ dataproduct_url = post_data_and_get_url(Dataproduct_test_data(), '/dataproduct/')
+
+ return {"dataproduct": dataproduct_url,
+ "storage_ticket": storage_ticket,
+ "public_since": datetime.utcnow().isoformat(),
+ "corrupted_since": datetime.utcnow().isoformat(),
+ "tags": ['tmss', 'testing']}
+
+def SubtaskInputSelectionTemplate_test_data(name="my_SubtaskInputSelectionTemplate"):
+ return {"name": name,
+ "description": 'My one date',
+ "version": 'v0.314159265359',
+ "schema": {"mykey": "my value"},
+ "tags": ["TMSS", "TESTING"]}
+
+def SubtaskInput_test_data(subtask_url=None, task_relation_blueprint_url=None, dataproduct_urls=None, subtask_connector_url=None, subtask_output_url=None, subtask_input_selection_template_url=None):
+ if subtask_url is None:
+ subtask_url = post_data_and_get_url(Subtask_test_data(), '/subtask/')
+
+ if task_relation_blueprint_url is None:
+ task_relation_blueprint_url = post_data_and_get_url(TaskRelationBlueprint_test_data(), '/task_relation_blueprint/')
+
+ if dataproduct_urls is None:
+ dataproduct_urls = [post_data_and_get_url(Dataproduct_test_data(), '/dataproduct/'),
+ post_data_and_get_url(Dataproduct_test_data(), '/dataproduct/')]
+
+ if subtask_connector_url is None:
+ subtask_connector_url = post_data_and_get_url(SubtaskConnector_test_data(), '/subtask_connector/')
+
+ if subtask_output_url is None:
+ subtask_output_url = post_data_and_get_url(SubtaskOutput_test_data(), '/subtask_output/')
+
+ if subtask_input_selection_template_url is None:
+ subtask_input_selection_template_url = post_data_and_get_url(SubtaskInputSelectionTemplate_test_data(), '/subtask_input_selection_template/')
+
+ return {"subtask": subtask_url,
+ "task_relation_blueprint": task_relation_blueprint_url,
+ "connector": subtask_connector_url,
+ "producer": subtask_output_url,
+ "dataproducts": dataproduct_urls,
+ "selection_doc": "{}",
+ "selection_template": subtask_input_selection_template_url,
+ "tags": []}
+
+def Filesystem_test_data(name="my_Filesystem", cluster_url=None):
+ if cluster_url is None:
+ cluster_url = post_data_and_get_url(Cluster_test_data(), '/cluster/')
+
+ return {"name": name,
+ "description": 'My one filesystem',
+ "capacity": 1111111111,
+ "cluster": cluster_url,
+ "tags": ['tmss', 'testing']}
--
GitLab