Allow manual lock even if WorkflowLock is currently applied

wagtail/admin/templates/wagtailadmin/shared/side_panels/includes/status/locked.html

@@ -28,11 +28,11 @@
 {% endblock %}
 
 {% block action %}
-    {% if user_can_unlock and lock %}
+    {% if user_can_unlock %}
         {% trans 'Unlock' as unlock_text %}
         {% include 'wagtailadmin/shared/side_panels/includes/side_panel_button.html' with data_url=unlock_url text=unlock_text %}
     {% endif %}
-    {% if user_can_lock and not lock %}
+    {% if user_can_lock %}
         {% trans 'Lock' as lock_text %}
         {% include 'wagtailadmin/shared/side_panels/includes/side_panel_button.html' with data_url=lock_url text=lock_text %}
     {% endif %}

wagtail/admin/tests/test_workflows.py

@@ -1,7 +1,7 @@
 import io
 import json
 import logging
-from unittest import expectedFailure, mock
+from unittest import expectedFailure, mock, skip
 
 from django.conf import settings
 from django.contrib.admin.utils import quote
@@ -15,7 +15,11 @@ from freezegun import freeze_time
 from openpyxl import load_workbook
 
 from wagtail.admin.admin_url_finder import AdminURLFinder
-from wagtail.admin.utils import get_admin_base_url, get_latest_str
+from wagtail.admin.utils import (
+    get_admin_base_url,
+    get_latest_str,
+    get_user_display_name,
+)
 from wagtail.models import (
     GroupApprovalTask,
     Page,
@@ -1101,10 +1105,16 @@ class BaseSnippetWorkflowTests(BasePageWorkflowTests):
             content_type__app_label="tests",
             codename=f"publish_{self.model._meta.model_name}",
         )
+        self.lock_permission = Permission.objects.filter(
+            content_type__app_label="tests",
+            codename=f"lock_{self.model._meta.model_name}",
+        ).first()
         self.submitter.user_permissions.add(self.edit_permission)
         self.moderator.user_permissions.add(
             self.edit_permission, self.publish_permission
         )
+        if self.lock_permission:
+            self.moderator.user_permissions.add(self.lock_permission)
 
     @property
     def model_name(self):
@@ -1174,12 +1184,179 @@ class TestSubmitPageToWorkflow(BasePageWorkflowTests):
         self.post("submit")
 
         response = self.client.get(edit_url)
+
+        # Should show the moderation status
         self.assertRegex(
             response.content.decode("utf-8"),
             r"Sent to[\s|\n]+{}".format(self.object.current_workflow_task.name),
         )
+        self.assertContains(response, "In Moderation")
         self.assertNotContains(response, "Draft")
 
+    def test_submit_for_approval_changes_lock_status(self):
+        edit_url = self.get_url("edit")
+
+        response = self.client.get(edit_url)
+
+        # Should show the lock information as unlocked
+        self.assertContains(response, "Unlocked", count=1)
+        self.assertContains(
+            response, f"Anyone can edit this {self.model_name}", count=1
+        )
+        self.assertNotContains(response, "Locked by workflow")
+        self.assertNotContains(
+            response,
+            f"Only reviewers can edit and approve the {self.model_name}",
+        )
+        self.assertNotContains(response, self.get_url("lock"))
+
+        # submit for approval
+        self.post("submit")
+
+        response = self.client.get(edit_url)
+
+        # Should show the lock information as locked
+        self.assertContains(response, "Locked by workflow", count=1)
+        self.assertContains(
+            response,
+            f"Only reviewers can edit and approve the {self.model_name}",
+            count=1,
+        )
+
+        self.assertNotContains(response, "Unlocked")
+        self.assertNotContains(
+            response,
+            f"Anyone can edit this {self.model_name}",
+        )
+
+        # Should be unable to unlock
+        self.assertNotContains(response, self.get_url("unlock"))
+
+    def test_can_manual_lock_while_in_workflow(self):
+        edit_url = self.get_url("edit")
+
+        # Submit to workflow as submitter
+        self.post("submit")
+
+        # Login as a moderator to have lock permission
+        self.login(self.moderator)
+
+        response = self.client.get(edit_url)
+
+        # Should show the lock information as unlocked
+        self.assertContains(response, "Unlocked", count=1)
+        self.assertContains(
+            response,
+            f"Reviewers can edit this {self.model_name} – lock it to prevent other reviewers from editing",
+            count=1,
+        )
+        self.assertContains(response, self.get_url("lock"), count=1)
+        self.assertNotContains(response, "Locked by workflow")
+        self.assertNotContains(
+            response,
+            f"Only reviewers can edit and approve the {self.model_name}",
+        )
+        self.assertNotContains(response, self.get_url("unlock"))
+
+    def test_can_unlock_manual_lock_while_in_workflow(self):
+        edit_url = self.get_url("edit")
+
+        # Submit to workflow as submitter
+        self.post("submit")
+
+        # Login as a moderator to have lock permission
+        self.login(self.moderator)
+
+        # Lock the object
+        self.client.post(self.get_url("lock"))
+
+        response = self.client.get(edit_url)
+
+        # Should show the lock information as locked
+        self.assertContains(response, "Locked by you", count=1)
+        self.assertContains(
+            response,
+            f"Only you can make changes while the {self.model_name} is locked",
+            count=1,
+        )
+        # One in the side panel, one in the message banner
+        self.assertContains(response, self.get_url("unlock"), count=2)
+        self.assertNotContains(response, "Locked by workflow")
+        self.assertNotContains(
+            response,
+            f"Only reviewers can edit and approve the {self.model_name}",
+        )
+        self.assertNotContains(response, self.get_url("lock"))
+
+    def test_can_unlock_other_users_manual_lock_while_in_workflow(self):
+        edit_url = self.get_url("edit")
+
+        # Submit to workflow as submitter
+        self.post("submit")
+
+        # Login as a moderator to have lock permission
+        self.login(self.moderator)
+
+        # Lock the object
+        self.client.post(self.get_url("lock"))
+
+        # Login as a superuser that has unlock permission
+        self.login(self.superuser)
+
+        response = self.client.get(edit_url)
+
+        display_name = get_user_display_name(self.moderator)
+        # Should show the lock information as locked
+        self.assertContains(response, "Locked by another user", count=1)
+        self.assertContains(
+            response,
+            f"Only {display_name} can make changes while the {self.model_name} is locked",
+            count=1,
+        )
+        # One in the side panel, one in the message banner
+        self.assertContains(response, self.get_url("unlock"), count=2)
+        self.assertNotContains(response, "Locked by workflow")
+        self.assertNotContains(
+            response,
+            f"Only reviewers can edit and approve the {self.model_name}",
+        )
+        self.assertNotContains(response, self.get_url("lock"))
+
+    def test_cannot_unlock_other_users_manual_lock_while_in_workflow(self):
+        edit_url = self.get_url("edit")
+
+        # Submit to workflow as submitter
+        self.post("submit")
+
+        # Login as a superuser to have lock permission
+        self.login(self.superuser)
+
+        # Lock the object
+        self.client.post(self.get_url("lock"))
+
+        # Login as a moderator that does not have unlock permission
+        # according to the workflow
+        self.login(self.moderator)
+
+        response = self.client.get(edit_url)
+
+        display_name = get_user_display_name(self.superuser)
+        # Should show the lock information as locked
+        self.assertContains(response, "Locked by another user", count=1)
+        self.assertContains(
+            response,
+            f"Only {display_name} can make changes while the {self.model_name} is locked",
+            count=1,
+        )
+        # Has no permission to unlock
+        self.assertNotContains(response, self.get_url("unlock"))
+        self.assertNotContains(response, "Locked by workflow")
+        self.assertNotContains(
+            response,
+            f"Only reviewers can edit and approve the {self.model_name}",
+        )
+        self.assertNotContains(response, self.get_url("lock"))
+
     def test_workflow_action_menu_items(self):
         edit_url = self.get_url("edit")
 
@@ -1662,6 +1839,48 @@ class TestSubmitSnippetToWorkflowNotLockable(TestSubmitSnippetToWorkflow):
             '<button type="submit" class="button action-save warning" disabled>',
         )
 
+    def test_submit_for_approval_changes_lock_status(self):
+        edit_url = self.get_url("edit")
+
+        # submit for approval
+        self.post("submit")
+
+        # Login as a moderator
+        self.login(self.moderator)
+
+        response = self.client.get(edit_url)
+
+        # Model is not lockable, should not show any lock information or buttons
+
+        self.assertNotContains(response, "Unlocked")
+        self.assertNotContains(response, f"Anyone can edit this {self.model_name}")
+        self.assertNotContains(
+            response,
+            f"Reviewers can edit this {self.model_name} – lock it to prevent other reviewers from editing",
+        )
+        self.assertNotContains(response, "Locked by workflow")
+        self.assertNotContains(
+            response,
+            f"Only reviewers can edit and approve the {self.model_name}",
+        )
+        self.assertNotContains(response, "Locked by another user")
+
+    @skip("Model is not lockable")
+    def test_can_manual_lock_while_in_workflow(self):
+        pass
+
+    @skip("Model is not lockable")
+    def test_can_unlock_manual_lock_while_in_workflow(self):
+        pass
+
+    @skip("Model is not lockable")
+    def test_can_unlock_other_users_manual_lock_while_in_workflow(self):
+        pass
+
+    @skip("Model is not lockable")
+    def test_cannot_unlock_other_users_manual_lock_while_in_workflow(self):
+        pass
+
 
 @freeze_time("2020-03-31 12:00:00")
 class TestApproveRejectPageWorkflow(BasePageWorkflowTests):

wagtail/admin/ui/side_panels.py

@@ -6,7 +6,6 @@ from django.utils.text import capfirst
 from django.utils.translation import gettext_lazy, ngettext
 
 from wagtail.admin.ui.components import Component
-from wagtail.locks import BasicLock
 from wagtail.models import (
     DraftStateMixin,
     LockableMixin,
@@ -152,15 +151,19 @@ class BaseStatusSidePanel(BaseSidePanel):
 
         return context
 
-    def get_lock_context(self):
+    def get_lock_context(self, parent_context):
         self.lock = None
         lock_context = {}
         if self.locking_enabled:
             self.lock = self.object.get_lock()
             if self.lock:
-                lock_context = self.lock.get_context_for_user(self.request.user)
+                lock_context = self.lock.get_context_for_user(
+                    self.request.user, parent_context
+                )
         return {
             "lock": self.lock,
+            "user_can_lock": parent_context.get("user_can_lock"),
+            "user_can_unlock": parent_context.get("user_can_unlock"),
             "lock_context": lock_context,
         }
 
@@ -178,7 +181,7 @@ class BaseStatusSidePanel(BaseSidePanel):
         context["base_model_name"] = context["model_name"]
         context["status_templates"] = self.get_status_templates(context)
         context.update(self.get_scheduled_publishing_context())
-        context.update(self.get_lock_context())
+        context.update(self.get_lock_context(parent_context))
         if self.object.pk:
             context.update(self.get_usage_context())
         return context
@@ -224,9 +227,6 @@ class PageStatusSidePanel(BaseStatusSidePanel):
                     "revisions_compare_url_name": "wagtailadmin_pages:revisions_compare",
                     "lock_url": reverse("wagtailadmin_pages:lock", args=(page.id,)),
                     "unlock_url": reverse("wagtailadmin_pages:unlock", args=(page.id,)),
-                    "user_can_lock": user_perms.for_page(page).can_lock(),
-                    "user_can_unlock": isinstance(self.lock, BasicLock)
-                    and user_perms.for_page(page).can_unlock(),
                     "locale": None,
                     "translations": [],
                 }

wagtail/admin/views/generic/mixins.py

@@ -19,7 +19,7 @@ from wagtail.admin import messages
 from wagtail.admin.templatetags.wagtailadmin_tags import user_display_name
 from wagtail.admin.ui.tables import TitleColumn
 from wagtail.admin.utils import get_latest_str
-from wagtail.locks import BasicLock, ScheduledForPublishLock
+from wagtail.locks import BasicLock, ScheduledForPublishLock, WorkflowLock
 from wagtail.log_actions import log
 from wagtail.log_actions import registry as log_registry
 from wagtail.models import (
@@ -270,14 +270,9 @@ class CreateEditViewOptionalFeaturesMixin:
 
         # Workflow lock/unlock methods take precedence before the base
         # "lock" and "unlock" permissions -- see PagePermissionTester for reference
-        if permission == "lock":
-            if self.lock:
-                return False
-            if self.current_workflow_task:
-                return self.current_workflow_task.user_can_lock(self.object, user)
+        if permission == "lock" and self.current_workflow_task:
+            return self.current_workflow_task.user_can_lock(self.object, user)
         if permission == "unlock":
-            if not isinstance(self.lock, BasicLock):
-                return False
             # Allow unlocking even if the user does not have the 'unlock' permission
             # if they are the user who locked the object
             if self.object.locked_by_id == user.pk:
@@ -597,7 +592,9 @@ class CreateEditViewOptionalFeaturesMixin:
         if not self.locking_enabled:
             return {}
 
-        user_can_lock = not self.lock and self.user_has_permission("lock")
+        user_can_lock = (
+            not self.lock or isinstance(self.lock, WorkflowLock)
+        ) and self.user_has_permission("lock")
         user_can_unlock = (
             isinstance(self.lock, BasicLock)
         ) and self.user_has_permission("unlock")

wagtail/admin/views/pages/edit.py

@@ -21,7 +21,7 @@ from wagtail.admin.ui.side_panels import PageSidePanels
 from wagtail.admin.utils import get_valid_next_url_from_request
 from wagtail.admin.views.generic import HookResponseMixin
 from wagtail.exceptions import PageClassNotFoundError
-from wagtail.locks import BasicLock, ScheduledForPublishLock
+from wagtail.locks import BasicLock, ScheduledForPublishLock, WorkflowLock
 from wagtail.models import (
     COMMENTS_RELATION_NAME,
     Comment,
@@ -843,6 +843,7 @@ class EditView(TemplateResponseMixin, ContextMixin, HookResponseMixin, View):
 
     def get_context_data(self, **kwargs):
         context = super().get_context_data(**kwargs)
+        user_perms = UserPagePermissionsProxy(self.request.user)
         bound_panel = self.edit_handler.get_bound_panel(
             instance=self.page, request=self.request, form=self.form
         )
@@ -886,6 +887,10 @@ class EditView(TemplateResponseMixin, ContextMixin, HookResponseMixin, View):
                     "wagtailadmin_pages:confirm_workflow_cancellation",
                     args=(self.page.id,),
                 ),
+                "user_can_lock": (not self.lock or isinstance(self.lock, WorkflowLock))
+                and user_perms.for_page(self.page).can_lock(),
+                "user_can_unlock": isinstance(self.lock, BasicLock)
+                and user_perms.for_page(self.page).can_unlock(),
                 "locale": None,
                 "translations": [],
                 "media": bound_panel.media
@@ -896,8 +901,6 @@ class EditView(TemplateResponseMixin, ContextMixin, HookResponseMixin, View):
         )
 
         if getattr(settings, "WAGTAIL_I18N_ENABLED", False):
-            user_perms = UserPagePermissionsProxy(self.request.user)
-
             context.update(
                 {
                     "locale": self.page.locale,

wagtail/locks.py

@@ -55,7 +55,7 @@ class BaseLock:
         """
         return capfirst(_("No one can make changes while the %(model_name)s is locked"))
 
-    def get_context_for_user(self, user):
+    def get_context_for_user(self, user, parent_context=None):
         """
         Returns a context dictionary to use in templates for the given user.
         """
@@ -198,18 +198,47 @@ class WorkflowLock(BaseLock):
 
             return mark_safe(workflow_info + " " + reviewers_info)
 
-    def get_icon(self, user):
+    def get_icon(self, user, can_lock=False):
+        if can_lock:
+            return "lock-open"
         return super().get_icon(user)
 
-    def get_locked_by(self, user):
+    def get_locked_by(self, user, can_lock=False):
+        if can_lock:
+            return _("Unlocked")
         return _("Locked by workflow")
 
-    def get_description(self, user):
+    def get_description(self, user, can_lock=False):
+        if can_lock:
+            return capfirst(
+                _(
+                    "Reviewers can edit this %(model_name)s – lock it to prevent other reviewers from editing"
+                )
+                % {"model_name": self.model_name}
+            )
         return capfirst(
             _("Only reviewers can edit and approve the %(model_name)s")
             % {"model_name": self.model_name}
         )
 
+    def get_context_for_user(self, user, parent_context=None):
+        context = super().get_context_for_user(user, parent_context)
+        # BasicLock can still be applied on top of WorkflowLock, so we need to
+        # check if the user can lock the object based on the parent context.
+        # We're utilising the parent context instead of self.task.user_can_lock()
+        # because the latter does not take into account the user's permissions,
+        # while the parent context does and also checks self.task.user_can_lock().
+        if parent_context and "user_can_lock" in parent_context:
+            can_lock = parent_context.get("user_can_lock", False)
+            context.update(
+                {
+                    "icon": self.get_icon(user, can_lock),
+                    "locked_by": self.get_locked_by(user, can_lock),
+                    "description": self.get_description(user, can_lock),
+                }
+            )
+        return context
+
 
 class ScheduledForPublishLock(BaseLock):
     """