Allow deletion of tasks by non-admin users.

attract/tasks/routes.py

@@ -35,10 +35,8 @@ def index():
 
 
 @blueprint.route('/<task_id>', methods=['DELETE'])
+@flask_login.login_required
 def delete(task_id):
-    if not current_attract.auth.current_user_may(current_attract.auth.Actions.USE):
-        raise wz_exceptions.Forbidden()
-
     log.info('Deleting task %s', task_id)
 
     etag = request.form['etag']

tests/test_tasks.py

@@ -158,3 +158,61 @@ class TaskWorkflowTest(AbstractAttractTest):
 
         # Test with Eve
         self.get(node_url, auth_token='token', expected_status=404)
+
+    @responses.activate
+    def test_delete_task_nonadmin(self):
+        from pillar.api.projects.utils import get_admin_group_id
+        from attract.tasks import routes
+
+        self.enter_app_context()
+
+        task = self.create_task()
+        task_id = task['_id']
+
+        # Create a project member who is not admin.
+        admin_gid = get_admin_group_id(self.proj_id)
+        self.create_user(6 * 'dafe',
+                         roles=('subscriber', 'attract-user'),
+                         groups=[admin_gid],
+                         token='mortal-token')
+
+        task = self.get(f'/api/nodes/{task_id}', auth_token='mortal-token').get_json()
+
+        with self.app.test_request_context(method='DELETE', data={'etag': task['_etag']}):
+            pillar.auth.login_user('mortal-token', load_from_db=True)
+            resp, status_code = routes.delete(str(task_id))
+            self.assertEqual(status_code, 204)
+            self.assertEqual(resp, '')
+
+        # Test directly with MongoDB
+        nodes_coll = self.app.data.driver.db['nodes']
+        found = nodes_coll.find_one(ObjectId(task_id))
+        self.assertTrue(found['_deleted'])
+
+        # Test with Eve
+        self.get(f'/api/nodes/{task_id}', auth_token='mortal-token', expected_status=404)
+
+    @responses.activate
+    def test_delete_task_nonmember(self):
+        from attract.tasks import routes
+
+        self.enter_app_context()
+
+        task = self.create_task()
+        task_id = task['_id']
+
+        # Create a user who is not admin and not a project member
+        self.create_user(6 * 'dafe',
+                         roles=('subscriber', 'attract-user'),
+                         groups=[],
+                         token='mortal-token')
+
+        with self.app.test_request_context(method='DELETE', data={'etag': task['_etag']}):
+            pillar.auth.login_user('mortal-token', load_from_db=True)
+            with self.assertRaises(sdk_exceptions.ForbiddenAccess):
+                routes.delete(str(task_id))
+
+        # Test directly with MongoDB
+        nodes_coll = self.app.data.driver.db['nodes']
+        found = nodes_coll.find_one(ObjectId(task_id))
+        self.assertFalse(found.get('_deleted', False))