Merge pull request #10659 from archesproject/raise-exception-arg

Add raise_exception arg to @group_required #10658

arches/app/utils/decorators.py

@@ -59,7 +59,7 @@ def new_func(*args, **kwargs):
     return new_func
 
 
-def group_required(*group_names):
+def group_required(*group_names, raise_exception=False):
     """
     Requires user membership in at least one of the groups passed in.
 
@@ -69,6 +69,9 @@ def in_groups(u):
         if u.is_authenticated:
             if u.is_superuser or bool(u.groups.filter(name__in=group_names)):
                 return True
+        if raise_exception:
+            raise PermissionDenied
+        # else: user_passes_test() redirects to nowhere
         return False
 
     return user_passes_test(in_groups)

arches/app/views/workflow_history.py

@@ -1,22 +1,23 @@
 import json
 
+from django.core.exceptions import PermissionDenied
 from django.db import transaction
 from django.db.models import F, JSONField, Value
 from django.db.models.expressions import CombinedExpression
 from django.db.utils import IntegrityError
+from django.utils.decorators import method_decorator
 from django.utils.translation import gettext as _
 from django.views.generic import View
 
-from arches.app.utils.permission_backend import user_is_resource_editor
+from arches.app.utils.decorators import group_required
 from arches.app.utils.response import JSONErrorResponse, JSONResponse
 from arches.app.models import models
 
 
+@method_decorator(group_required("Resource Editor", raise_exception=True), name="dispatch")
 class WorkflowHistoryView(View):
 
     def get(self, request, workflowid):
-        if not user_is_resource_editor(request.user):
-            return JSONErrorResponse(_("Request Failed"), _("Permission Denied"), status=403)
         try:
             if request.user.is_superuser:
                 workflow_history = models.WorkflowHistory.objects.get(workflowid=workflowid)
@@ -28,9 +29,6 @@ def get(self, request, workflowid):
         return JSONResponse(workflow_history, status=200)
 
     def post(self, request, workflowid):
-        if not user_is_resource_editor(request.user):
-            return JSONErrorResponse(_("Request Failed"), _("Permission Denied"), status=403)
-
         data = json.loads(request.body)
         stepdata = data.get("stepdata", {})
         componentdata = data.get("componentdata", {})
@@ -60,8 +58,8 @@ def post(self, request, workflowid):
                         workflowid = workflowid,
                     )
                 else:
-                    return JSONErrorResponse(_("Request Failed"), _("Permission Denied"), status=403)
-            
+                    raise PermissionDenied
+
             if not created:
                 if history.completed:
                     return JSONErrorResponse(

tests/views/workflow_tests.py

@@ -1,20 +1,26 @@
 import uuid
 import datetime
 
-from django.contrib.auth.models import User
+from django.contrib.auth.models import Group, User
 from django.urls import reverse
 from django.test.client import Client
 
 from arches.app.models.models import WorkflowHistory
 from tests.base_test import ArchesTestCase
 
+# these tests can be run from the command line via
+# python manage.py test tests.views.workflow_tests --settings="tests.test_settings"
+
 
 class WorkflowHistoryTests(ArchesTestCase):
     @classmethod
     def setUpClass(cls):
         cls.client = Client()
         cls.admin = User.objects.get(username="admin")
         cls.anonymous = User.objects.get(username="anonymous")
+        cls.editor = User.objects.create_user(username="editor", email="[email protected]", password="Test12345!")
+        group = Group.objects.get(name="Resource Editor")
+        group.user_set.add(cls.editor)
         super().setUpClass()
 
     def setUp(self):
@@ -73,7 +79,7 @@ def test_get_workflow_history(self):
         response = self.client.get(reverse("workflow_history", kwargs={"workflowid": str(self.history.workflowid)}))
 
         self.assertEqual(response.status_code, 403)
-        self.assertIn(b"Permission Denied", response.content)
+        self.assertIn(b"Forbidden", response.content)
 
         self.client.force_login(self.admin)
         response = self.client.get(reverse("workflow_history", kwargs={"workflowid": str(self.history.workflowid)}))
@@ -83,13 +89,6 @@ def test_get_workflow_history(self):
 
     def test_post_workflow_history(self):
         """Partial updates of componentdata and stepdata are allowed."""
-        self.client.force_login(self.anonymous)
-        response = self.client.post(reverse("workflow_history", kwargs={"workflowid": str(self.history.workflowid)}))
-
-        self.assertEqual(response.status_code, 403)
-        self.assertIn(b"Permission Denied", response.content)
-
-        self.client.force_login(self.admin)
         post_data = {
             "workflowid": str(self.history.workflowid),  # required
             "workflowname": 'test-name',
@@ -128,6 +127,18 @@ def test_post_workflow_history(self):
             },
         }
 
+        # Non-superuser cannot update someone else's workflow.
+        self.client.force_login(self.editor)
+        response = self.client.post(
+            reverse("workflow_history", kwargs={"workflowid": str(self.history.workflowid)}),
+            post_data,
+            content_type="application/json",
+        )
+
+        self.assertEqual(response.status_code, 403)
+        self.assertIn(b"Forbidden", response.content)
+
+        self.client.force_login(self.admin)
         response = self.client.post(
             reverse("workflow_history", kwargs={"workflowid": str(self.history.workflowid)}),
             post_data,