Use @group_required in workflow history view re #10658

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,7 +1,7 @@
 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
 
@@ -15,6 +15,9 @@ 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="sam", 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 +76,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 +86,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 +124,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,