Register workflow and task permissions, add tests for workflow admin views with and without permissions, and ensure the workflow menu item is only shown if the workflow index is accessible (#6159)

wagtail/admin/tests/test_workflows.py

@@ -4,7 +4,7 @@ from unittest import mock
 
 from django.conf import settings
 from django.contrib.auth import get_user_model
-from django.contrib.auth.models import Group
+from django.contrib.auth.models import Group, Permission
 from django.core import mail
 from django.core.mail import EmailMultiAlternatives
 from django.test import TestCase, override_settings
@@ -32,6 +32,23 @@ class TestWorkflowsIndexView(TestCase, WagtailTestUtils):
         delete_existing_workflows()
         self.login()
 
+        self.editor = get_user_model().objects.create_user(
+            username='editor',
+            email='editor@email.com',
+            password='password',
+        )
+        editors = Group.objects.get(name='Editors')
+        editors.user_set.add(self.editor)
+
+        self.moderator = get_user_model().objects.create_user(
+            username='moderator',
+            email='moderator@email.com',
+            password='password',
+        )
+        moderators = Group.objects.get(name='Moderators')
+        moderators.user_set.add(self.moderator)
+        moderators.permissions.add(Permission.objects.get(codename="add_workflow"))
+
     def get(self, params={}):
         return self.client.get(reverse('wagtailadmin_workflows:index'), params)
 
@@ -67,6 +84,17 @@ class TestWorkflowsIndexView(TestCase, WagtailTestUtils):
         self.assertEqual(response.status_code, 200)
         self.assertContains(response, "There are no enabled workflows.")
 
+    def test_permissions(self):
+        self.login(user=self.editor)
+        response = self.get()
+        self.assertEqual(response.status_code, 302)
+        full_context = {key: value for context in response.context for key, value in context.items()}
+        self.assertEqual(full_context['message'], 'Sorry, you do not have permission to access this area.')
+
+        self.login(user=self.moderator)
+        response = self.get()
+        self.assertEqual(response.status_code, 200)
+
 
 class TestWorkflowsCreateView(TestCase, WagtailTestUtils):
 
@@ -76,6 +104,23 @@ class TestWorkflowsCreateView(TestCase, WagtailTestUtils):
         self.task_1 = SimpleTask.objects.create(name="first_task")
         self.task_2 = SimpleTask.objects.create(name="second_task")
 
+        self.editor = get_user_model().objects.create_user(
+            username='editor',
+            email='editor@email.com',
+            password='password',
+        )
+        editors = Group.objects.get(name='Editors')
+        editors.user_set.add(self.editor)
+
+        self.moderator = get_user_model().objects.create_user(
+            username='moderator',
+            email='moderator@email.com',
+            password='password',
+        )
+        moderators = Group.objects.get(name='Moderators')
+        moderators.user_set.add(self.moderator)
+        moderators.permissions.add(Permission.objects.get(codename="add_workflow"))
+
 
     def get(self, params={}):
         return self.client.get(reverse('wagtailadmin_workflows:add'), params)
@@ -113,6 +158,17 @@ class TestWorkflowsCreateView(TestCase, WagtailTestUtils):
         self.assertEqual(WorkflowTask.objects.get(workflow=workflow, task=self.task_1.task_ptr).sort_order, 0)
         self.assertEqual(WorkflowTask.objects.get(workflow=workflow, task=self.task_2.task_ptr).sort_order, 1)
 
+    def test_permissions(self):
+        self.login(user=self.editor)
+        response = self.get()
+        self.assertEqual(response.status_code, 302)
+        full_context = {key: value for context in response.context for key, value in context.items()}
+        self.assertEqual(full_context['message'], 'Sorry, you do not have permission to access this area.')
+
+        self.login(user=self.moderator)
+        response = self.get()
+        self.assertEqual(response.status_code, 200)
+
 
 class TestWorkflowsEditView(TestCase, WagtailTestUtils):
 
@@ -127,6 +183,23 @@ class TestWorkflowsEditView(TestCase, WagtailTestUtils):
         self.page = Page.objects.first()
         WorkflowPage.objects.create(workflow=self.workflow, page=self.page)
 
+        self.editor = get_user_model().objects.create_user(
+            username='editor',
+            email='editor@email.com',
+            password='password',
+        )
+        editors = Group.objects.get(name='Editors')
+        editors.user_set.add(self.editor)
+
+        self.moderator = get_user_model().objects.create_user(
+            username='moderator',
+            email='moderator@email.com',
+            password='password',
+        )
+        moderators = Group.objects.get(name='Moderators')
+        moderators.user_set.add(self.moderator)
+        moderators.permissions.add(Permission.objects.get(codename="change_workflow"))
+
 
     def get(self, params={}):
         return self.client.get(reverse('wagtailadmin_workflows:edit', args=[self.workflow.id]), params)
@@ -181,6 +254,17 @@ class TestWorkflowsEditView(TestCase, WagtailTestUtils):
         self.assertEqual(WorkflowTask.objects.get(workflow=workflow, task=self.task_1.task_ptr).sort_order, 0)
         self.assertEqual(WorkflowTask.objects.get(workflow=workflow, task=self.task_2.task_ptr).sort_order, 1)
 
+    def test_permissions(self):
+        self.login(user=self.editor)
+        response = self.get()
+        self.assertEqual(response.status_code, 302)
+        full_context = {key: value for context in response.context for key, value in context.items()}
+        self.assertEqual(full_context['message'], 'Sorry, you do not have permission to access this area.')
+
+        self.login(user=self.moderator)
+        response = self.get()
+        self.assertEqual(response.status_code, 200)
+
 
 class TestAddWorkflowToPage(TestCase, WagtailTestUtils):
     fixtures = ['test.json']
@@ -194,6 +278,23 @@ class TestAddWorkflowToPage(TestCase, WagtailTestUtils):
         self.other_page = Page.objects.last()
         WorkflowPage.objects.create(workflow=self.other_workflow, page=self.other_page)
 
+        self.editor = get_user_model().objects.create_user(
+            username='editor',
+            email='editor@email.com',
+            password='password',
+        )
+        editors = Group.objects.get(name='Editors')
+        editors.user_set.add(self.editor)
+
+        self.moderator = get_user_model().objects.create_user(
+            username='moderator',
+            email='moderator@email.com',
+            password='password',
+        )
+        moderators = Group.objects.get(name='Moderators')
+        moderators.user_set.add(self.moderator)
+        moderators.permissions.add(Permission.objects.get(codename="change_workflow"))
+
     def get(self, params={}):
         return self.client.get(reverse('wagtailadmin_workflows:add_to_page', args=[self.workflow.id]), params)
 
@@ -218,6 +319,15 @@ class TestAddWorkflowToPage(TestCase, WagtailTestUtils):
         self.post({'page': str(self.other_page.id), 'overwrite_existing': 'True', 'workflow': str(self.workflow.id)})
         self.assertEqual(WorkflowPage.objects.filter(workflow=self.workflow, page=self.other_page).count(), 1)
 
+    def test_permissions(self):
+        self.login(user=self.editor)
+        response = self.get()
+        self.assertEqual(response.status_code, 403)
+
+        self.login(user=self.moderator)
+        response = self.get()
+        self.assertEqual(response.status_code, 200)
+
 
 class TestRemoveWorkflow(TestCase, WagtailTestUtils):
     fixtures = ['test.json']
@@ -229,6 +339,23 @@ class TestRemoveWorkflow(TestCase, WagtailTestUtils):
         self.page = Page.objects.first()
         WorkflowPage.objects.create(workflow=self.workflow, page=self.page)
 
+        self.editor = get_user_model().objects.create_user(
+            username='editor',
+            email='editor@email.com',
+            password='password',
+        )
+        editors = Group.objects.get(name='Editors')
+        editors.user_set.add(self.editor)
+
+        self.moderator = get_user_model().objects.create_user(
+            username='moderator',
+            email='moderator@email.com',
+            password='password',
+        )
+        moderators = Group.objects.get(name='Moderators')
+        moderators.user_set.add(self.moderator)
+        moderators.permissions.add(Permission.objects.get(codename="change_workflow"))
+
     def post(self, post_data={}):
         return self.client.post(reverse('wagtailadmin_workflows:remove', args=[self.page.id, self.workflow.id]), post_data)
 
@@ -237,6 +364,16 @@ class TestRemoveWorkflow(TestCase, WagtailTestUtils):
         self.post()
         self.assertEqual(WorkflowPage.objects.filter(workflow=self.workflow, page=self.page).count(), 0)
 
+    def test_no_permissions(self):
+        self.login(user=self.editor)
+        response = self.post()
+        self.assertEqual(response.status_code, 403)
+
+    def test_post_with_permission(self):
+        self.login(user=self.moderator)
+        response = self.post()
+        self.assertEqual(response.status_code, 302)
+
 
 class TestTaskIndexView(TestCase, WagtailTestUtils):
 
@@ -244,6 +381,23 @@ class TestTaskIndexView(TestCase, WagtailTestUtils):
         delete_existing_workflows()
         self.login()
 
+        self.editor = get_user_model().objects.create_user(
+            username='editor',
+            email='editor@email.com',
+            password='password',
+        )
+        editors = Group.objects.get(name='Editors')
+        editors.user_set.add(self.editor)
+
+        self.moderator = get_user_model().objects.create_user(
+            username='moderator',
+            email='moderator@email.com',
+            password='password',
+        )
+        moderators = Group.objects.get(name='Moderators')
+        moderators.user_set.add(self.moderator)
+        moderators.permissions.add(Permission.objects.get(codename="change_task"))
+
     def get(self, params={}):
         return self.client.get(reverse('wagtailadmin_workflows:task_index'), params)
 
@@ -280,6 +434,17 @@ class TestTaskIndexView(TestCase, WagtailTestUtils):
         self.assertContains(response, "There are no enabled tasks")
         self.assertNotContains(response, "test_task")
 
+    def test_permissions(self):
+        self.login(user=self.editor)
+        response = self.get()
+        self.assertEqual(response.status_code, 302)
+        full_context = {key: value for context in response.context for key, value in context.items()}
+        self.assertEqual(full_context['message'], 'Sorry, you do not have permission to access this area.')
+
+        self.login(user=self.moderator)
+        response = self.get()
+        self.assertEqual(response.status_code, 200)
+
 
 class TestCreateTaskView(TestCase, WagtailTestUtils):
 
@@ -287,6 +452,23 @@ class TestCreateTaskView(TestCase, WagtailTestUtils):
         delete_existing_workflows()
         self.login()
 
+        self.editor = get_user_model().objects.create_user(
+            username='editor',
+            email='editor@email.com',
+            password='password',
+        )
+        editors = Group.objects.get(name='Editors')
+        editors.user_set.add(self.editor)
+
+        self.moderator = get_user_model().objects.create_user(
+            username='moderator',
+            email='moderator@email.com',
+            password='password',
+        )
+        moderators = Group.objects.get(name='Moderators')
+        moderators.user_set.add(self.moderator)
+        moderators.permissions.add(Permission.objects.get(codename="add_task"))
+
     def get(self, params={}):
         return self.client.get(reverse('wagtailadmin_workflows:add_task', kwargs={'app_label': SimpleTask._meta.app_label, 'model_name': SimpleTask._meta.model_name}), params)
 
@@ -308,6 +490,18 @@ class TestCreateTaskView(TestCase, WagtailTestUtils):
         tasks = Task.objects.filter(name="test_task", active=True)
         self.assertEqual(tasks.count(), 1)
 
+    def test_permissions(self):
+        self.login(user=self.editor)
+        response = self.get()
+        self.assertEqual(response.status_code, 302)
+        full_context = {key: value for context in response.context for key, value in context.items()}
+        self.assertEqual(full_context['message'], 'Sorry, you do not have permission to access this area.')
+
+        self.login(user=self.moderator)
+        response = self.get()
+        self.assertEqual(response.status_code, 200)
+
+
 
 class TestSelectTaskTypeView(TestCase, WagtailTestUtils):
 
@@ -335,6 +529,23 @@ class TestEditTaskView(TestCase, WagtailTestUtils):
         self.login()
         self.task = SimpleTask.objects.create(name="test_task")
 
+        self.editor = get_user_model().objects.create_user(
+            username='editor',
+            email='editor@email.com',
+            password='password',
+        )
+        editors = Group.objects.get(name='Editors')
+        editors.user_set.add(self.editor)
+
+        self.moderator = get_user_model().objects.create_user(
+            username='moderator',
+            email='moderator@email.com',
+            password='password',
+        )
+        moderators = Group.objects.get(name='Moderators')
+        moderators.user_set.add(self.moderator)
+        moderators.permissions.add(Permission.objects.get(codename="change_task"))
+
     def get(self, params={}):
         return self.client.get(reverse('wagtailadmin_workflows:edit_task', args=[self.task.id]), params)
 
@@ -356,6 +567,17 @@ class TestEditTaskView(TestCase, WagtailTestUtils):
         task = Task.objects.get(id=self.task.id)
         self.assertEqual(task.name, "test_task_modified")
 
+    def test_permissions(self):
+        self.login(user=self.editor)
+        response = self.get()
+        self.assertEqual(response.status_code, 302)
+        full_context = {key: value for context in response.context for key, value in context.items()}
+        self.assertEqual(full_context['message'], 'Sorry, you do not have permission to access this area.')
+
+        self.login(user=self.moderator)
+        response = self.get()
+        self.assertEqual(response.status_code, 200)
+
 
 class TestSubmitToWorkflow(TestCase, WagtailTestUtils):
     def setUp(self):

wagtail/admin/views/workflows.py

@@ -193,7 +193,7 @@ def remove_workflow(request, page_pk, workflow_pk=None):
     page = get_object_or_404(Page, id=page_pk)
 
     # Check permissions
-    if not workflow_permission_policy.user_has_permission(request.user, 'remove_from_page'):
+    if not workflow_permission_policy.user_has_permission(request.user, 'change'):
         raise PermissionDenied
 
     if hasattr(page, 'workflowpage'):
@@ -214,7 +214,7 @@ def remove_workflow(request, page_pk, workflow_pk=None):
 def add_to_page(request, workflow_pk):
     # Assign a workflow to a Page, including a confirmation step if the Page has a different Workflow assigned already.
 
-    if not workflow_permission_policy.user_has_permission(request.user, 'add_to_page'):
+    if not workflow_permission_policy.user_has_permission(request.user, 'change'):
         raise PermissionDenied
 
     workflow = get_object_or_404(Workflow, pk=workflow_pk)

wagtail/admin/wagtail_hooks.py

@@ -25,7 +25,7 @@ from wagtail.admin.viewsets import viewsets
 from wagtail.admin.widgets import Button, ButtonWithDropdownFromHook, PageListingButton
 from wagtail.core import hooks
 from wagtail.core.models import UserPagePermissionsProxy
-from wagtail.core.permissions import collection_permission_policy
+from wagtail.core.permissions import collection_permission_policy, workflow_permission_policy
 from wagtail.core.whitelist import allow_without_attributes, attribute_rule, check_url
 
 
@@ -101,9 +101,16 @@ def register_collections_menu_item():
     return CollectionsMenuItem(_('Collections'), reverse('wagtailadmin_collections:index'), icon_name='folder-open-1', order=700)
 
 
+class WorkflowsMenuItem(MenuItem):
+    def is_shown(self, request):
+        return workflow_permission_policy.user_has_any_permission(
+            request.user, ['add', 'change', 'delete']
+        )
+
+
 @hooks.register('register_settings_menu_item')
 def register_workflows_menu_item():
-    return MenuItem(_('Workflows'), reverse('wagtailadmin_workflows:index'), icon_name='clipboard-list', order=100)
+    return WorkflowsMenuItem(_('Workflows'), reverse('wagtailadmin_workflows:index'), icon_name='clipboard-list', order=100)
 
 
 @hooks.register('register_page_listing_buttons')

wagtail/core/wagtail_hooks.py

@@ -59,3 +59,19 @@ def register_collection_permissions():
         content_type__app_label='wagtailcore',
         codename__in=['add_collection', 'change_collection', 'delete_collection']
     )
+
+
+@hooks.register('register_permissions')
+def register_workflow_permissions():
+    return Permission.objects.filter(
+        content_type__app_label='wagtailcore',
+        codename__in=['add_workflow', 'change_workflow', 'delete_workflow']
+    )
+
+
+@hooks.register('register_permissions')
+def register_task_permissions():
+    return Permission.objects.filter(
+        content_type__app_label='wagtailcore',
+        codename__in=['add_task', 'change_task', 'delete_task']
+    )