[feature] Improve the mass command admin action

- Move the action into BatchCommandAdmin as a static method guarded by
  the device change permission
- Hide the target fields when the command runs system wide
- Move the selection warning to the messages framework
- Add unit and selenium tests for the organization scope and the
  permissions of the action

Author: dee077

docs/user/shell-commands.rst

@@ -221,13 +221,15 @@ A mass command can also be started from the device list: select the
 devices with their checkboxes, choose *Execute mass command* from the
 actions dropdown and click *Go*.
 
-The first step opens with the selection already applied: the devices it
-will run on are shown above the targets, and the organization is filled in
-and cannot be changed, while device group and location are not asked for,
-since the devices are already known.
+The first step opens with the selection already applied: a message at the
+top of the page states how many devices the command will run on, and the
+organization is filled in and cannot be changed, while device group and
+location are not asked for, since the devices are already known.
 
 The selected devices must belong to the same organization, otherwise the
-action refuses to start.
+action refuses to start. The exception is a superuser selecting every
+device of the system: the command then runs on all of them and no target
+is asked for.
 
 The rest of the workflow is the same as described below: the devices can
 still be reviewed and left out before executing.

openwisp_controller/connection/admin.py

@@ -103,9 +103,12 @@ class Media:
             ]
         }
 
-    def __init__(self, *args, request=None, device_ids=None, **kwargs):
+    def __init__(
+        self, *args, request=None, device_ids=None, system_wide=False, **kwargs
+    ):
         super().__init__(*args, **kwargs)
         self.request = request
+        self.system_wide = system_wide
         self.device_ids = self._scope_devices(device_ids)
         if self.device_ids:
             self.fields["devices"].initial = ",".join(self.device_ids)
@@ -167,10 +170,7 @@ def clean(self):
             )
         if len(self._organization_ids) > 1:
             raise ValidationError(
-                _(
-                    "All devices must belong to the same organization,"
-                    " unless it is a system wide command."
-                )
+                _("All devices must belong to the same organization.")
             )
         if self.request is None or self.request.user.is_superuser:
             return cleaned_data
@@ -195,13 +195,12 @@ def fieldsets(self):
         fields: the sections are listed here, in the order they are shown,
         and the template renders each field with the admin markup.
         """
-        targets = ("organization",)
-        if not self.device_ids:
-            targets += ("location", "group")
-        sections = (
-            (_("Command"), ("type", "input", "label", "notes")),
-            (_("Targets"), targets),
-        )
+        sections = [(_("Command"), ("type", "input", "label", "notes"))]
+        if not self.system_wide:
+            targets = ("organization",)
+            if not self.device_ids:
+                targets += ("location", "group")
+            sections.append((_("Targets"), targets))
         return [
             (title, [self[field_name] for field_name in field_names])
             for title, field_names in sections
@@ -620,14 +619,27 @@ def execute_command_view(self, request):
         return self._render_execute_page(request, form)
 
     def _render_execute_page(self, request, form):
+        device_count = len(form.device_ids)
+        if device_count:
+            messages.warning(
+                request,
+                ngettext(
+                    "The command will run on the device you selected.",
+                    "The command will run on the %(count)d devices you selected.",
+                    device_count,
+                )
+                % {"count": device_count},
+            )
+        elif form.system_wide:
+            messages.warning(request, _("The command will run on all devices."))
         context = {
             **self.admin_site.each_context(request),
             "title": _("Execute mass command"),
             "opts": self.opts,
             "form": form,
             "media": form.media,
             "has_view_permission": self.has_view_permission(request),
-            "device_count": len(form.device_ids),
+            "device_count": device_count,
         }
         return TemplateResponse(request, self.execute_command_template, context)
 
@@ -1172,46 +1184,35 @@ def change_view(self, request, object_id, form_url="", extra_context=None):
             )
         return super().change_view(request, object_id, extra_context=extra_context)
 
-
-admin.site.register(BatchCommand, BatchCommandAdmin)
-
-
-@admin.action(
-    description=_("Execute mass command"),
-    permissions=["execute_mass_command"],
-)
-def execute_mass_command(modeladmin, request, queryset):
-    """Second entry point of the mass command workflow: the devices are
-    picked one by one instead of being matched by organization, group or
-    location. The selection travels in the form rather than in the session,
-    so it cannot outlive the wizard it belongs to.
-    """
-    # TODO: replace _registry with get_model_admin once Django 4.2 is dropped
-    batch_admin = modeladmin.admin_site._registry[BatchCommand]
-    batch_admin._check_add_permission(request)
-    organization_ids = set(queryset.values_list("organization_id", flat=True))
-    if len(organization_ids) > 1:
-        modeladmin.message_user(
-            request,
-            _(
-                "All devices must belong to the same organization,"
-                " unless it is a system wide command."
-            ),
-            messages.ERROR,
+    @staticmethod
+    @admin.action(description=_("Execute mass command"), permissions=["change"])
+    def execute_mass_command_admin_action(modeladmin, request, queryset):
+        """Second entry point of the mass command workflow: the devices are
+        picked one by one instead of being matched by organization, group or
+        location. The selection travels in the form rather than in the session,
+        so it cannot outlive the wizard it belongs to.
+        """
+        batch_admin = modeladmin.admin_site.get_model_admin(BatchCommand)
+        batch_admin._check_add_permission(request)
+        organization_ids = set(queryset.values_list("organization_id", flat=True))
+        if len(organization_ids) > 1:
+            if request.user.is_superuser and queryset.count() == Device.objects.count():
+                return batch_admin._render_execute_page(
+                    request,
+                    BatchCommandExecutionForm(request=request, system_wide=True),
+                )
+            modeladmin.message_user(
+                request,
+                _("All devices must belong to the same organization."),
+                messages.ERROR,
+            )
+            return HttpResponseRedirect(request.get_full_path())
+        form = BatchCommandExecutionForm(
+            request=request,
+            device_ids=[str(pk) for pk in queryset.values_list("pk", flat=True)],
         )
-        return HttpResponseRedirect(request.get_full_path())
-    form = BatchCommandExecutionForm(
-        request=request,
-        device_ids=[str(pk) for pk in queryset.values_list("pk", flat=True)],
-    )
-    return batch_admin._render_execute_page(request, form)
+        return batch_admin._render_execute_page(request, form)
 
 
-def has_execute_mass_command_permission(self, request):
-    options = BatchCommand._meta
-    return request.user.has_perm(f"{options.app_label}.add_{options.model_name}")
-
-
-DeviceAdmin.execute_mass_command = execute_mass_command
-DeviceAdmin.has_execute_mass_command_permission = has_execute_mass_command_permission
-DeviceAdmin.actions += ["execute_mass_command"]
+admin.site.register(BatchCommand, BatchCommandAdmin)
+DeviceAdmin.actions += [BatchCommandAdmin.execute_mass_command_admin_action]

openwisp_controller/connection/static/connection/css/batch-command.css

@@ -307,10 +307,6 @@
 .execute-batch-command .form-row.field-input.errors .flex-container {
   display: none;
 }
-.execute-batch-command ul.selected-devices-message li {
-  padding: 15px 10px 15px 42px !important;
-  background-position: 15px 17px !important;
-}
 #main .form-row .ow-text-field {
   width: 320px;
 }

openwisp_controller/connection/templates/admin/connection/batch_command/execute_command.html

@@ -33,7 +33,7 @@
 {% block content %}
 <div id="content-main">
   {# the action renders this page from the device changelist, so the target is explicit #}
-  <form method="post" novalidate action="{% url opts|admin_urlname:'execute' %}">
+  <form method="post" novalidate class="execute-form" action="{% url opts|admin_urlname:'execute' %}">
     {% csrf_token %}
     {{ form.devices }}
     <nav class="stepper" aria-label="{% trans 'Execution progress' %}">
@@ -67,13 +67,6 @@
     the markup of the admin, so the page is styled like any other admin form
     {% endcomment %}
     {% for title, fields in form.fieldsets %}
-    {% if forloop.last and device_count %}
-    <ul class="messagelist selected-devices-message">
-      <li class="warning">
-        {% blocktrans count counter=device_count %}The command will run on the device you selected.{% plural %}The command will run on the {{ counter }} devices you selected.{% endblocktrans %}
-      </li>
-    </ul>
-    {% endif %}
     <fieldset class="module aligned">
       <h2>{{ title }}</h2>
       {% for field in fields %}

openwisp_controller/connection/tests/test_admin.py

@@ -883,7 +883,7 @@ def test_device_action_selection(self):
             self.assertNotContains(response, 'name="group"')
             self.assertNotContains(response, 'name="location"')
         with self.subTest("devices of different organizations are refused"):
-            response = self._post_device_action(devices + [device_org2])
+            response = self._post_device_action([devices[0], device_org2])
             self.assertRedirects(response, self.device_changelist_url)
             self.assertIn(
                 "All devices must belong to the same organization",
@@ -904,6 +904,74 @@ def test_device_action_selection(self):
             self._post_confirm(wizard["token"])
             batch = BatchCommand.objects.get()
             self.assertEqual(set(batch.devices.all()), set(devices))
+        with self.subTest("a single device is announced in the singular"):
+            response = self._post_device_action(devices[:1])
+            self.assertContains(
+                response, "The command will run on the device you selected."
+            )
+            self._start_wizard(devices=self._pk_list(devices[:1]))
+            response = self.client.get(self.confirm_url)
+            self.assertEqual(response.context["targets_display"], "1 selected device")
+
+    def test_device_action_system_wide(self):
+        org = self._get_org()
+        org2 = self._create_org(name="org2", slug="org2")
+        devices = [
+            self._create_device(
+                name=f"device{index}",
+                mac_address=f"00:11:22:33:44:0{index}",
+                organization=org,
+            )
+            for index in range(2)
+        ]
+        device_org2 = self._create_device(
+            name="device-org2", mac_address="00:11:22:33:44:09", organization=org2
+        )
+        self._login()
+        with self.subTest("selecting every device is system wide"):
+            response = self._post_device_action(
+                devices + [device_org2], select_across=True
+            )
+            self.assertEqual(response.status_code, 200)
+            form = response.context["form"]
+            self.assertEqual(form.device_ids, [])
+            self.assertEqual(response.context["device_count"], 0)
+            self.assertIsNone(form.fields["organization"].initial)
+            for field_name in ("organization", "group", "location"):
+                self.assertFalse(form.fields[field_name].disabled)
+            self.assertContains(response, "The command will run on all devices.")
+            self.assertNotContains(response, 'name="organization"')
+            self.assertNotContains(response, 'name="group"')
+            self.assertNotContains(response, 'name="location"')
+        with self.subTest("the excluded devices are left out of the batch"):
+            wizard = self._start_wizard()
+            self.client.get(self.confirm_url)
+            response = self._post_confirm(wizard["token"], excluded=str(device_org2.pk))
+            self.assertEqual(response.status_code, 302)
+            self.assertIn(
+                "Mass command executed successfully.", self._messages(response)
+            )
+            batch = BatchCommand.objects.get()
+            self.assertIsNone(batch.organization_id)
+            self.assertIsNone(batch.group_id)
+            self.assertIsNone(batch.location_id)
+            self.assertEqual(set(batch.devices.all()), set(devices))
+        with self.subTest("a partial multi organization selection is refused"):
+            self._create_device(
+                name="excluded-by-the-search",
+                mac_address="00:11:22:33:44:08",
+                organization=org2,
+            )
+            response = self._post_device_action(
+                devices + [device_org2],
+                select_across=True,
+                query={"q": "device"},
+            )
+            self.assertRedirects(response, f"{self.device_changelist_url}?q=device")
+            self.assertIn(
+                "All devices must belong to the same organization",
+                " ".join(self._messages(response)),
+            )
 
     def test_device_action_permissions_and_scope(self):
         org = self._get_org()
@@ -912,9 +980,10 @@ def test_device_action_permissions_and_scope(self):
         device2 = self._create_device(
             name="device2", mac_address="00:11:22:33:44:02", organization=org2
         )
-        device_admin = admin.site._registry[Device]
+        device_admin = admin.site.get_model_admin(Device)
         request = RequestFactory().get(self.device_changelist_url)
-        with self.subTest("the add permission is required"):
+        action_name = "execute_mass_command_admin_action"
+        with self.subTest("the device change permission is required"):
             viewer = self._create_operator(
                 organizations=[org], username="viewer", email="viewer@test.com"
             )
@@ -923,13 +992,19 @@ def test_device_action_permissions_and_scope(self):
                 Permission.objects.filter(codename="view_batchcommand")
             )
             request.user = viewer
-            self.assertFalse(device_admin.has_execute_mass_command_permission(request))
-            self.assertNotIn("execute_mass_command", device_admin.get_actions(request))
+            self.assertNotIn(action_name, device_admin.get_actions(request))
         with self.subTest("the operator group can use the action"):
             operator = self._create_operator(organizations=[org])
             request.user = operator
-            self.assertTrue(device_admin.has_execute_mass_command_permission(request))
-            self.assertIn("execute_mass_command", device_admin.get_actions(request))
+            self.assertIn(action_name, device_admin.get_actions(request))
+        with self.subTest("the batch command add permission is enforced"):
+            viewer.user_permissions.set(
+                Permission.objects.filter(
+                    codename__in=["view_device", "change_device", "view_batchcommand"]
+                )
+            )
+            self.client.force_login(viewer)
+            self.assertEqual(self._post_device_action([device]).status_code, 403)
         with self.subTest("devices of unmanaged organizations are dropped"):
             self.client.force_login(operator)
             response = self._post_execute(devices=self._pk_list([device, device2]))
@@ -951,6 +1026,28 @@ def test_device_action_permissions_and_scope(self):
                 "All devices must belong to the same organization",
                 " ".join(form.errors["__all__"]),
             )
+        with self.subTest("selecting every device is not system wide for operators"):
+            multi_operator = self._create_operator(
+                organizations=[org, org2], username="multi", email="multi@test.com"
+            )
+            self.client.force_login(multi_operator)
+            response = self._post_device_action([device, device2], select_across=True)
+            self.assertRedirects(response, self.device_changelist_url)
+            self.assertIn(
+                "All devices must belong to the same organization",
+                " ".join(self._messages(response)),
+            )
+        with self.subTest("an operator executes the devices it manages"):
+            self.client.force_login(operator)
+            wizard = self._start_wizard(devices=self._pk_list([device]))
+            self.assertEqual(wizard["device_ids"], [str(device.pk)])
+            self.assertEqual(wizard["organization_id"], str(org.pk))
+            self.client.get(self.confirm_url)
+            response = self._post_confirm(wizard["token"])
+            self.assertEqual(response.status_code, 302)
+            batch = BatchCommand.objects.get()
+            self.assertEqual(batch.organization_id, org.pk)
+            self.assertEqual(set(batch.devices.all()), {device})
 
     def test_changelist_multitenancy(self):
         org = self._get_org()