diff --git a/SAS/LSMR/src/lsmr/lsmrapp/models/specification.py b/SAS/LSMR/src/lsmr/lsmrapp/models/specification.py
index 79bef909d9b2b8a0da9ccd29a3033186ec4de57e..7c0f89963882595ebb87dbefef03e6eb7c0c7ec5 100644
--- a/SAS/LSMR/src/lsmr/lsmrapp/models/specification.py
+++ b/SAS/LSMR/src/lsmr/lsmrapp/models/specification.py
@@ -113,8 +113,8 @@ class CopyReasonChoice(AbstractChoice):
"""Defines the model and predefined list of possible CopyReasonChoice's for RunDraft and WorkRequestDraft.
The items in the Choises class below are automagically populated into the database via a data migration."""
class Choices(Enum):
- TEMPLATE = "TEMPLATE"
- REPEATED = "REPEATED"
+ TEMPLATE = "template"
+ REPEATED = "repeated"
# concrete models
diff --git a/SAS/LSMR/test/t_functional.py b/SAS/LSMR/test/t_functional.py
index 8e50404768b4d4a92acdd93a1ab4a3ba354bbd1f..ab1d862dabd7c4f1ca027090b4d519257ece1139 100755
--- a/SAS/LSMR/test/t_functional.py
+++ b/SAS/LSMR/test/t_functional.py
@@ -31,6 +31,7 @@
import unittest
import requests
import json
+from datetime import datetime
BASE_URL = 'http://localhost:8777'
@@ -108,7 +109,7 @@ def DELETE_and_assert_gone(self, url):
response = requests.get(url)
if response.status_code != 404:
- print("[%s] - %s %s: %s" % (self.id(), 'GET', url, response.content))
+ print("!!! Unexpected: [%s] - %s %s: %s" % (self.id(), 'GET', url, response.content))
self.assertEqual(response.status_code, 404)
@@ -584,6 +585,845 @@ class DefaultTemplates(unittest.TestCase):
POST_and_assert_expected_response(self, BASE_URL + '/default_work_relation_selection_template/', test_data_1, 201, test_data_1)
+class CycleTestCase(unittest.TestCase):
+
+ # test data
+ test_data_1 = {"name": 'my_cycle',
+ "description": "my one cycle",
+ "tags": [],
+ "start": datetime.utcnow().isoformat()}
+
+ test_data_2 = {"name": 'my_other_cycle',
+ "description": "This is my other cycle",
+ "tags": ['othercycle'],
+ "start": datetime.utcnow().isoformat()}
+
+ 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')
+ self.assertEqual(r.status_code, 200)
+ self.assertTrue("Cycle List" in r.content.decode('utf8'))
+
+ def test_cycle_GET_nonexistant_raises_error(self):
+ GET_and_assert_expected_response(self, BASE_URL + '/cycle/1234321/', 404, {})
+
+ def test_cycle_POST_and_GET(self):
+
+ # POST and GET a new item and assert correctness
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ def test_cycle_PUT_invalid_raises_error(self):
+ PUT_and_assert_expected_response(self, BASE_URL + '/cycle/9876789876/', self.test_data_1, 404, {})
+
+ def test_cycle_PUT(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+
+ def test_cycle_PATCH(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+ GET_and_assert_expected_response(self, url, 200, expected_data)
+
+ def test_cycle_DELETE(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/cycle/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # DELETE and check it's gone
+ DELETE_and_assert_gone(self, url)
+
+
+class ProjectTestCase(unittest.TestCase):
+
+ # test data
+ test_data_1 = {"name": 'my_project',
+ "description": "This is my project",
+ "tags": [],
+ "priority": 1,
+ "can_trigger": False,
+ "private_data": True}
+
+ test_data_2 = {"name": 'my_other_project',
+ "description": "This is my other project",
+ "tags": ['othercycle'],
+ "priority": 42,
+ "can_trigger": True,
+ "private_data": False}
+
+ test_patch = {"priority": 500,
+ "tags": ["SUPERIMPORTANT"]}
+
+ def test_project_list_apiformat(self):
+ r = requests.get(BASE_URL + '/project/?format=api')
+ self.assertEqual(r.status_code, 200)
+ self.assertTrue("Project List" in r.content.decode('utf8'))
+
+ def test_project_GET_nonexistant_raises_error(self):
+ GET_and_assert_expected_response(self, BASE_URL + '/project/1234321/', 404, {})
+
+ def test_project_POST_and_GET(self):
+
+ # POST and GET a new item and assert correctness
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ def test_project_PUT_invalid_raises_error(self):
+ PUT_and_assert_expected_response(self, BASE_URL + '/project/9876789876/', self.test_data_1, 404, {})
+
+ def test_project_PUT(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+
+ def test_project_PATCH(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+ GET_and_assert_expected_response(self, url, 200, expected_data)
+
+ def test_project_DELETE(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/project/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # DELETE and check it's gone
+ DELETE_and_assert_gone(self, url)
+
+
+class RunSetTestCase(unittest.TestCase):
+
+ def setUp(self):
+
+ # related items
+ project_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/project/',
+ ProjectTestCase.test_data_1, 201,
+ ProjectTestCase.test_data_1)['url']
+
+ template_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/generator_template/',
+ GeneratorTemplateTestCase.test_data_1, 201,
+ GeneratorTemplateTestCase.test_data_1)['url']
+
+ # test data
+ self.test_data_1 = {"name": 'my_run_set',
+ "description": "This is my run set",
+ "tags": [],
+ "generator_parameters_doc": "{}",
+ "project": project_url,
+ "generator_template": None}
+
+ # test data
+ self.test_data_2 = {"name": 'my_other_run_set',
+ "description": "This is my other run set",
+ "tags": ['othercycle'],
+ "generator_parameters_doc": "{}",
+ "project": project_url,
+ "generator_template": template_url}
+
+ self.test_patch = {"description": "This is a new and improved description",
+ "generator_parameters_doc": "{'para': 'meter'}"}
+
+ def test_run_set_list_apiformat(self):
+ r = requests.get(BASE_URL + '/run_set/?format=api')
+ self.assertEqual(r.status_code, 200)
+ self.assertTrue("Run Set List" in r.content.decode('utf8'))
+
+ def test_run_set_GET_nonexistant_raises_error(self):
+ GET_and_assert_expected_response(self, BASE_URL + '/run_set/1234321/', 404, {})
+
+ def test_run_set_POST_and_GET(self):
+
+ # POST and GET a new item and assert correctness
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/run_set/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ def test_run_set_PUT_invalid_raises_error(self):
+ PUT_and_assert_expected_response(self, BASE_URL + '/run_set/9876789876/', self.test_data_1, 404, {})
+
+ def test_run_set_PUT(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/run_set/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+
+ def test_run_set_PATCH(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/run_set/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+ GET_and_assert_expected_response(self, url, 200, expected_data)
+
+ def test_run_set_DELETE(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/run_set/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # DELETE and check it's gone
+ DELETE_and_assert_gone(self, url)
+
+
+
+class RunDraftTestCase(unittest.TestCase):
+
+ def setUp(self):
+
+ # related items
+
+ rstc = RunSetTestCase()
+ rstc.setUp()
+
+ run_set_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/run_set/',
+ rstc.test_data_1, 201,
+ rstc.test_data_1)['url']
+
+ template_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/run_template/',
+ RunTemplateTestCase.test_data_1, 201,
+ RunTemplateTestCase.test_data_1)['url']
+
+ # test data
+
+ self.test_data_1 = {"name": 'my_run_draft',
+ "description": "This is my run draft",
+ "tags": [],
+ "requirements_doc": "{}",
+ "copy_reason": BASE_URL + '/copy_reason_choice/template/',
+ "generator_param": "para",
+ "copies": None,
+ "run_set": run_set_url,
+ "generator_source": None,
+ "template": template_url}
+
+ # test data
+ self.test_data_2 = {"name": 'my_other_run_draft',
+ "description": "This is my other run draft",
+ "tags": [],
+ "requirements_doc": "{}",
+ "copy_reason": BASE_URL + '/copy_reason_choice/repeated/',
+ "generator_param": "meter",
+ "copies": None,
+ "run_set": run_set_url,
+ "generator_source": None,
+ "template": template_url}
+
+ self.test_patch = {"description": "This is a new and improved description",
+ "requirements_doc": "{'para': 'meter'}"}
+
+ def test_run_draft_list_apiformat(self):
+ r = requests.get(BASE_URL + '/run_draft/?format=api')
+ self.assertEqual(r.status_code, 200)
+ self.assertTrue("Run Draft List" in r.content.decode('utf8'))
+
+ def test_run_draft_GET_nonexistant_raises_error(self):
+ GET_and_assert_expected_response(self, BASE_URL + '/run_draft/1234321/', 404, {})
+
+ def test_run_draft_POST_and_GET(self):
+
+ # POST and GET a new item and assert correctness
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/run_draft/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ def test_run_draft_PUT_invalid_raises_error(self):
+ PUT_and_assert_expected_response(self, BASE_URL + '/run_draft/9876789876/', self.test_data_1, 404, {})
+
+ def test_run_draft_PUT(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/run_draft/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+
+ def test_run_draft_PATCH(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/run_draft/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+ GET_and_assert_expected_response(self, url, 200, expected_data)
+
+ def test_run_draft_DELETE(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/run_draft/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # DELETE and check it's gone
+ DELETE_and_assert_gone(self, url)
+
+
+class WorkRequestDraftTestCase(unittest.TestCase):
+
+ def setUp(self):
+
+ # related items
+
+ rdtc = RunDraftTestCase()
+ rdtc.setUp()
+
+ run_draft_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/run_draft/',
+ rdtc.test_data_1, 201,
+ rdtc.test_data_1)['url']
+
+ template_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/work_request_template/',
+ WorkRequestTemplateTestCase.test_data_1, 201,
+ WorkRequestTemplateTestCase.test_data_1)['url']
+
+ # test data
+
+ self.test_data_1 = {"name": 'my_work_request_draft',
+ "description": "This is my work request draft",
+ "tags": [],
+ "requirements_doc": "{}",
+ "copy_reason": BASE_URL + '/copy_reason_choice/template/',
+ "copies": None,
+ "run_draft": run_draft_url,
+ "template": template_url}
+
+ # test data
+ self.test_data_2 = {"name": 'my_other_work_request_draft',
+ "description": "This is my other work request draft",
+ "tags": [],
+ "requirements_doc": "{}",
+ "copy_reason": BASE_URL + '/copy_reason_choice/template/',
+ "copies": None,
+ "run_draft": run_draft_url,
+ "template": template_url}
+
+ self.test_patch = {"description": "This is a new and improved description",
+ "requirements_doc": "{'para': 'meter'}"}
+
+ def test_work_request_draft_list_apiformat(self):
+ r = requests.get(BASE_URL + '/work_request_draft/?format=api')
+ self.assertEqual(r.status_code, 200)
+ self.assertTrue("Work Request Draft List" in r.content.decode('utf8'))
+
+ def test_work_request_draft_GET_nonexistant_raises_error(self):
+ GET_and_assert_expected_response(self, BASE_URL + '/work_request_draft/1234321/', 404, {})
+
+ def test_work_request_draft_POST_and_GET(self):
+
+ # POST and GET a new item and assert correctness
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_draft/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ def test_work_request_draft_PUT_invalid_raises_error(self):
+ PUT_and_assert_expected_response(self, BASE_URL + '/work_request_draft/9876789876/', self.test_data_1, 404, {})
+
+ def test_work_request_draft_PUT(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_draft/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+
+ def test_work_request_draft_PATCH(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_draft/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+ GET_and_assert_expected_response(self, url, 200, expected_data)
+
+ def test_work_request_draft_DELETE(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_draft/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # DELETE and check it's gone
+ DELETE_and_assert_gone(self, url)
+
+
+class WorkRequestRelationDraftTestCase(unittest.TestCase):
+
+ def setUp(self):
+
+ # related items
+
+ wrdtc = WorkRequestDraftTestCase()
+ wrdtc.setUp()
+
+ wiortc = WorkIORoleTestCase()
+ wiortc.setUp()
+
+ work_request_draft_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/work_request_draft/',
+ wrdtc.test_data_1, 201,
+ wrdtc.test_data_1)['url']
+
+ template_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/work_relation_selection_template/',
+ WorkRelationSelectionTemplateTestCase.test_data_1, 201,
+ WorkRelationSelectionTemplateTestCase.test_data_1)['url']
+
+ work_io_role_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/work_io_role/',
+ wiortc.test_data_1, 201,
+ wiortc.test_data_1)['url']
+
+ # test data
+
+ # test data
+ self.test_data_1 = {"tags": [],
+ "selection_doc": "{}",
+ "dataformat": BASE_URL + "/dataformat_choice/HDF5/",
+ "producer": work_request_draft_url,
+ "consumer": work_request_draft_url,
+ "input_role": work_io_role_url,
+ "output_role": work_io_role_url,
+ "selection_template": template_url}
+
+ # test data
+ self.test_data_2 = {"tags": [],
+ "selection_doc": "{}",
+ "dataformat": BASE_URL + "/dataformat_choice/HDF5/",
+ "producer": work_request_draft_url,
+ "consumer": work_request_draft_url,
+ "input_role": work_io_role_url,
+ "output_role": work_io_role_url,
+ "selection_template": template_url}
+
+ self.test_patch = {"selection_doc": "{'para': 'meter'}"}
+
+ def test_work_request_relation_draft_list_apiformat(self):
+ r = requests.get(BASE_URL + '/work_request_relation_draft/?format=api')
+ self.assertEqual(r.status_code, 200)
+ self.assertTrue("Work Request Relation Draft List" in r.content.decode('utf8'))
+
+ def test_work_request_relation_draft_GET_nonexistant_raises_error(self):
+ GET_and_assert_expected_response(self, BASE_URL + '/work_request_relation_draft/1234321/', 404, {})
+
+ def test_work_request_relation_draft_POST_and_GET(self):
+
+ # POST and GET a new item and assert correctness
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_relation_draft/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ def test_work_request_relation_draft_PUT_invalid_raises_error(self):
+ PUT_and_assert_expected_response(self, BASE_URL + '/work_request_relation_draft/9876789876/', self.test_data_1, 404, {})
+
+ def test_work_request_relation_draft_PUT(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_relation_draft/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+
+ def test_work_request_relation_draft_PATCH(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_relation_draft/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+ GET_and_assert_expected_response(self, url, 200, expected_data)
+
+ def test_work_request_relation_draft_DELETE(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_relation_draft/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # DELETE and check it's gone
+ DELETE_and_assert_gone(self, url)
+
+
+class RunBlueprintTestCase(unittest.TestCase):
+
+ def setUp(self):
+
+ # related items
+
+ rdtc = RunDraftTestCase()
+ rdtc.setUp()
+
+ run_draft_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/run_draft/',
+ rdtc.test_data_1, 201,
+ rdtc.test_data_1)['url']
+
+ template_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/run_template/',
+ RunTemplateTestCase.test_data_1, 201,
+ RunTemplateTestCase.test_data_1)['url']
+
+ # test data
+ self.test_data_1 = {"name": 'my_run_blueprint',
+ "description": "This is my run blueprint",
+ "tags": [],
+ "requirements_doc": "{}",
+ "do_cancel": False,
+ "draft": run_draft_url,
+ "template": template_url}
+
+ # test data
+ self.test_data_2 = {"name": 'my_other_run_blueprint',
+ "description": "This is my other run blueprint",
+ "tags": [],
+ "requirements_doc": "{}",
+ "do_cancel": True,
+ "draft": run_draft_url,
+ "template": template_url}
+
+ self.test_patch = {"description": "This is an updated description",
+ "do_cancel": True}
+
+ def test_run_blueprint_list_apiformat(self):
+ r = requests.get(BASE_URL + '/run_blueprint/?format=api')
+ self.assertEqual(r.status_code, 200)
+ self.assertTrue("Run Blueprint List" in r.content.decode('utf8'))
+
+ def test_run_blueprint_GET_nonexistant_raises_error(self):
+ GET_and_assert_expected_response(self, BASE_URL + '/run_blueprint/1234321/', 404, {})
+
+ def test_run_blueprint_POST_and_GET(self):
+
+ # POST and GET a new item and assert correctness
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/run_blueprint/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ def test_run_blueprint_PUT_invalid_raises_error(self):
+ PUT_and_assert_expected_response(self, BASE_URL + '/run_blueprint/9876789876/', self.test_data_1, 404, {})
+
+ def test_run_blueprint_PUT(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/run_blueprint/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+
+ def test_run_blueprint_PATCH(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/run_blueprint/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+ GET_and_assert_expected_response(self, url, 200, expected_data)
+
+ def test_run_blueprint_DELETE(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/run_blueprint/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # DELETE and check it's gone
+ DELETE_and_assert_gone(self, url)
+
+
+class WorkRequestBlueprintTestCase(unittest.TestCase):
+
+ def setUp(self):
+
+ # related items
+
+ wrdtc = WorkRequestDraftTestCase()
+ wrdtc.setUp()
+
+ rbtc = RunBlueprintTestCase()
+ rbtc.setUp()
+
+
+ draft_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/work_request_draft/',
+ wrdtc.test_data_1, 201,
+ wrdtc.test_data_1)['url']
+
+ template_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/work_request_template/',
+ WorkRequestTemplateTestCase.test_data_1, 201,
+ WorkRequestTemplateTestCase.test_data_1)['url']
+
+ run_blueprint_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/run_blueprint/',
+ rbtc.test_data_1, 201,
+ rbtc.test_data_1)['url']
+
+ # test data
+ self.test_data_1 = {"name": 'my_work_request_blueprint',
+ "description": "This is my work request blueprint",
+ "tags": [],
+ "requirements_doc": "{}",
+ "do_cancel": False,
+ "draft": draft_url,
+ "template": template_url,
+ "run_blueprint": run_blueprint_url}
+
+ # test data
+ self.test_data_2 = {"name": 'my_other_work_request_blueprin', # <- Missing a 't'? Well, varchar(30) is shorter than you'd think...
+ "description": "This is my other work request blueprint",
+ "tags": [],
+ "requirements_doc": "{}",
+ "do_cancel": True,
+ "draft": draft_url,
+ "template": template_url,
+ "run_blueprint": run_blueprint_url}
+
+ self.test_patch = {"description": "This is an updated description",
+ "do_cancel": True}
+
+ def test_work_request_blueprint_list_apiformat(self):
+ r = requests.get(BASE_URL + '/work_request_blueprint/?format=api')
+ self.assertEqual(r.status_code, 200)
+ self.assertTrue("Work Request Blueprint List" in r.content.decode('utf8'))
+
+ def test_work_request_blueprint_GET_nonexistant_raises_error(self):
+ GET_and_assert_expected_response(self, BASE_URL + '/work_request_blueprint/1234321/', 404, {})
+
+ def test_work_request_blueprint_POST_and_GET(self):
+
+ # POST and GET a new item and assert correctness
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_blueprint/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ def test_work_request_blueprint_PUT_invalid_raises_error(self):
+ PUT_and_assert_expected_response(self, BASE_URL + '/work_request_blueprint/9876789876/', self.test_data_1, 404, {})
+
+ def test_work_request_blueprint_PUT(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_blueprint/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+
+ def test_work_request_blueprint_PATCH(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_blueprint/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+ GET_and_assert_expected_response(self, url, 200, expected_data)
+
+ def test_work_request_blueprint_DELETE(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_blueprint/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # DELETE and check it's gone
+ DELETE_and_assert_gone(self, url)
+
+
+class WorkRequestRelationBlueprintTestCase(unittest.TestCase):
+
+ def setUp(self):
+
+ # related items
+
+ wrrdtc = WorkRequestRelationDraftTestCase()
+ wrrdtc.setUp()
+
+ wiortc = WorkIORoleTestCase()
+ wiortc.setUp()
+
+ wrbtc = WorkRequestBlueprintTestCase()
+ wrbtc.setUp()
+
+ draft_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/work_request_relation_draft/',
+ wrrdtc.test_data_1, 201,
+ wrrdtc.test_data_1)['url']
+
+ template_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/work_relation_selection_template/',
+ WorkRelationSelectionTemplateTestCase.test_data_1, 201,
+ WorkRelationSelectionTemplateTestCase.test_data_1)['url']
+
+ work_io_role_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/work_io_role/',
+ wiortc.test_data_1, 201,
+ wiortc.test_data_1)['url']
+
+ work_request_blueprint_url = POST_and_assert_expected_response(self,
+ BASE_URL + '/work_request_blueprint/',
+ wrbtc.test_data_1, 201,
+ wrbtc.test_data_1)['url']
+
+ # test data
+ self.test_data_1 = {"tags": [],
+ "selection_doc": "{}",
+ "dataformat": BASE_URL + '/dataformat_choice/MeasurementSet/',
+ "input_role": work_io_role_url,
+ "output_role": work_io_role_url,
+ "draft": draft_url,
+ "selection_template": template_url,
+ "producer": work_request_blueprint_url,
+ "consumer": work_request_blueprint_url}
+
+ # test data
+ self.test_data_2 = {"tags": ['GUTEN', 'TAG'],
+ "selection_doc": "{}",
+ "dataformat": BASE_URL + '/dataformat_choice/HDF5/',
+ "input_role": work_io_role_url,
+ "output_role": work_io_role_url,
+ "draft": draft_url,
+ "selection_template": template_url,
+ "producer": work_request_blueprint_url,
+ "consumer": work_request_blueprint_url}
+
+ self.test_patch = {"selection_doc": "{'new': 'doc'}"}
+
+ def test_work_request_relation_blueprint_list_apiformat(self):
+ r = requests.get(BASE_URL + '/work_request_relation_blueprint/?format=api')
+ self.assertEqual(r.status_code, 200)
+ self.assertTrue("Work Request Relation Blueprint List" in r.content.decode('utf8'))
+
+ def test_work_request_relation_blueprint_GET_nonexistant_raises_error(self):
+ GET_and_assert_expected_response(self, BASE_URL + '/work_request_relation_blueprint/1234321/', 404, {})
+
+ def test_work_request_relation_blueprint_POST_and_GET(self):
+
+ # POST and GET a new item and assert correctness
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_relation_blueprint/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ def test_work_request_relation_blueprint_PUT_invalid_raises_error(self):
+ PUT_and_assert_expected_response(self, BASE_URL + '/work_request_relation_blueprint/9876789876/', self.test_data_1, 404, {})
+
+ def test_work_request_relation_blueprint_PUT(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_relation_blueprint/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+
+ def test_work_request_relation_blueprint_PATCH(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_relation_blueprint/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # 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)
+ GET_and_assert_expected_response(self, url, 200, expected_data)
+
+ def test_work_request_relation_blueprint_DELETE(self):
+
+ # POST new item, verify
+ r_dict = POST_and_assert_expected_response(self, BASE_URL + '/work_request_relation_blueprint/', self.test_data_1, 201, self.test_data_1)
+ url = r_dict['url']
+ GET_and_assert_expected_response(self, url, 200, self.test_data_1)
+
+ # DELETE and check it's gone
+ DELETE_and_assert_gone(self, url)
+
+
+
if __name__ == "__main__":
unittest.main()