[change] Make batch user creation organization field readonly

docs/user/rest-api.rst

@@ -820,7 +820,7 @@ This API endpoint allows to use the features described in
     This API endpoint allows to use the features described in
     :doc:`importing_users` and :doc:`generating_users`.
 
-Responds only to **POST**, used to save a ``RadiusBatch`` instance.
+Responds only to **POST**, used to create a ``RadiusBatch`` instance.
 
 It is possible to generate the users of the ``RadiusBatch`` with two
 different strategies: csv or prefix.
@@ -856,6 +856,35 @@ When using this strategy, in the response you can find the field
 ``pdf_link`` which can be used to download a PDF file containing the user
 credentials.
 
+Batch retrieve and update
++++++++++++++++++++++++++
+
+.. code-block:: text
+
+    /api/v1/radius/batch/<id>/
+
+Responds to **GET**, **PUT**, and **PATCH** methods.
+
+Used to retrieve or update a ``RadiusBatch`` instance.
+
+.. note::
+
+    The ``organization`` field is **read-only** for existing batch objects
+    and cannot be changed via the API. This is intentional as changing the
+    organization after batch creation would be inconsistent.
+
+Parameters for **GET**:
+
+===== =================
+Param Description
+===== =================
+id    UUID of the batch
+===== =================
+
+The ``organization`` field is the only field that is explicitly
+**read-only** and cannot be updated via this endpoint. All other editable
+fields can be modified through **PUT** or **PATCH** requests.
+
 Batch CSV Download
 ++++++++++++++++++
 

openwisp_radius/admin.py

@@ -463,23 +463,21 @@ def delete_selected_batches(self, request, queryset):
         )
 
     def get_readonly_fields(self, request, obj=None):
-        readonly_fields = super(RadiusBatchAdmin, self).get_readonly_fields(
-            request, obj
-        )
+        readonly_fields = super().get_readonly_fields(request, obj)
         if obj and obj.status != "pending":
             return (
                 "strategy",
+                "organization",
                 "prefix",
                 "csvfile",
                 "number_of_users",
                 "users",
                 "expiration_date",
                 "name",
-                "organization",
                 "status",
             ) + readonly_fields
         elif obj:
-            return ("status",) + readonly_fields
+            return ("organization", "status") + readonly_fields
         return ("status",) + readonly_fields
 
     def has_delete_permission(self, request, obj=None):

openwisp_radius/api/serializers.py

@@ -515,6 +515,54 @@ class Meta:
         read_only_fields = ("status", "user_credentials", "created", "modified")
 
 
+class RadiusBatchUpdateSerializer(RadiusBatchSerializer):
+    """
+    Serializer for updating RadiusBatch objects.
+    Makes the organization field readonly for existing objects.
+    """
+
+    organization_slug = RadiusOrganizationField(
+        help_text=("Slug of the organization for creating radius batch."),
+        required=False,
+        label="organization",
+        slug_field="slug",
+        write_only=True,
+    )
+
+    def validate(self, data):
+        """
+        Validates partial updates while preserving model-level validation.
+        Ignores organization_slug if provided since organization is readonly.
+        """
+        data.pop("organization_slug", None)
+        strategy = data.get("strategy") or (self.instance and self.instance.strategy)
+        if (
+            strategy == "prefix"
+            and "number_of_users" in data
+            and not data.get("number_of_users")
+        ):
+            raise serializers.ValidationError(
+                {"number_of_users": _("The field number_of_users cannot be empty")}
+            )
+        validated_data = serializers.ModelSerializer.validate(self, data)
+        # Run model-level validation (full_clean) for update
+        if self.instance:
+            batch_data = {
+                field: validated_data.get(field, getattr(self.instance, field))
+                for field in validated_data
+            }
+            batch_data.pop("number_of_users", None)
+            for field, value in batch_data.items():
+                setattr(self.instance, field, value)
+            self.instance.full_clean()
+        return validated_data
+
+    class Meta:
+        model = RadiusBatch
+        fields = "__all__"
+        read_only_fields = ("created", "modified", "user_credentials", "organization")
+
+
 class PasswordResetSerializer(BasePasswordResetSerializer):
     input = serializers.CharField()
     email = None

openwisp_radius/api/urls.py

@@ -78,6 +78,9 @@ def get_api_urls(api_views=None):
                 name="phone_number_change",
             ),
             path("radius/batch/", api_views.batch, name="batch"),
+            path(
+                "radius/batch/<uuid:pk>/", api_views.batch_detail, name="batch_detail"
+            ),
             path(
                 "radius/organization/<slug:slug>/batch/<uuid:pk>/pdf/",
                 api_views.download_rad_batch_pdf,

openwisp_radius/api/views.py

@@ -34,6 +34,7 @@
     ListAPIView,
     ListCreateAPIView,
     RetrieveAPIView,
+    RetrieveUpdateAPIView,
     RetrieveUpdateDestroyAPIView,
 )
 from rest_framework.pagination import PageNumberPagination
@@ -72,6 +73,7 @@
     ChangePhoneNumberSerializer,
     RadiusAccountingSerializer,
     RadiusBatchSerializer,
+    RadiusBatchUpdateSerializer,
     RadiusGroupSerializer,
     RadiusUserGroupSerializer,
     UserRadiusUsageSerializer,
@@ -157,6 +159,34 @@ def validate_membership(self, user):
             raise serializers.ValidationError({"non_field_errors": [message]})
 
 
+class BatchUpdateView(ThrottledAPIMixin, RetrieveUpdateAPIView):
+    """
+    API view for updating existing RadiusBatch objects.
+    Organization field is readonly for existing objects.
+    """
+
+    authentication_classes = (BearerAuthentication, SessionAuthentication)
+    permission_classes = (IsOrganizationManager, IsAdminUser, DjangoModelPermissions)
+    queryset = RadiusBatch.objects.none()
+    serializer_class = RadiusBatchSerializer
+
+    def get_queryset(self):
+        """Filter batches by user's organization membership"""
+        user = self.request.user
+        if user.is_superuser:
+            return RadiusBatch.objects.all()
+        return RadiusBatch.objects.filter(organization__in=user.organizations_managed)
+
+    def get_serializer_class(self):
+        """Use RadiusBatchUpdateSerializer for PUT/PATCH requests"""
+        if self.request.method in ("PUT", "PATCH"):
+            return RadiusBatchUpdateSerializer
+        return RadiusBatchSerializer
+
+
+batch_detail = BatchUpdateView.as_view()
+
+
 class DownloadRadiusBatchPdfView(ThrottledAPIMixin, DispatchOrgMixin, RetrieveAPIView):
     authentication_classes = (BearerAuthentication, SessionAuthentication)
     permission_classes = (IsOrganizationManager, IsAdminUser, DjangoModelPermissions)

openwisp_radius/tests/test_api/test_api.py

@@ -1556,6 +1556,112 @@ def test_radius_user_group_serializer_without_view_context(self):
         self.assertEqual(serializer._user, None)
         self.assertEqual(serializer.fields["group"].queryset.count(), 0)
 
+    def _get_admin_auth_header(self):
+        """Helper method to get admin authentication header"""
+        login_payload = {"username": "admin", "password": "tester"}
+        login_url = reverse("radius:user_auth_token", args=[self.default_org.slug])
+        login_response = self.client.post(login_url, data=login_payload)
+        self.assertEqual(login_response.status_code, status.HTTP_200_OK)
+        return f"Bearer {login_response.json()['key']}"
+
+    def test_batch_update_organization_readonly(self):
+        """
+        Test that organization field is readonly when updating RadiusBatch objects
+        """
+        data = self._radius_batch_prefix_data()
+        response = self._radius_batch_post_request(data)
+        self.assertEqual(response.status_code, 201)
+        batch = RadiusBatch.objects.get()
+        original_org = batch.organization
+
+        new_org = self._create_org(**{"name": "new-org", "slug": "new-org"})
+
+        header = self._get_admin_auth_header()
+
+        url = reverse("radius:batch_detail", args=[batch.pk])
+        update_data = {
+            "name": "updated-batch-name",
+            "organization": str(new_org.pk),
+        }
+        response = self.client.patch(
+            url,
+            json.dumps(update_data),
+            HTTP_AUTHORIZATION=header,
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, 200)
+
+        batch.refresh_from_db()
+        self.assertEqual(batch.organization, original_org)
+        self.assertEqual(batch.name, "updated-batch-name")
+
+    def test_batch_retrieve_and_update_api(self):
+        """
+        Test retrieving and updating RadiusBatch objects via API
+        """
+        data = self._radius_batch_prefix_data()
+        response = self._radius_batch_post_request(data)
+        self.assertEqual(response.status_code, 201)
+        batch = RadiusBatch.objects.get()
+
+        header = self._get_admin_auth_header()
+
+        url = reverse("radius:batch_detail", args=[batch.pk])
+        response = self.client.get(url, HTTP_AUTHORIZATION=header)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data["name"], batch.name)
+        self.assertEqual(str(response.data["organization"]), str(batch.organization.pk))
+
+        update_data = {
+            "name": "updated-batch-name",
+            "strategy": "prefix",
+            "prefix": batch.prefix,
+            "organization_slug": batch.organization.slug,
+        }
+        response = self.client.put(
+            url,
+            json.dumps(update_data),
+            HTTP_AUTHORIZATION=header,
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, 200)
+        batch.refresh_from_db()
+        self.assertEqual(batch.name, "updated-batch-name")
+
+    def test_batch_update_permissions(self):
+        """
+        Test that proper permissions are required for updating RadiusBatch objects
+        """
+        data = self._radius_batch_prefix_data()
+        response = self._radius_batch_post_request(data)
+        self.assertEqual(response.status_code, 201)
+        batch = RadiusBatch.objects.get()
+
+        url = reverse("radius:batch_detail", args=[batch.pk])
+
+        response = self.client.patch(url, {"name": "new-name"})
+        self.assertEqual(response.status_code, 401)
+
+        user = self._get_user()
+        user_token = Token.objects.create(user=user)
+        header = f"Bearer {user_token.key}"
+        response = self.client.patch(
+            url,
+            json.dumps({"name": "new-name"}),
+            HTTP_AUTHORIZATION=header,
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, 403)
+
+        header = self._get_admin_auth_header()
+        response = self.client.patch(
+            url,
+            json.dumps({"name": "new-name"}),
+            HTTP_AUTHORIZATION=header,
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, 200)
+
 
 class TestTransactionApi(AcctMixin, ApiTokenMixin, BaseTransactionTestCase):
     def test_user_radius_usage_view(self):

tests/openwisp2/sample_radius/api/views.py

@@ -1,6 +1,7 @@
 from openwisp_radius.api.freeradius_views import AccountingView as BaseAccountingView
 from openwisp_radius.api.freeradius_views import AuthorizeView as BaseAuthorizeView
 from openwisp_radius.api.freeradius_views import PostAuthView as BasePostAuthView
+from openwisp_radius.api.views import BatchUpdateView as BaseBatchUpdateView
 from openwisp_radius.api.views import BatchView as BaseBatchView
 from openwisp_radius.api.views import ChangePhoneNumberView as BaseChangePhoneNumberView
 from openwisp_radius.api.views import CreatePhoneTokenView as BaseCreatePhoneTokenView
@@ -104,6 +105,10 @@ class RadiusAccountingView(BaseRadiusAccountingView):
     pass
 
 
+class BatchUpdateView(BaseBatchUpdateView):
+    pass
+
+
 authorize = AuthorizeView.as_view()
 postauth = PostAuthView.as_view()
 accounting = AccountingView.as_view()
@@ -122,6 +127,7 @@ class RadiusAccountingView(BaseRadiusAccountingView):
 change_phone_number = ChangePhoneNumberView.as_view()
 download_rad_batch_pdf = DownloadRadiusBatchPdfView.as_view()
 radius_accounting = RadiusAccountingView.as_view()
+batch_detail = BatchUpdateView.as_view()
 radius_group_list = RadiusGroupListView.as_view()
 radius_group_detail = RadiusGroupDetailView.as_view()
 radius_user_group_list = RadiusUserGroupListCreateView.as_view()