Loading taskmanager/__main__.py +3 −0 Original line number Diff line number Diff line Loading @@ -109,7 +109,10 @@ def add_task(args=None): with open(args.task) as task_json: task_info['content'] = task_json.read() if isinstance(task_info['content'], list): task_info['template'] = json.loads(task_info['content'])[0]['template'] else: task_info['template'] = json.loads(task_info['content'])['template'] print("task_info: {}".format(task_info)) Loading taskmanager/api/tests/v1/test_api_lock.py 0 → 100644 +136 −0 Original line number Diff line number Diff line # Copyright 2017 Biomedical Imaging Group Rotterdam, Departments of # Medical Informatics and Radiology, Erasmus MC, Rotterdam, The Netherlands # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import re from flask import url_for from flask_security.utils import hash_password from taskmanager.tests.helpers import basic_auth_authorization_header TIMESTAMP_REGEX = r'\d\d\d\d-\d\d-\d\dT\d\d:\d\d:\d\d' def test_lock_get_as_admin(client, random_test_data): """ An admin user should be able to get the tags from a task. """ headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.lock', id=1), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_lock_get_as_superuser(client, random_test_data): """ An admin user should be able to get the tags from a task. """ headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.lock', id=2), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_lock_get_as_user(client, random_test_data): """ A user should be able to get the tags from a task assigned to him/her. """ headers = basic_auth_authorization_header('user1', 'user1') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.lock', id=2), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) #TODO Who is able to see locks on tasks? def test_lock_put_as_admin(client, random_test_data): """ An admin user should be able to get the tags from a task. """ body = { "lock": 'Testing', } headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.lock', id=1), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) #TODO Why cant the admin put a lock? How do you put a lock in the REST-API anyway? def test_lock_put_as_superuser(client, random_test_data): """ An admin user should be able to get the tags from a task. """ body = { "lock": 'Testing', } headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.lock', id=1), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_lock_put_as_user(client, random_test_data): """ A user should be able to get the tags from a task assigned to him/her. """ body = { "lock": 'Testing', } headers = basic_auth_authorization_header('user1', 'user1') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.lock', id=2), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) #TODO Who is able to put locks on tasks? def test_lock_delete_as_admin(client, random_test_data): """ An admin user should be able to get the tags from a task. """ headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.delete(url_for('api_v1.lock', id=1), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) #TODO Why cant the admin delete a lock? How do you put a lock in the REST-API anyway? def test_lock_delete_as_superuser(client, random_test_data): """ An admin user should be able to get the tags from a task. """ headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) response = client.delete(url_for('api_v1.lock', id=1), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_lock_delete_as_user(client, random_test_data): """ A user should be able to get the tags from a task assigned to him/her. """ headers = basic_auth_authorization_header('user1', 'user1') headers.update({'accept': 'application/json'}) response = client.delete(url_for('api_v1.lock', id=1), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) #TODO Who is able to delete locks on tasks? How do you delete locks anyway? taskmanager/api/tests/v1/test_api_role.py 0 → 100644 +105 −0 Original line number Diff line number Diff line # Copyright 2017 Biomedical Imaging Group Rotterdam, Departments of # Medical Informatics and Radiology, Erasmus MC, Rotterdam, The Netherlands # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import re from flask import url_for from flask_security.utils import hash_password from taskmanager.tests.helpers import basic_auth_authorization_header TIMESTAMP_REGEX = r'\d\d\d\d-\d\d-\d\dT\d\d:\d\d:\d\d' def test_role_get_as_admin(client, random_test_data): """ An admin user should be able to get the tags from a task. """ headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.roles', id=3), headers=headers) assert response.status_code == 200 # data = response.get_json() # assert 'tasks' in data # assert isinstance(data['tasks'], list) # assert 'status' in data['tasks'][4] # assert 'tags' in data['tasks'][4] # print(data['tasks'][4]) def test_role_get_as_superuser(client, random_test_data): """ An admin user should be able to get the tags from a task. """ headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.roles', id=2), headers=headers) assert response.status_code == 403 # data = response.get_json() # assert 'tasks' in data # assert isinstance(data['tasks'], list) # assert 'status' in data['tasks'][4] # assert 'tags' in data['tasks'][4] # print(data['tasks'][4]) def test_role_get_as_user(client, random_test_data): """ A user should be able to get the tags from a task assigned to him/her. """ headers = basic_auth_authorization_header('user1', 'user1') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.roles', id=1), headers=headers) assert response.status_code == 403 # data = response.get_json() # assert 'tasks' in data # assert isinstance(data['tasks'], list) # assert 'status' in data['tasks'][4] # assert 'tags' in data['tasks'][4] # print(data['tasks'][4]) def test_roles_get_all_as_admin(client, random_test_data): """ An admin user should be able to get an overview of all. """ headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) for x in range(1,4): response = client.get(url_for('api_v1.role', id=x), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_roles_get_all_as_superuser(client, random_test_data): """ A superuser should be able to get an overview of all. """ headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) for x in range(1,4): response = client.get(url_for('api_v1.role', id=x), headers=headers) assert response.status_code == 403 data = response.get_json() print(data) def test_roles_get_all_as_user(client, random_test_data): """ A user should be able to get an overview of all. """ #TODO Is that really what we wish? headers = basic_auth_authorization_header('user1', 'user1') headers.update({'accept': 'application/json'}) for x in range(1,4): response = client.get(url_for('api_v1.role', id=x), headers=headers) assert response.status_code == 403 data = response.get_json() print(data) taskmanager/api/tests/v1/test_api_tag.py +5 −2 Original line number Diff line number Diff line Loading @@ -154,6 +154,7 @@ def test_tag_post_new_tag_as_admin(client, random_test_data): assert response.status_code == 200 data = response.get_json() print(data) #TODO posting of tags seem to be possible, putting not ??? def test_tag_post_new_tag_as_superuser(client, random_test_data): Loading Loading @@ -297,20 +298,22 @@ def test_tag_put_nonexisting_tag_as_admin(client, app_config): } headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.tag', id=2), json=body, headers=headers) response = client.put(url_for('api_v1.tag', id=1), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) # Get and check if the data is correct in the response. response = client.get(url_for('api_v1.tag', id=2), headers=headers) response = client.get(url_for('api_v1.tag', id=1), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) assert data["name"] == body["name"] #TODO are there more datafields that can be set in group? # tag 2 seems to be not present. not found. # tag 1 (Testing) is not changed with this put! def test_tag_put_nonexisting_tag_as_superuser(client, app_config): Loading taskmanager/api/tests/v1/test_api_taskgroup.py 0 → 100644 +327 −0 Original line number Diff line number Diff line # Copyright 2017 Biomedical Imaging Group Rotterdam, Departments of # Medical Informatics and Radiology, Erasmus MC, Rotterdam, The Netherlands # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import re from flask import url_for from flask_security.utils import hash_password from taskmanager.tests.helpers import basic_auth_authorization_header TIMESTAMP_REGEX = r'\d\d\d\d-\d\d-\d\dT\d\d:\d\d:\d\d' def test_taskgroups_get_as_admin(client, random_test_data): """ An admin user should be able to get the taskgroups. """ headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.taskgroups', id=1), headers=headers) assert response.status_code == 200 #data = response.get_json() #assert 'tasks' in data #assert isinstance(data['tasks'], list) #assert 'status' in data['tasks'][4] #assert 'tags' in data['tasks'][4] #print(data['tasks'][4]) def test_taskgroup_get_as_superuser(client, random_test_data): """ An admin user should be able to get the tags from a task. """ headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.taskgroups', id=2), headers=headers) assert response.status_code == 200 def test_taskgroup_get_as_user(client, random_test_data): """ A user should be able to get the tags from a task assigned to him/her. """ headers = basic_auth_authorization_header('user1', 'user1') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.taskgroups', id=3), headers=headers) assert response.status_code == 200 def test_taskgroups_get_all_as_admin(client, random_test_data): """ An admin user should be able to get an overview of all. """ headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) for x in range(1,4): response = client.get(url_for('api_v1.taskgroup', id=x), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_taskgroups_get_all_as_superuser(client, random_test_data): """ A superuser should be able to get an overview of all. """ headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) for x in range(1,4): response = client.get(url_for('api_v1.taskgroup', id=x), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_taskgroups_get_all_as_user(client, random_test_data): """ A user should be able to get an overview of all. """ #TODO Is that really what we wish? headers = basic_auth_authorization_header('user1', 'user1') headers.update({'accept': 'application/json'}) for x in range(1,4): response = client.get(url_for('api_v1.taskgroup', id=x), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_taskgroup_post_new_taskgroup_as_admin(client, random_test_data): """ An admin should be able to post a new tag. """ body = { "label": "TaskGroup 77", # "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useful\"}", "tasks": [{ "name": "newtaskingroup", "status": "queued", }] } headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.post(url_for('api_v1.taskgroups'), json=body, headers=headers) assert response.status_code == 201 for x in range(1,5): response = client.get(url_for('api_v1.taskgroups', id=x), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_taskgroup_put_as_admin(client, random_test_data): """ Test if the admin can put a tag to a specific task. This should be allowed. """ body = { "label": "TaskGroup 111", "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useless\"}", # "tasks": [ # { # "name": "blabla", # "url": "/api/v1/tasks/1", # "id": "1", # "status": "queued", # "project": "test", # "template": "/api/v1/task_templates/22", # "tags": [] # } # ] } headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.taskgroup', id=1), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) # Check if the data is correct in the response. assert data["label"] == body["label"] #TODO are there more datafields that can be set in group? def test_taskgroup_put_as_superuser(client, random_test_data): """ Test if the admin can put a tag to a specific task. This should be allowed. """ body = { "label": "TaskGroup 111", "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useless\"}", # "tasks": [ # { # "name": "blabla", # "url": "/api/v1/tasks/1", # "id": "1", # "status": "queued", # "project": "test", # "template": "/api/v1/task_templates/22", # "tags": [] # } # ] } headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.taskgroup', id=1), json=body, headers=headers) assert response.status_code == 403 def test_taskgroup_put_as_user(client, random_test_data): """ Test if the admin can put a tag to a specific task. This should be allowed. """ body = { "label": "TaskGroup 111", "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useless\"}", # "tasks": [ # { # "name": "blabla", # "url": "/api/v1/tasks/1", # "id": "1", # "status": "queued", # "project": "test", # "template": "/api/v1/task_templates/22", # "tags": [] # } # ] } headers = basic_auth_authorization_header('user2', 'user2') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.taskgroup', id=1), json=body, headers=headers) assert response.status_code == 403 def test_taskgroup_post_as_admin(client, random_test_data): """ Test if the admin can put a tag to a specific task. This should be allowed. """ body = { "label": "TaskGroup 222", "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useless\"}", # "tasks": [ # { # "name": "blabla", # "url": "/api/v1/tasks/1", # "id": "1", # "status": "queued", # "project": "test", # "template": "/api/v1/task_templates/22", # "tags": [] # } # ] } headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.post(url_for('api_v1.taskgroups'), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) # Check if the data is correct in the response. assert data["label"] == body["label"] #TODO are there more datafields that can be set in group? def test_taskgroup_post_as_superuser(client, random_test_data): """ Test if the admin can put a tag to a specific task. This should be allowed. """ body = { "label": "TaskGroup 333", "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useless\"}", # "tasks": [ # { # "name": "blabla", # "url": "/api/v1/tasks/1", # "id": "1", # "status": "queued", # "project": "test", # "template": "/api/v1/task_templates/22", # "tags": [] # } # ] } headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) response = client.post(url_for('api_v1.taskgroups'), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) # Check if the data is correct in the response. assert data["label"] == body["label"] #TODO are there more datafields that can be set in group? def test_taskgroup_post_as_user(client, random_test_data): """ Test if the admin can put a tag to a specific task. This should be allowed. """ body = { "label": "TaskGroup 7", "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useless\"}", # "tasks": [ # { # "name": "blabla", # "url": "/api/v1/tasks/1", # "id": "1", # "status": "queued", # "project": "test", # "template": "/api/v1/task_templates/22", # "tags": [] # } # ] } headers = basic_auth_authorization_header('user2', 'user2') headers.update({'accept': 'application/json'}) response = client.post(url_for('api_v1.taskgroups'), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) # Check if the data is correct in the response. assert data["label"] == body["label"] #TODO are there more datafields that can be set in group? def test_taskgroup_put_nonexisting_as_admin(client, random_test_data): """ Test if the admin can put a tag to a specific task. This should be allowed. """ body = { "label": "TaskGroup 999", "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useless\"}", # "tasks": [ # { # "name": "blabla", # "url": "/api/v1/tasks/1", # "id": "1", # "status": "queued", # "project": "test", # "template": "/api/v1/task_templates/22", # "tags": [] # } # ] } headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.taskgroup', id=10000), json=body, headers=headers) assert response.status_code == 404 Loading
taskmanager/__main__.py +3 −0 Original line number Diff line number Diff line Loading @@ -109,7 +109,10 @@ def add_task(args=None): with open(args.task) as task_json: task_info['content'] = task_json.read() if isinstance(task_info['content'], list): task_info['template'] = json.loads(task_info['content'])[0]['template'] else: task_info['template'] = json.loads(task_info['content'])['template'] print("task_info: {}".format(task_info)) Loading
taskmanager/api/tests/v1/test_api_lock.py 0 → 100644 +136 −0 Original line number Diff line number Diff line # Copyright 2017 Biomedical Imaging Group Rotterdam, Departments of # Medical Informatics and Radiology, Erasmus MC, Rotterdam, The Netherlands # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import re from flask import url_for from flask_security.utils import hash_password from taskmanager.tests.helpers import basic_auth_authorization_header TIMESTAMP_REGEX = r'\d\d\d\d-\d\d-\d\dT\d\d:\d\d:\d\d' def test_lock_get_as_admin(client, random_test_data): """ An admin user should be able to get the tags from a task. """ headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.lock', id=1), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_lock_get_as_superuser(client, random_test_data): """ An admin user should be able to get the tags from a task. """ headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.lock', id=2), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_lock_get_as_user(client, random_test_data): """ A user should be able to get the tags from a task assigned to him/her. """ headers = basic_auth_authorization_header('user1', 'user1') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.lock', id=2), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) #TODO Who is able to see locks on tasks? def test_lock_put_as_admin(client, random_test_data): """ An admin user should be able to get the tags from a task. """ body = { "lock": 'Testing', } headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.lock', id=1), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) #TODO Why cant the admin put a lock? How do you put a lock in the REST-API anyway? def test_lock_put_as_superuser(client, random_test_data): """ An admin user should be able to get the tags from a task. """ body = { "lock": 'Testing', } headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.lock', id=1), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_lock_put_as_user(client, random_test_data): """ A user should be able to get the tags from a task assigned to him/her. """ body = { "lock": 'Testing', } headers = basic_auth_authorization_header('user1', 'user1') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.lock', id=2), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) #TODO Who is able to put locks on tasks? def test_lock_delete_as_admin(client, random_test_data): """ An admin user should be able to get the tags from a task. """ headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.delete(url_for('api_v1.lock', id=1), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) #TODO Why cant the admin delete a lock? How do you put a lock in the REST-API anyway? def test_lock_delete_as_superuser(client, random_test_data): """ An admin user should be able to get the tags from a task. """ headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) response = client.delete(url_for('api_v1.lock', id=1), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_lock_delete_as_user(client, random_test_data): """ A user should be able to get the tags from a task assigned to him/her. """ headers = basic_auth_authorization_header('user1', 'user1') headers.update({'accept': 'application/json'}) response = client.delete(url_for('api_v1.lock', id=1), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) #TODO Who is able to delete locks on tasks? How do you delete locks anyway?
taskmanager/api/tests/v1/test_api_role.py 0 → 100644 +105 −0 Original line number Diff line number Diff line # Copyright 2017 Biomedical Imaging Group Rotterdam, Departments of # Medical Informatics and Radiology, Erasmus MC, Rotterdam, The Netherlands # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import re from flask import url_for from flask_security.utils import hash_password from taskmanager.tests.helpers import basic_auth_authorization_header TIMESTAMP_REGEX = r'\d\d\d\d-\d\d-\d\dT\d\d:\d\d:\d\d' def test_role_get_as_admin(client, random_test_data): """ An admin user should be able to get the tags from a task. """ headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.roles', id=3), headers=headers) assert response.status_code == 200 # data = response.get_json() # assert 'tasks' in data # assert isinstance(data['tasks'], list) # assert 'status' in data['tasks'][4] # assert 'tags' in data['tasks'][4] # print(data['tasks'][4]) def test_role_get_as_superuser(client, random_test_data): """ An admin user should be able to get the tags from a task. """ headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.roles', id=2), headers=headers) assert response.status_code == 403 # data = response.get_json() # assert 'tasks' in data # assert isinstance(data['tasks'], list) # assert 'status' in data['tasks'][4] # assert 'tags' in data['tasks'][4] # print(data['tasks'][4]) def test_role_get_as_user(client, random_test_data): """ A user should be able to get the tags from a task assigned to him/her. """ headers = basic_auth_authorization_header('user1', 'user1') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.roles', id=1), headers=headers) assert response.status_code == 403 # data = response.get_json() # assert 'tasks' in data # assert isinstance(data['tasks'], list) # assert 'status' in data['tasks'][4] # assert 'tags' in data['tasks'][4] # print(data['tasks'][4]) def test_roles_get_all_as_admin(client, random_test_data): """ An admin user should be able to get an overview of all. """ headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) for x in range(1,4): response = client.get(url_for('api_v1.role', id=x), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_roles_get_all_as_superuser(client, random_test_data): """ A superuser should be able to get an overview of all. """ headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) for x in range(1,4): response = client.get(url_for('api_v1.role', id=x), headers=headers) assert response.status_code == 403 data = response.get_json() print(data) def test_roles_get_all_as_user(client, random_test_data): """ A user should be able to get an overview of all. """ #TODO Is that really what we wish? headers = basic_auth_authorization_header('user1', 'user1') headers.update({'accept': 'application/json'}) for x in range(1,4): response = client.get(url_for('api_v1.role', id=x), headers=headers) assert response.status_code == 403 data = response.get_json() print(data)
taskmanager/api/tests/v1/test_api_tag.py +5 −2 Original line number Diff line number Diff line Loading @@ -154,6 +154,7 @@ def test_tag_post_new_tag_as_admin(client, random_test_data): assert response.status_code == 200 data = response.get_json() print(data) #TODO posting of tags seem to be possible, putting not ??? def test_tag_post_new_tag_as_superuser(client, random_test_data): Loading Loading @@ -297,20 +298,22 @@ def test_tag_put_nonexisting_tag_as_admin(client, app_config): } headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.tag', id=2), json=body, headers=headers) response = client.put(url_for('api_v1.tag', id=1), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) # Get and check if the data is correct in the response. response = client.get(url_for('api_v1.tag', id=2), headers=headers) response = client.get(url_for('api_v1.tag', id=1), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) assert data["name"] == body["name"] #TODO are there more datafields that can be set in group? # tag 2 seems to be not present. not found. # tag 1 (Testing) is not changed with this put! def test_tag_put_nonexisting_tag_as_superuser(client, app_config): Loading
taskmanager/api/tests/v1/test_api_taskgroup.py 0 → 100644 +327 −0 Original line number Diff line number Diff line # Copyright 2017 Biomedical Imaging Group Rotterdam, Departments of # Medical Informatics and Radiology, Erasmus MC, Rotterdam, The Netherlands # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import re from flask import url_for from flask_security.utils import hash_password from taskmanager.tests.helpers import basic_auth_authorization_header TIMESTAMP_REGEX = r'\d\d\d\d-\d\d-\d\dT\d\d:\d\d:\d\d' def test_taskgroups_get_as_admin(client, random_test_data): """ An admin user should be able to get the taskgroups. """ headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.taskgroups', id=1), headers=headers) assert response.status_code == 200 #data = response.get_json() #assert 'tasks' in data #assert isinstance(data['tasks'], list) #assert 'status' in data['tasks'][4] #assert 'tags' in data['tasks'][4] #print(data['tasks'][4]) def test_taskgroup_get_as_superuser(client, random_test_data): """ An admin user should be able to get the tags from a task. """ headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.taskgroups', id=2), headers=headers) assert response.status_code == 200 def test_taskgroup_get_as_user(client, random_test_data): """ A user should be able to get the tags from a task assigned to him/her. """ headers = basic_auth_authorization_header('user1', 'user1') headers.update({'accept': 'application/json'}) response = client.get(url_for('api_v1.taskgroups', id=3), headers=headers) assert response.status_code == 200 def test_taskgroups_get_all_as_admin(client, random_test_data): """ An admin user should be able to get an overview of all. """ headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) for x in range(1,4): response = client.get(url_for('api_v1.taskgroup', id=x), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_taskgroups_get_all_as_superuser(client, random_test_data): """ A superuser should be able to get an overview of all. """ headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) for x in range(1,4): response = client.get(url_for('api_v1.taskgroup', id=x), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_taskgroups_get_all_as_user(client, random_test_data): """ A user should be able to get an overview of all. """ #TODO Is that really what we wish? headers = basic_auth_authorization_header('user1', 'user1') headers.update({'accept': 'application/json'}) for x in range(1,4): response = client.get(url_for('api_v1.taskgroup', id=x), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_taskgroup_post_new_taskgroup_as_admin(client, random_test_data): """ An admin should be able to post a new tag. """ body = { "label": "TaskGroup 77", # "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useful\"}", "tasks": [{ "name": "newtaskingroup", "status": "queued", }] } headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.post(url_for('api_v1.taskgroups'), json=body, headers=headers) assert response.status_code == 201 for x in range(1,5): response = client.get(url_for('api_v1.taskgroups', id=x), headers=headers) assert response.status_code == 200 data = response.get_json() print(data) def test_taskgroup_put_as_admin(client, random_test_data): """ Test if the admin can put a tag to a specific task. This should be allowed. """ body = { "label": "TaskGroup 111", "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useless\"}", # "tasks": [ # { # "name": "blabla", # "url": "/api/v1/tasks/1", # "id": "1", # "status": "queued", # "project": "test", # "template": "/api/v1/task_templates/22", # "tags": [] # } # ] } headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.taskgroup', id=1), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) # Check if the data is correct in the response. assert data["label"] == body["label"] #TODO are there more datafields that can be set in group? def test_taskgroup_put_as_superuser(client, random_test_data): """ Test if the admin can put a tag to a specific task. This should be allowed. """ body = { "label": "TaskGroup 111", "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useless\"}", # "tasks": [ # { # "name": "blabla", # "url": "/api/v1/tasks/1", # "id": "1", # "status": "queued", # "project": "test", # "template": "/api/v1/task_templates/22", # "tags": [] # } # ] } headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.taskgroup', id=1), json=body, headers=headers) assert response.status_code == 403 def test_taskgroup_put_as_user(client, random_test_data): """ Test if the admin can put a tag to a specific task. This should be allowed. """ body = { "label": "TaskGroup 111", "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useless\"}", # "tasks": [ # { # "name": "blabla", # "url": "/api/v1/tasks/1", # "id": "1", # "status": "queued", # "project": "test", # "template": "/api/v1/task_templates/22", # "tags": [] # } # ] } headers = basic_auth_authorization_header('user2', 'user2') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.taskgroup', id=1), json=body, headers=headers) assert response.status_code == 403 def test_taskgroup_post_as_admin(client, random_test_data): """ Test if the admin can put a tag to a specific task. This should be allowed. """ body = { "label": "TaskGroup 222", "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useless\"}", # "tasks": [ # { # "name": "blabla", # "url": "/api/v1/tasks/1", # "id": "1", # "status": "queued", # "project": "test", # "template": "/api/v1/task_templates/22", # "tags": [] # } # ] } headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.post(url_for('api_v1.taskgroups'), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) # Check if the data is correct in the response. assert data["label"] == body["label"] #TODO are there more datafields that can be set in group? def test_taskgroup_post_as_superuser(client, random_test_data): """ Test if the admin can put a tag to a specific task. This should be allowed. """ body = { "label": "TaskGroup 333", "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useless\"}", # "tasks": [ # { # "name": "blabla", # "url": "/api/v1/tasks/1", # "id": "1", # "status": "queued", # "project": "test", # "template": "/api/v1/task_templates/22", # "tags": [] # } # ] } headers = basic_auth_authorization_header('superuser', 'superuser') headers.update({'accept': 'application/json'}) response = client.post(url_for('api_v1.taskgroups'), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) # Check if the data is correct in the response. assert data["label"] == body["label"] #TODO are there more datafields that can be set in group? def test_taskgroup_post_as_user(client, random_test_data): """ Test if the admin can put a tag to a specific task. This should be allowed. """ body = { "label": "TaskGroup 7", "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useless\"}", # "tasks": [ # { # "name": "blabla", # "url": "/api/v1/tasks/1", # "id": "1", # "status": "queued", # "project": "test", # "template": "/api/v1/task_templates/22", # "tags": [] # } # ] } headers = basic_auth_authorization_header('user2', 'user2') headers.update({'accept': 'application/json'}) response = client.post(url_for('api_v1.taskgroups'), json=body, headers=headers) assert response.status_code == 200 data = response.get_json() print(data) # Check if the data is correct in the response. assert data["label"] == body["label"] #TODO are there more datafields that can be set in group? def test_taskgroup_put_nonexisting_as_admin(client, random_test_data): """ Test if the admin can put a tag to a specific task. This should be allowed. """ body = { "label": "TaskGroup 999", "status": "queued", # "callback_url": "http://localhost:5000/experiment/1", # "callback_content": "{\"function\": \"do_something_useless\"}", # "tasks": [ # { # "name": "blabla", # "url": "/api/v1/tasks/1", # "id": "1", # "status": "queued", # "project": "test", # "template": "/api/v1/task_templates/22", # "tags": [] # } # ] } headers = basic_auth_authorization_header('admin', 'admin') headers.update({'accept': 'application/json'}) response = client.put(url_for('api_v1.taskgroup', id=10000), json=body, headers=headers) assert response.status_code == 404
