fixes the locks test (d4c0ab73) · Commits · Radiology Erasmus MC / Infrastructure / Task Manager

taskmanager/api/tests/v1/test_api_lock.py

+29 −25

Original line number	Diff line number	Diff line
		@@ -23,8 +23,8 @@ 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. """
		def test_lock_get_as_admin(client, test_locks):
		""" An admin user should be able to get the lock on 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)
		@@ -34,8 +34,8 @@ def test_lock_get_as_admin(client, random_test_data):
		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. """
		def test_lock_get_as_superuser(client, test_locks):
		""" A superuser should be able to get the locks on 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)
		@@ -45,11 +45,11 @@ def test_lock_get_as_superuser(client, random_test_data):
		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. """
		def test_lock_get_as_user(client, test_locks):
		""" A user should be able to get the lock on its own task. """
		headers = basic_auth_authorization_header('user1', 'user1')
		headers.update({'accept': 'application/json'})
		response = client.get(url_for('api_v1.lock', id=2), headers=headers)
		response = client.get(url_for('api_v1.lock', id=3), headers=headers)
		assert response.status_code == 200

		data = response.get_json()
		@@ -57,10 +57,10 @@ def test_lock_get_as_user(client, random_test_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. """
		def test_lock_put_as_admin(client, test_locks):
		""" An admin should be able to update the lock on a task. """
		body = {
		"lock": 'Testing',
		"lock": 'user1',
		}
		headers = basic_auth_authorization_header('admin', 'admin')
		headers.update({'accept': 'application/json'})
		@@ -72,28 +72,29 @@ def test_lock_put_as_admin(client, random_test_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):
		def test_lock_put_as_superuser(client, test_locks):
		""" An admin user should be able to get the tags from a task. """
		body = {
		"lock": 'Testing',
		"lock": 'user1',
		}
		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)
		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)


		def test_lock_put_as_user(client, random_test_data):
		def test_lock_put_as_user(client, test_locks):
		""" A user should be able to get the tags from a task assigned to him/her. """
		user = 'user1'
		body = {
		"lock": 'Testing',
		"lock": user,
		}
		headers = basic_auth_authorization_header('user1', 'user1')
		headers = basic_auth_authorization_header(user, user)
		headers.update({'accept': 'application/json'})
		response = client.put(url_for('api_v1.lock', id=2), json=body, headers=headers)
		response = client.put(url_for('api_v1.lock', id=4), json=body, headers=headers)
		assert response.status_code == 200

		data = response.get_json()
		@@ -101,36 +102,39 @@ def test_lock_put_as_user(client, random_test_data):
		#TODO Who is able to put locks on tasks?


		def test_lock_delete_as_admin(client, random_test_data):
		def test_lock_delete_as_admin(client, test_locks):
		""" 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

		# Assure the lock is released.
		data = response.get_json()
		print(data)
		data['lock'] is None
		#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):
		def test_lock_delete_as_superuser(client, test_locks):
		""" 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)
		response = client.delete(url_for('api_v1.lock', id=2), headers=headers)
		assert response.status_code == 200

		# Assure the lock is released.
		data = response.get_json()
		print(data)
		data['lock'] is None


		def test_lock_delete_as_user(client, random_test_data):
		def test_lock_delete_as_user(client, test_locks):
		""" 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)
		response = client.delete(url_for('api_v1.lock', id=3), headers=headers)
		assert response.status_code == 200

		# Assure the lock is released.
		data = response.get_json()
		print(data)
		data['lock'] is None
		#TODO Who is able to delete locks on tasks? How do you delete locks anyway?

taskmanager/api/v1.py

+6 −6

Original line number	Diff line number	Diff line
		@@ -718,14 +718,14 @@ class LockAPI(Resource):
		return result

		@http_auth_required
		@permissions_accepted('task_read', 'task_read_all')
		@permissions_accepted('task_read', 'task_update_lock_all', 'task_update_all')
		@api.marshal_with(lock_get)
		@api.response(200, 'Success')
		@api.response(403, 'You are not authorized to release the lock of another user')
		@api.response(404, 'Could not find selected task or user')
		def delete(self, id):
		query = models.Task.query.filter(models.Task.id == id)
		if not has_permission_any('task_update_lock_all'):
		if not has_permission_any('task_update_lock_all', 'task_update_all'):
		tasks_via_user_query = query.filter(models.Task.users.contains(current_user))
		tasks_via_group_query = query.filter(models.Task.users_via_group.contains(current_user))
		query = tasks_via_user_query.union(tasks_via_group_query)
		@@ -737,7 +737,7 @@ class LockAPI(Resource):
		"lock": None}
		return result, 404

		if task.lock is not None and task.lock != current_user and not has_permission_all('task_update_lock_all'):
		if task.lock is not None and task.lock != current_user and not has_permission_any('task_update_lock_all', 'task_update_all'):
		# If the task is locked and not by you and you are not authorized to update all locks, throw 403.
		result = {"error": 'You are not authorized to release the lock of a task, assigned to someone else!',
		"username": None, "lock": None}
		@@ -759,7 +759,7 @@ class LockAPI(Resource):
		@api.response(409, 'Task already locked by other user')
		def put(self, id):
		query = models.Task.query.filter(models.Task.id == id)
		if not has_permission_any('task_update_lock_all'):
		if not has_permission_any('task_update_lock_all', 'task_update_all'):
		tasks_via_user_query = query.filter(models.Task.users.contains(current_user))
		tasks_via_group_query = query.filter(models.Task.users_via_group.contains(current_user))
		query = tasks_via_user_query.union(tasks_via_group_query)
		@@ -787,7 +787,7 @@ class LockAPI(Resource):
		return result, 404

		if task.lock is not None and task.lock != user:
		if has_permission_any('task_update_lock_all'):
		if has_permission_any('task_update_lock_all', 'task_update_all'):
		# This user has the right permission to overwrite any lock, so proceed.
		task.lock = user
		else:
		@@ -797,7 +797,7 @@ class LockAPI(Resource):
		"error": 'Task already locked!'}
		return result, 409

		if not has_permission_any('task_update_lock_all'):
		if not has_permission_any('task_update_lock_all', 'task_update_all'):
		# You can only lock tasks to your own account if are not authorized to update all locks.
		if user == current_user:
		# This is allowed! A user can lock its own to task itself.

taskmanager/conftest.py

+10 −0

Original line number	Diff line number	Diff line
		@@ -22,6 +22,7 @@ from . import models
		from . import create_app
		from .models import db

		from .tests.loaders import create_lock
		from .tests.loaders import create_random_test_tasks
		from .tests.loaders import create_random_test_taskgroup

		@@ -77,6 +78,15 @@ def random_test_data(app, init_db, app_config):
		yield app


		@pytest.fixture(scope="session")
		def test_locks(app, init_db, app_config, random_test_data):
		create_lock(app, 1, 1)
		create_lock(app, 2, 2)
		create_lock(app, 3, 3)

		yield app


		@pytest.fixture
		def client(app):
		"""A test client for the app."""

taskmanager/tests/loaders.py

+9 −0

Original line number	Diff line number	Diff line
		@@ -6,6 +6,15 @@ from .. import models
		from ..control import insert_task


		def create_lock(app, task_id, user_id):
		db = models.db
		with app.app_context():
		task = models.Task.query.get(task_id)
		user = models.User.query.get(user_id)
		task.lock = user
		db.session.commit()


		def create_random_test_taskgroup(app, num_tasks=3, silent=False):
		db = models.db
		with app.app_context():