[feature] Mass command asynchronous execution pipeline

.github/workflows/ci.yml

@@ -5,13 +5,13 @@ on:
   push:
     branches:
       - master
+      - "gsoc26-*"
   pull_request:
     branches:
       - master
       - "1.1"
       - "1.2"
-      - "gsoc26-x509-certificate-generator-templates"
-      - "gsoc26-mass-commands"
+      - "gsoc26-*"
 
 jobs:
   build:
@@ -100,7 +100,7 @@ jobs:
         with:
           parallel: true
           format: cobertura
-          flag-name: python-${{ matrix.env.env }}
+          flag-name: python-${{ matrix.python-version }}-${{ matrix.django-version }}
           github-token: ${{ secrets.GITHUB_TOKEN }}
           fail-on-error: false
 

openwisp_controller/connection/api/serializers.py

@@ -12,6 +12,7 @@
 DeviceConnection = load_model("connection", "DeviceConnection")
 Credentials = load_model("connection", "Credentials")
 Device = load_model("config", "Device")
+BatchCommand = load_model("connection", "BatchCommand")
 
 
 class ValidatedDeviceFieldSerializer(ValidatedModelSerializer):
@@ -43,6 +44,10 @@ class CommandSerializer(ValidatedDeviceFieldSerializer):
         required=False,
         pk_field=serializers.UUIDField(format="hex_verbose"),
     )
+    batch_command = serializers.PrimaryKeyRelatedField(
+        read_only=True,
+        pk_field=serializers.UUIDField(format="hex_verbose"),
+    )
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -115,3 +120,70 @@ class Meta:
             "is_working": {"read_only": True},
         }
         read_only_fields = ("created", "modified")
+
+
+class BatchCommandExecuteSerializer(
+    FilterSerializerByOrgManaged, serializers.ModelSerializer
+):
+    type = serializers.CharField(source="command_type")
+    input = serializers.JSONField(
+        source="command_input", allow_null=True, required=False
+    )
+    devices = serializers.PrimaryKeyRelatedField(
+        many=True,
+        queryset=Device.objects.all(),
+        required=False,
+        allow_empty=True,
+        pk_field=serializers.UUIDField(format="hex_verbose"),
+    )
+    execute_all = serializers.BooleanField(required=False, default=True)
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        request = self.context.get("request")
+        if request and request.method == "GET":
+            self.fields["type"].required = False
+
+    class Meta:
+        model = BatchCommand
+        fields = (
+            "organization",
+            "type",
+            "input",
+            "devices",
+            "group",
+            "location",
+            "execute_all",
+        )
+        extra_kwargs = {
+            "organization": {"required": False, "allow_null": True},
+        }
+
+    def validate(self, data):
+        org = data.get("organization")
+        execute_all = data.get("execute_all", False)
+        devices = data.get("devices")
+        group = data.get("group")
+        location = data.get("location")
+        if not org and not self.context["request"].user.is_superuser:
+            raise serializers.ValidationError(
+                _("Only superusers can execute batch commands without an organization.")
+            )
+        if not execute_all and not org and not devices and not group and not location:
+            raise serializers.ValidationError(
+                _(
+                    "Specify at least one targeting option "
+                    "or set execute_all to true."
+                )
+            )
+        if devices:
+            for device in devices:
+                if org and device.organization_id != org.id:
+                    raise serializers.ValidationError(
+                        {
+                            "devices": _(
+                                "All devices must belong to the same organization."
+                            )
+                        }
+                    )
+        return data

openwisp_controller/connection/api/urls.py

@@ -40,6 +40,11 @@ def get_api_urls(api_views):
             api_views.deviceconnection_detail_view,
             name="deviceconnection_detail",
         ),
+        path(
+            "api/v1/controller/batch-command/execute/",
+            api_views.batch_command_execute_view,
+            name="batch_command_execute",
+        ),
     ]
 
 

openwisp_controller/connection/api/views.py

@@ -1,12 +1,16 @@
+from django.core.exceptions import ValidationError
 from django.utils.translation import gettext_lazy as _
 from drf_yasg import openapi
 from drf_yasg.utils import swagger_auto_schema
+from rest_framework import status
 from rest_framework.generics import (
+    GenericAPIView,
     ListCreateAPIView,
     RetrieveAPIView,
     RetrieveUpdateDestroyAPIView,
     get_object_or_404,
 )
+from rest_framework.response import Response
 from swapper import load_model
 
 from openwisp_utils.api.pagination import OpenWispPagination
@@ -17,6 +21,7 @@
     RelatedDeviceProtectedAPIMixin,
 )
 from .serializers import (
+    BatchCommandExecuteSerializer,
     CommandSerializer,
     CredentialSerializer,
     DeviceConnectionSerializer,
@@ -26,6 +31,7 @@
 Device = load_model("config", "Device")
 Credentials = load_model("connection", "Credentials")
 DeviceConnection = load_model("connection", "DeviceConnection")
+BatchCommand = load_model("connection", "BatchCommand")
 
 
 class BaseCommandView(RelatedDeviceProtectedAPIMixin):
@@ -138,6 +144,37 @@ class DeviceConnectionListCreateView(BaseDeviceConnection, ListCreateAPIView):
 DeviceConnenctionListCreateView = DeviceConnectionListCreateView
 
 
+class BatchCommandExecuteView(ProtectedAPIMixin, GenericAPIView):
+    model = BatchCommand
+    queryset = BatchCommand.objects.all()
+    serializer_class = BatchCommandExecuteSerializer
+
+    def post(self, request):
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        try:
+            batch = BatchCommand.execute(**serializer.validated_data)
+        except ValidationError as e:
+            return Response(
+                getattr(e, "message_dict", e.messages),
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        return Response({"batch": str(batch.pk)}, status=201)
+
+    def get(self, request):
+        serializer = self.get_serializer(data=request.query_params)
+        serializer.is_valid(raise_exception=True)
+        try:
+            data = BatchCommand.dry_run(**serializer.validated_data)
+        except ValidationError as e:
+            return Response(
+                getattr(e, "message_dict", e.messages),
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        data["devices"] = [str(d.pk) for d in data["devices"]]
+        return Response(data)
+
+
 class DeviceConnectionDetailView(BaseDeviceConnection, RetrieveUpdateDestroyAPIView):
     def get_object(self):
         queryset = self.filter_queryset(self.get_queryset())
@@ -158,3 +195,5 @@ def get_object(self):
 
 # TODO: remove in version 1.4
 deviceconnection_details_view = deviceconnection_detail_view
+
+batch_command_execute_view = BatchCommandExecuteView.as_view()