diff --git a/SAS/TMSS/test/t_tmssapp_specification_functional.py b/SAS/TMSS/test/t_tmssapp_specification_functional.py
index 2333d78c4dc5970eb3a332acb37892ac5bcd3a5c..48c38de1c0b119616bf901f13d0a9a5011062cb7 100755
--- a/SAS/TMSS/test/t_tmssapp_specification_functional.py
+++ b/SAS/TMSS/test/t_tmssapp_specification_functional.py
@@ -34,119 +34,13 @@ logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=loggin
# use setup/teardown magic for tmss test database
from lofar.sas.tmss.test.tmss_test_environment_unittest_setup import *
-from requests.auth import HTTPBasicAuth
-AUTH = HTTPBasicAuth(tmss_test_env.ldap_server.dbcreds.user, tmss_test_env.ldap_server.dbcreds.password)
-BASE_URL = tmss_test_env.django_server.url
-
-import os
-import sys
import unittest
-import requests
-import json
-from datetime import datetime
-from lofar.common import dbcredentials
-import logging
-import uuid
-
-def post_data_and_get_url(data, url_postfix):
- """POST the given data the BASE_URL+url_postfix, and return the response's url"""
- return json.loads(requests.post(BASE_URL + url_postfix, json=data, auth=AUTH).content.decode('utf-8'))['url']
-
-def _call_API_and_assert_expected_response(self, url, call, data, expected_code, expected_content):
- """
- Call API method on the provided url and assert the expected code is returned and the expected content is in the response content
- :return: response as dict. This either contains the data of an entry or error details. If JSON cannot be parsed, return string.
- """
- if call == 'PUT':
- response = requests.put(url, json=data, auth=AUTH)
- elif call == 'POST':
- response = requests.post(url, json=data, auth=AUTH)
- elif call == 'GET':
- response = requests.get(url, auth=AUTH)
- elif call == 'PATCH':
- response = requests.patch(url, json=data, auth=AUTH)
- elif call == 'DELETE':
- response = requests.delete(url, auth=AUTH)
- else:
- raise ValueError("The provided call '%s' is not a valid API method choice" % call)
-
- if response.status_code != expected_code:
- print("!!! Unexpected: [%s] - %s %s: %s" % (self.id(), call, url, response.content.decode('utf-8').strip()))
- self.assertEqual(response.status_code, expected_code)
- r_dict = json.loads(response.content.decode('utf-8'))
- for key, value in expected_content.items():
- if key not in r_dict.keys():
- print('!!! Missing key: %s in %s' % (key, r_dict.keys()))
- self.assertTrue(key in r_dict.keys())
- if type(value) is list:
- self.assertEqual(sorted(value), sorted(r_dict[key])) # compare lists independent of ordering
- else:
- self.assertEqual(value, r_dict[key])
- return r_dict
-
-def PUT_and_assert_expected_response(self, url, data, expected_code, expected_content):
- """
- PUT data on url and assert the expected code is returned and the expected content is in the response content
- """
-
- r_dict = _call_API_and_assert_expected_response(self, url, 'PUT', data, expected_code, expected_content)
- return r_dict
-
-def POST_and_assert_expected_response(self, url, data, expected_code, expected_content):
- """
- POST data on url and assert the expected code is returned and the expected content is in the response content
- :return: response dict
- """
-
- r_dict = _call_API_and_assert_expected_response(self, url, 'POST', data, expected_code, expected_content)
- return r_dict
-
-def GET_and_assert_expected_response(self, url, expected_code, expected_content):
- """
- GET from url and assert the expected code is returned and the expected content is in the response content
- """
-
- r_dict = _call_API_and_assert_expected_response(self, url, 'GET', {}, expected_code, expected_content)
- return r_dict
-
-def PATCH_and_assert_expected_response(self, url, data, expected_code, expected_content):
- """
- POST data on url and assert the provided values have changed based on the server response.
- :return: url for new item
- """
-
- r_dict = _call_API_and_assert_expected_response(self, url, 'PATCH', data, expected_code, expected_content)
- return r_dict
-
-def DELETE_and_assert_gone(self, url):
- """
- DELETE item at provided url and assert that the request was accepted by the server
- :return: url for new item
- """
- response = requests.delete(url, auth=AUTH)
- if response.status_code != 204:
- print("!!! Unexpected: [%s] - %s %s: %s" % (self.id(), 'DELETE', url, response.content))
- self.assertEqual(response.status_code, 204)
-
- response = requests.get(url, auth=AUTH)
- if response.status_code != 404:
- print("!!! Unexpected: [%s] - %s %s: %s" % (self.id(), 'GET', url, response.content))
- self.assertEqual(response.status_code, 404)
-
class BasicFunctionTestCase(unittest.TestCase):
# todo: test_welcome_page (once we have one :))
pass
-def GeneratorTemplate_test_data(name="generatortemplate"):
- return {"name": name,
- "description": 'My one observation',
- "version": 'v0.314159265359',
- "schema": {"mykey": "my value"},
- "create_function": 'Funky',
- "tags": ["TMSS", "TESTING"]}
-
class GeneratorTemplateTestCase(unittest.TestCase):
test_patch = {"version": 'v6.28318530718',
"schema": {"mykey": "my better value"},
@@ -204,13 +98,6 @@ class GeneratorTemplateTestCase(unittest.TestCase):
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
-def SchedulingUnitTemplate_test_data(name="schedulingunittemplate1"):
- return { "name": name,
- "description": 'My description',
- "version": 'v0.314159265359',
- "schema": {"mykey": "my value"},
- "tags": ["TMSS", "TESTING"]}
-
class SchedulingUnitTemplateTestCase(unittest.TestCase):
test_patch = {"version": 'v6.28318530718',
@@ -269,14 +156,6 @@ class SchedulingUnitTemplateTestCase(unittest.TestCase):
# DELETE and check it's gone
DELETE_and_assert_gone(self, url)
-def TaskTemplate_test_data(name="tasktemplate1"):
- return {"name": name,
- "description": 'My one observation',
- "version": 'v0.314159265359',
- "schema": {"mykey": "my value"},
- "tags": ["TMSS", "TESTING"],
- "validation_code_js": "???"}
-
class TaskTemplateTestCase(unittest.TestCase):
test_patch = {"version": 'v6.28318530718',
"schema": {"mykey": "my better value"},
@@ -335,13 +214,6 @@ class TaskTemplateTestCase(unittest.TestCase):
DELETE_and_assert_gone(self, url)
-def WorkRelationSelectionTemplate_test_data(name="workrelationselectiontemplate1"):
- return {"name": name,
- "description": 'My one observation',
- "version": 'v0.314159265359',
- "schema": {"mykey": "my value"},
- "tags": ["TMSS", "TESTING"]}
-
class WorkRelationSelectionTemplateTestCase(unittest.TestCase):
test_patch = {"version": 'v6.28318530718',
"schema": {"mykey": "my better value"},
@@ -400,20 +272,6 @@ class WorkRelationSelectionTemplateTestCase(unittest.TestCase):
DELETE_and_assert_gone(self, url)
-def TaskConnectors_test_data(role="correlator", input_of_url=None, output_of_url=None):
- if input_of_url is None:
- input_of_url = post_data_and_get_url(TaskTemplate_test_data(), '/task_template/')
-
- if output_of_url is None:
- output_of_url = post_data_and_get_url(TaskTemplate_test_data(), '/task_template/')
-
- return {"role": BASE_URL + '/role/%s/'%role,
- "datatype": BASE_URL + '/datatype/image/',
- "dataformats": [BASE_URL + '/dataformat/Beamformed/'],
- "output_of": output_of_url,
- "input_of": input_of_url,
- "tags": []}
-
class TaskConnectorsTestCase(unittest.TestCase):
test_patch = {"role": BASE_URL + '/role/calibrator/',
@@ -560,11 +418,6 @@ class TaskConnectorsTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 404, {})
-def DefaultTemplates_test_data(name="defaulttemplate"):
- return {"name": name,
- "template": None,
- "tags": []}
-
class DefaultTemplates(unittest.TestCase):
def test_default_generator_template_POST(self):
r_dict = POST_and_assert_expected_response(self, BASE_URL + '/generator_template/',
@@ -681,18 +534,6 @@ class DefaultTemplates(unittest.TestCase):
GET_and_assert_expected_response(self, template_url, 200, WorkRelationSelectionTemplate_test_data())
-def Cycle_test_data(description="my cycle description"):
- return {"name": 'my_cycle_' + str(uuid.uuid4()),
- "description": description,
- "tags": [],
- "start": datetime.utcnow().isoformat(),
- "stop": datetime.utcnow().isoformat(),
- "number": 1,
- "standard_hours": 2,
- "expert_hours": 3,
- "filler_hours": 4,
- "projects": []}
-
class CycleTestCase(unittest.TestCase):
test_patch = {"start": datetime(year=2015, month=10, day=21).isoformat()}
@@ -756,14 +597,6 @@ class CycleTestCase(unittest.TestCase):
DELETE_and_assert_gone(self, url)
-def Project_test_data(description="my project description"):
- return {"name": 'my_project_' + str(uuid.uuid4()),
- "description": description,
- "tags": [],
- "priority": 1,
- "can_trigger": False,
- "private_data": True}
-
class ProjectTestCase(unittest.TestCase):
test_patch = {"priority": 500,
"tags": ["SUPERIMPORTANT"]}
@@ -848,21 +681,6 @@ class ProjectTestCase(unittest.TestCase):
self.assertTrue("ProtectedError" in str(response.content))
GET_and_assert_expected_response(self, cycle_url, 200, cycle_test_data)
-def SchedulingSet_test_data(name="my_scheduling_set", project_url=None, generator_template_url=None):
- if project_url is None:
- project_url = post_data_and_get_url(Project_test_data(), '/project/')
-
- if generator_template_url is None:
- generator_template_url = post_data_and_get_url(GeneratorTemplate_test_data(), '/generator_template/')
-
- return {"name": name,
- "description": "This is my scheduling set",
- "tags": [],
- "generator_doc": "{}",
- "project": project_url,
- "generator_template": generator_template_url,
- "generator_source": None,
- "scheduling_unit_drafts": []}
class SchedulingSetTestCase(unittest.TestCase):
test_patch = {"description": "This is a new and improved description",
@@ -963,24 +781,6 @@ class SchedulingSetTestCase(unittest.TestCase):
test_data['generator_template'] = None
GET_and_assert_expected_response(self, url, 200, test_data)
-def SchedulingUnitDraft_test_data(name="my_scheduling_unit_draft", scheduling_set_url=None, template_url=None):
- if scheduling_set_url is None:
- scheduling_set_url = post_data_and_get_url(SchedulingSet_test_data(), '/scheduling_set/')
-
- if template_url is None:
- template_url = post_data_and_get_url(SchedulingUnitTemplate_test_data(), '/scheduling_unit_template/')
-
- return {"name": name,
- "description": "This is my run draft",
- "tags": [],
- "requirements_doc": "{}",
- "copy_reason": BASE_URL + '/copy_reason/template/',
- "generator_instance_doc": "{}",
- "copies": None,
- "scheduling_set": scheduling_set_url,
- "requirements_template": template_url,
- "related_scheduling_unit_blueprint": [],
- "task_drafts": []}
class SchedulingUnitDraftTestCase(unittest.TestCase):
test_patch = {"description": "This is a new and improved description",
@@ -1097,26 +897,6 @@ class SchedulingUnitDraftTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 200, test_data)
-def TaskDraft_test_data(name='my_task_draft', scheduling_unit_draft_url=None, template_url=None):
- if scheduling_unit_draft_url is None:
- scheduling_unit_draft_url = post_data_and_get_url(SchedulingUnitDraft_test_data(), '/scheduling_unit_draft/')
-
- if template_url is None:
- template_url = post_data_and_get_url(TaskTemplate_test_data(), '/task_template/')
-
- return {"name": name,
- "description": "This is my task draft",
- "tags": [],
- "specifications_doc": "{}",
- "copy_reason": BASE_URL + '/copy_reason/template/',
- "copies": None,
- "scheduling_unit_draft": scheduling_unit_draft_url,
- "specifications_template": template_url,
- 'related_task_blueprint': [],
- 'produced_by': [],
- 'consumed_by': []}
-
-
class TaskDraftTestCase(unittest.TestCase):
test_patch = {"description": "This is a new and improved description",
"specifications_doc": "{'para': 'meter'}"}
@@ -1231,31 +1011,6 @@ class TaskDraftTestCase(unittest.TestCase):
test_data['copies'] = None
GET_and_assert_expected_response(self, url, 200, test_data)
-def TaskRelationDraft_test_data(name="myTaskRelationDraft", producer_url=None, consumer_url=None, template_url=None, input_url=None, output_url=None):
- if producer_url is None:
- producer_url = post_data_and_get_url(TaskDraft_test_data(), '/task_draft/')
-
- if consumer_url is None:
- consumer_url = post_data_and_get_url(TaskDraft_test_data(),'/task_draft/')
-
- if template_url is None:
- template_url = post_data_and_get_url(WorkRelationSelectionTemplate_test_data(), '/work_relation_selection_template/')
-
- if input_url is None:
- input_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
-
- if output_url is None:
- output_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
-
- return {"tags": [],
- "selection_doc": "{}",
- "dataformat": BASE_URL + "/dataformat/Beamformed/",
- "producer": producer_url,
- "consumer": consumer_url,
- "input": input_url,
- "output": output_url,
- "selection_template": template_url,
- 'related_task_relation_blueprint': []}
class TaskRelationDraftTestCase(unittest.TestCase):
test_patch = {"selection_doc": "{'para': 'meter'}"}
@@ -1406,21 +1161,6 @@ class TaskRelationDraftTestCase(unittest.TestCase):
GET_and_assert_expected_response(self, url, 404, {})
-def SchedulingUnitBlueprint_test_data(name="my_scheduling_unit_blueprint", scheduling_unit_draft_url=None, template_url=None):
- if scheduling_unit_draft_url is None:
- scheduling_unit_draft_url = post_data_and_get_url(SchedulingUnitDraft_test_data(), '/scheduling_unit_draft/')
-
- if template_url is None:
- template_url = post_data_and_get_url(SchedulingUnitTemplate_test_data(), '/scheduling_unit_template/')
-
- return {"name": name,
- "description": "This is my run blueprint",
- "tags": [],
- "requirements_doc": "{}",
- "do_cancel": False,
- "draft": scheduling_unit_draft_url,
- "requirements_template": template_url}
-
class SchedulingUnitBlueprintTestCase(unittest.TestCase):
test_patch = {"description": "This is an updated description",
"do_cancel": True}
@@ -1513,28 +1253,6 @@ class SchedulingUnitBlueprintTestCase(unittest.TestCase):
# assert
GET_and_assert_expected_response(self, url, 404, {})
-def TaskBlueprint_test_data(name="my_TaskBlueprint", draft_url=None, template_url=None, scheduling_unit_blueprint_url=None):
- if draft_url is None:
- draft_url = post_data_and_get_url(TaskDraft_test_data(), '/task_draft/')
-
- if template_url is None:
- template_url = post_data_and_get_url(TaskTemplate_test_data(), '/task_template/')
-
- if scheduling_unit_blueprint_url is None:
- scheduling_unit_blueprint_url = post_data_and_get_url(SchedulingUnitBlueprint_test_data(), '/scheduling_unit_blueprint/')
-
- return {"name": name,
- "description": "This is my work request blueprint",
- "tags": [],
- "specifications_doc": "{}",
- "do_cancel": False,
- "draft": draft_url,
- "specifications_template": template_url,
- "scheduling_unit_blueprint": scheduling_unit_blueprint_url,
- "subtasks": [],
- "produced_by": [],
- "consumed_by": []}
-
class TaskBlueprintTestCase(unittest.TestCase):
test_patch = {"description": "This is an updated description",
"do_cancel": True}
@@ -1675,36 +1393,6 @@ class TaskBlueprintTestCase(unittest.TestCase):
# assert
GET_and_assert_expected_response(self, url, 404, {})
-def TaskRelationBlueprint_test_data(draft_url=None, template_url=None, input_url=None, output_url=None, consumer_url=None, producer_url=None):
- if draft_url is None:
- draft_url = post_data_and_get_url(TaskRelationDraft_test_data(), '/task_relation_draft/')
-
- if producer_url is None:
- producer_url = post_data_and_get_url(TaskBlueprint_test_data(), '/task_blueprint/')
-
- if consumer_url is None:
- consumer_url = post_data_and_get_url(TaskBlueprint_test_data(),'/task_blueprint/')
-
- if template_url is None:
- template_url = post_data_and_get_url(WorkRelationSelectionTemplate_test_data(), '/work_relation_selection_template/')
-
- if input_url is None:
- input_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
-
- if output_url is None:
- output_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
-
- # test data
- return {"tags": [],
- "selection_doc": "{}",
- "dataformat": BASE_URL + '/dataformat/MeasurementSet/',
- "input": input_url,
- "output": output_url,
- "draft": draft_url,
- "selection_template": template_url,
- "producer": producer_url,
- "consumer": consumer_url}
-
class TaskRelationBlueprintTestCase(unittest.TestCase):
test_patch = {"selection_doc": "{'new': 'doc'}"}
diff --git a/SAS/TMSS/test/tmss_test_environment_unittest_setup.py b/SAS/TMSS/test/tmss_test_environment_unittest_setup.py
index 72ad999bc211b8d3c2e769ca9f9b90da12e3f8d9..6b3c7a7ffbba23adb899ecfea0ca2e7077972e92 100644
--- a/SAS/TMSS/test/tmss_test_environment_unittest_setup.py
+++ b/SAS/TMSS/test/tmss_test_environment_unittest_setup.py
@@ -32,3 +32,329 @@ tmss_test_env.start()
# tell unittest to stop (and automagically cleanup) the test database once all testing is done.
def tearDownModule():
tmss_test_env.stop()
+
+
+
+################################################################################################
+# the methods below can be used to to HTTP REST calls to the django server and check the results
+################################################################################################
+
+from requests.auth import HTTPBasicAuth
+AUTH = HTTPBasicAuth(tmss_test_env.ldap_server.dbcreds.user, tmss_test_env.ldap_server.dbcreds.password)
+BASE_URL = tmss_test_env.django_server.url
+
+import requests
+import json
+
+def post_data_and_get_url(data, url_postfix):
+ """POST the given data the BASE_URL+url_postfix, and return the response's url"""
+ return json.loads(requests.post(BASE_URL + url_postfix, json=data, auth=AUTH).content.decode('utf-8'))['url']
+
+def _call_API_and_assert_expected_response(test_instance, url, call, data, expected_code, expected_content):
+ """
+ Call API method on the provided url and assert the expected code is returned and the expected content is in the response content
+ :return: response as dict. This either contains the data of an entry or error details. If JSON cannot be parsed, return string.
+ """
+ if call == 'PUT':
+ response = requests.put(url, json=data, auth=AUTH)
+ elif call == 'POST':
+ response = requests.post(url, json=data, auth=AUTH)
+ elif call == 'GET':
+ response = requests.get(url, auth=AUTH)
+ elif call == 'PATCH':
+ response = requests.patch(url, json=data, auth=AUTH)
+ elif call == 'DELETE':
+ response = requests.delete(url, auth=AUTH)
+ else:
+ raise ValueError("The provided call '%s' is not a valid API method choice" % call)
+
+ if response.status_code != expected_code:
+ print("!!! Unexpected: [%s] - %s %s: %s" % (test_instance.id(), call, url, response.content.decode('utf-8').strip()))
+ test_instance.assertEqual(response.status_code, expected_code)
+ r_dict = json.loads(response.content.decode('utf-8'))
+ for key, value in expected_content.items():
+ if key not in r_dict.keys():
+ print('!!! Missing key: %s in %s' % (key, r_dict.keys()))
+ test_instance.assertTrue(key in r_dict.keys())
+ if type(value) is list:
+ test_instance.assertEqual(sorted(value), sorted(r_dict[key])) # compare lists independent of ordering
+ else:
+ test_instance.assertEqual(value, r_dict[key])
+ return r_dict
+
+def PUT_and_assert_expected_response(test_instance, url, data, expected_code, expected_content):
+ """
+ PUT data on url and assert the expected code is returned and the expected content is in the response content
+ """
+
+ r_dict = _call_API_and_assert_expected_response(test_instance, url, 'PUT', data, expected_code, expected_content)
+ return r_dict
+
+def POST_and_assert_expected_response(test_instance, url, data, expected_code, expected_content):
+ """
+ POST data on url and assert the expected code is returned and the expected content is in the response content
+ :return: response dict
+ """
+
+ r_dict = _call_API_and_assert_expected_response(test_instance, url, 'POST', data, expected_code, expected_content)
+ return r_dict
+
+def GET_and_assert_expected_response(test_instance, url, expected_code, expected_content):
+ """
+ GET from url and assert the expected code is returned and the expected content is in the response content
+ """
+
+ r_dict = _call_API_and_assert_expected_response(test_instance, url, 'GET', {}, expected_code, expected_content)
+ return r_dict
+
+def PATCH_and_assert_expected_response(test_instance, url, data, expected_code, expected_content):
+ """
+ POST data on url and assert the provided values have changed based on the server response.
+ :return: url for new item
+ """
+
+ r_dict = _call_API_and_assert_expected_response(test_instance, url, 'PATCH', data, expected_code, expected_content)
+ return r_dict
+
+def DELETE_and_assert_gone(test_instance, url):
+ """
+ DELETE item at provided url and assert that the request was accepted by the server
+ :return: url for new item
+ """
+ response = requests.delete(url, auth=AUTH)
+ if response.status_code != 204:
+ print("!!! Unexpected: [%s] - %s %s: %s" % (test_instance.id(), 'DELETE', url, response.content))
+ test_instance.assertEqual(response.status_code, 204)
+
+ response = requests.get(url, auth=AUTH)
+ if response.status_code != 404:
+ print("!!! Unexpected: [%s] - %s %s: %s" % (test_instance.id(), 'GET', url, response.content))
+ test_instance.assertEqual(response.status_code, 404)
+
+
+
+###################################################
+# the methods below can be used to create test data
+###################################################
+
+from datetime import datetime
+import uuid
+
+def GeneratorTemplate_test_data(name="generatortemplate"):
+ return {"name": name,
+ "description": 'My one observation',
+ "version": 'v0.314159265359',
+ "schema": {"mykey": "my value"},
+ "create_function": 'Funky',
+ "tags": ["TMSS", "TESTING"]}
+
+def SchedulingUnitTemplate_test_data(name="schedulingunittemplate1"):
+ return { "name": name,
+ "description": 'My description',
+ "version": 'v0.314159265359',
+ "schema": {"mykey": "my value"},
+ "tags": ["TMSS", "TESTING"]}
+
+def TaskTemplate_test_data(name="tasktemplate1"):
+ return {"name": name,
+ "description": 'My one observation',
+ "version": 'v0.314159265359',
+ "schema": {"mykey": "my value"},
+ "tags": ["TMSS", "TESTING"],
+ "validation_code_js": "???"}
+
+def WorkRelationSelectionTemplate_test_data(name="workrelationselectiontemplate1"):
+ return {"name": name,
+ "description": 'My one observation',
+ "version": 'v0.314159265359',
+ "schema": {"mykey": "my value"},
+ "tags": ["TMSS", "TESTING"]}
+
+def TaskConnectors_test_data(role="correlator", input_of_url=None, output_of_url=None):
+ if input_of_url is None:
+ input_of_url = post_data_and_get_url(TaskTemplate_test_data(), '/task_template/')
+
+ if output_of_url is None:
+ output_of_url = post_data_and_get_url(TaskTemplate_test_data(), '/task_template/')
+
+ return {"role": BASE_URL + '/role/%s/'%role,
+ "datatype": BASE_URL + '/datatype/image/',
+ "dataformats": [BASE_URL + '/dataformat/Beamformed/'],
+ "output_of": output_of_url,
+ "input_of": input_of_url,
+ "tags": []}
+
+def DefaultTemplates_test_data(name="defaulttemplate"):
+ return {"name": name,
+ "template": None,
+ "tags": []}
+
+def Cycle_test_data(description="my cycle description"):
+ return {"name": 'my_cycle_' + str(uuid.uuid4()),
+ "description": description,
+ "tags": [],
+ "start": datetime.utcnow().isoformat(),
+ "stop": datetime.utcnow().isoformat(),
+ "number": 1,
+ "standard_hours": 2,
+ "expert_hours": 3,
+ "filler_hours": 4,
+ "projects": []}
+
+def Project_test_data(description="my project description"):
+ return {"name": 'my_project_' + str(uuid.uuid4()),
+ "description": description,
+ "tags": [],
+ "priority": 1,
+ "can_trigger": False,
+ "private_data": True}
+
+def SchedulingSet_test_data(name="my_scheduling_set", project_url=None, generator_template_url=None):
+ if project_url is None:
+ project_url = post_data_and_get_url(Project_test_data(), '/project/')
+
+ if generator_template_url is None:
+ generator_template_url = post_data_and_get_url(GeneratorTemplate_test_data(), '/generator_template/')
+
+ return {"name": name,
+ "description": "This is my scheduling set",
+ "tags": [],
+ "generator_doc": "{}",
+ "project": project_url,
+ "generator_template": generator_template_url,
+ "generator_source": None,
+ "scheduling_unit_drafts": []}
+
+def SchedulingUnitDraft_test_data(name="my_scheduling_unit_draft", scheduling_set_url=None, template_url=None):
+ if scheduling_set_url is None:
+ scheduling_set_url = post_data_and_get_url(SchedulingSet_test_data(), '/scheduling_set/')
+
+ if template_url is None:
+ template_url = post_data_and_get_url(SchedulingUnitTemplate_test_data(), '/scheduling_unit_template/')
+
+ return {"name": name,
+ "description": "This is my run draft",
+ "tags": [],
+ "requirements_doc": "{}",
+ "copy_reason": BASE_URL + '/copy_reason/template/',
+ "generator_instance_doc": "{}",
+ "copies": None,
+ "scheduling_set": scheduling_set_url,
+ "requirements_template": template_url,
+ "related_scheduling_unit_blueprint": [],
+ "task_drafts": []}
+
+def TaskDraft_test_data(name='my_task_draft', scheduling_unit_draft_url=None, template_url=None):
+ if scheduling_unit_draft_url is None:
+ scheduling_unit_draft_url = post_data_and_get_url(SchedulingUnitDraft_test_data(), '/scheduling_unit_draft/')
+
+ if template_url is None:
+ template_url = post_data_and_get_url(TaskTemplate_test_data(), '/task_template/')
+
+ return {"name": name,
+ "description": "This is my task draft",
+ "tags": [],
+ "specifications_doc": "{}",
+ "copy_reason": BASE_URL + '/copy_reason/template/',
+ "copies": None,
+ "scheduling_unit_draft": scheduling_unit_draft_url,
+ "specifications_template": template_url,
+ 'related_task_blueprint': [],
+ 'produced_by': [],
+ 'consumed_by': []}
+
+
+def TaskRelationDraft_test_data(name="myTaskRelationDraft", producer_url=None, consumer_url=None, template_url=None, input_url=None, output_url=None):
+ if producer_url is None:
+ producer_url = post_data_and_get_url(TaskDraft_test_data(), '/task_draft/')
+
+ if consumer_url is None:
+ consumer_url = post_data_and_get_url(TaskDraft_test_data(),'/task_draft/')
+
+ if template_url is None:
+ template_url = post_data_and_get_url(WorkRelationSelectionTemplate_test_data(), '/work_relation_selection_template/')
+
+ if input_url is None:
+ input_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
+
+ if output_url is None:
+ output_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
+
+ return {"tags": [],
+ "selection_doc": "{}",
+ "dataformat": BASE_URL + "/dataformat/Beamformed/",
+ "producer": producer_url,
+ "consumer": consumer_url,
+ "input": input_url,
+ "output": output_url,
+ "selection_template": template_url,
+ 'related_task_relation_blueprint': []}
+
+def SchedulingUnitBlueprint_test_data(name="my_scheduling_unit_blueprint", scheduling_unit_draft_url=None, template_url=None):
+ if scheduling_unit_draft_url is None:
+ scheduling_unit_draft_url = post_data_and_get_url(SchedulingUnitDraft_test_data(), '/scheduling_unit_draft/')
+
+ if template_url is None:
+ template_url = post_data_and_get_url(SchedulingUnitTemplate_test_data(), '/scheduling_unit_template/')
+
+ return {"name": name,
+ "description": "This is my run blueprint",
+ "tags": [],
+ "requirements_doc": "{}",
+ "do_cancel": False,
+ "draft": scheduling_unit_draft_url,
+ "requirements_template": template_url}
+
+def TaskBlueprint_test_data(name="my_TaskBlueprint", draft_url=None, template_url=None, scheduling_unit_blueprint_url=None):
+ if draft_url is None:
+ draft_url = post_data_and_get_url(TaskDraft_test_data(), '/task_draft/')
+
+ if template_url is None:
+ template_url = post_data_and_get_url(TaskTemplate_test_data(), '/task_template/')
+
+ if scheduling_unit_blueprint_url is None:
+ scheduling_unit_blueprint_url = post_data_and_get_url(SchedulingUnitBlueprint_test_data(), '/scheduling_unit_blueprint/')
+
+ return {"name": name,
+ "description": "This is my work request blueprint",
+ "tags": [],
+ "specifications_doc": "{}",
+ "do_cancel": False,
+ "draft": draft_url,
+ "specifications_template": template_url,
+ "scheduling_unit_blueprint": scheduling_unit_blueprint_url,
+ "subtasks": [],
+ "produced_by": [],
+ "consumed_by": []}
+
+def TaskRelationBlueprint_test_data(draft_url=None, template_url=None, input_url=None, output_url=None, consumer_url=None, producer_url=None):
+ if draft_url is None:
+ draft_url = post_data_and_get_url(TaskRelationDraft_test_data(), '/task_relation_draft/')
+
+ if producer_url is None:
+ producer_url = post_data_and_get_url(TaskBlueprint_test_data(), '/task_blueprint/')
+
+ if consumer_url is None:
+ consumer_url = post_data_and_get_url(TaskBlueprint_test_data(),'/task_blueprint/')
+
+ if template_url is None:
+ template_url = post_data_and_get_url(WorkRelationSelectionTemplate_test_data(), '/work_relation_selection_template/')
+
+ if input_url is None:
+ input_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
+
+ if output_url is None:
+ output_url = post_data_and_get_url(TaskConnectors_test_data(), '/task_connectors/')
+
+ # test data
+ return {"tags": [],
+ "selection_doc": "{}",
+ "dataformat": BASE_URL + '/dataformat/MeasurementSet/',
+ "input": input_url,
+ "output": output_url,
+ "draft": draft_url,
+ "selection_template": template_url,
+ "producer": producer_url,
+ "consumer": consumer_url}
+
+