Block edits on activation date and permission groups in pools with registrations

Author: Tinde von Wachenfeldt

lego/apps/events/models.py

@@ -849,6 +849,9 @@ def decrement(self) -> Pool:
         self.save(update_fields=["counter"])
         return self
 
+    def permission_group_ids(self) -> set[int]:
+        return set(self.permission_groups.values_list("id", flat=True))
+
     @abakus_cached_property
     def all_permission_groups(self):
         groups = self.permission_groups.all()

lego/apps/events/serializers/events.py

@@ -435,9 +435,20 @@ def create(self, validated_data):
         with transaction.atomic():
             event = super().create(validated_data)
             for pool in pools:
-                permission_groups = pool.pop("permission_groups")
-                created_pool = Pool.objects.create(event=event, **pool)
-                created_pool.permission_groups.set(permission_groups)
+                permission_groups = pool.get("permission_groups", [])
+                pool_data = {
+                    "name": pool.get("name"),
+                    "capacity": pool.get("capacity"),
+                    "activation_date": pool.get("activation_date"),
+                    "permission_groups": [
+                        getattr(gr, "id", gr) for gr in permission_groups
+                    ],
+                }
+                pool_serializer = PoolCreateAndUpdateSerializer(
+                    data=pool_data, context={**self.context, "event": event}
+                )
+                pool_serializer.is_valid(raise_exception=True)
+                pool_serializer.save()
             return event
 
     def update(self, instance, validated_data):
@@ -454,24 +465,40 @@ def update(self, instance, validated_data):
             pools[0]["capacity"] = 0
         with transaction.atomic():
             if pools is not None:
-                existing_pools = list(instance.pools.all().values_list("id", flat=True))
+                existing_ids = set(instance.pools.values_list("id", flat=True))
                 for pool in pools:
-                    pool_id = pool.get("id", None)
-                    if pool_id in existing_pools:
-                        existing_pools.remove(pool_id)
-                    permission_groups = pool.pop("permission_groups")
-                    created_pool = Pool.objects.update_or_create(
-                        event=instance,
-                        id=pool_id,
-                        defaults={
-                            "name": pool.get("name"),
-                            "capacity": pool.get("capacity", 0),
-                            "activation_date": pool.get("activation_date"),
-                        },
-                    )[0]
-                    created_pool.permission_groups.set(permission_groups)
-                for pool_id in existing_pools:
-                    Pool.objects.get(id=pool_id).delete()
+                    pool_id = pool.get("id")
+                    pool_instance = (
+                        Pool.objects.filter(id=pool_id, event=instance)
+                        .select_for_update()
+                        .first()
+                        if pool_id
+                        else None
+                    )
+                    if pool_instance:
+                        existing_ids.discard(pool_id)
+                    permission_groups = pool.get("permission_groups", [])
+                    pool_data = {
+                        "name": pool.get("name"),
+                        "capacity": pool.get("capacity"),
+                        "activation_date": pool.get("activation_date"),
+                        "permission_groups": [
+                            getattr(gr, "id", gr) for gr in permission_groups
+                        ],
+                    }
+                    pool_serializer = PoolCreateAndUpdateSerializer(
+                        instance=pool_instance,
+                        data=pool_data,
+                        context={**self.context, "event": instance},
+                        partial=True,
+                    )
+                    pool_serializer.is_valid(raise_exception=True)
+                    pool_serializer.save()
+                if existing_ids:
+                    for pool_obj in Pool.objects.filter(
+                        event=instance, id__in=existing_ids
+                    ).iterator():
+                        pool_obj.delete()
             return super().update(instance, validated_data)
 
 

lego/apps/events/serializers/pools.py

@@ -1,7 +1,7 @@
 from rest_framework import serializers
 
 from lego.apps.events.fields import RegistrationCountField
-from lego.apps.events.models import Event, Pool
+from lego.apps.events.models import Pool
 from lego.apps.events.serializers.registrations import (
     RegistrationPaymentReadSerializer,
     RegistrationReadDetailedAllergiesSerializer,
@@ -29,14 +29,6 @@ class Meta:
         )
         read_only = True
 
-    def create(self, validated_data):
-        event = Event.objects.get(pk=self.context["view"].kwargs["event_pk"])
-        permission_groups = validated_data.pop("permission_groups")
-        pool = Pool.objects.create(event=event, **validated_data)
-        pool.permission_groups.set(permission_groups)
-
-        return pool
-
 
 class PoolReadAuthSerializer(PoolReadSerializer):
     registrations = serializers.SerializerMethodField()
@@ -90,10 +82,32 @@ class Meta:
             "registrations": {"read_only": True},
         }
 
+    def validate(self, attrs):
+        instance = getattr(self, "instance", None)
+        if not instance or instance.registration_count < 1:
+            return attrs
+
+        if "permission_groups" in attrs:
+            new_ids = {getattr(gr, "id", gr) for gr in attrs["permission_groups"]}
+            old_ids = instance.permission_group_ids()
+            if new_ids != old_ids:
+                raise serializers.ValidationError(
+                    {
+                        "permission_groups": "Group edits are disabled for pools with registrations."
+                    }
+                )
+        if "activation_date" in attrs:
+            if attrs["activation_date"] != instance.activation_date:
+                raise serializers.ValidationError(
+                    {
+                        "activation_date": "Time travel is disabled for pools with registrations."
+                    }
+                )
+        return attrs
+
     def create(self, validated_data):
-        event = Event.objects.get(pk=self.context["view"].kwargs["event_pk"])
+        event = validated_data.pop("event", None) or self.context.get("event")
         permission_groups = validated_data.pop("permission_groups")
         pool = Pool.objects.create(event=event, **validated_data)
         pool.permission_groups.set(permission_groups)
-
         return pool

lego/apps/events/tests/test_events_api.py

@@ -1000,6 +1000,31 @@ def test_delete_pool_without_registrations_as_abakus(self):
         pool_response = self.client.delete(_get_pools_detail_url(1, pool.id))
         self.assertEqual(pool_response.status_code, status.HTTP_403_FORBIDDEN)
 
+    def test_patch_permission_groups_in_pool_with_registrations(self):
+        """Test that change of permission group is not possible in pool with registrations"""
+        AbakusGroup.objects.get(name="Bedkom").add_user(self.abakus_user)
+        self.client.force_authenticate(self.abakus_user)
+        new_group = AbakusGroup.objects.get(name="Webkom")
+        pool_response = self.client.patch(
+            _get_pools_detail_url(1, 1),
+            {"permissionGroups": [new_group.id]},
+            format="json",
+        )
+        self.assertEqual(pool_response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn("permissionGroups", pool_response.json())
+
+    def test_patch_activation_date_in_pool_with_registrations(self):
+        """Test that change of activation date is not possible in pool with registrations"""
+        AbakusGroup.objects.get(name="Bedkom").add_user(self.abakus_user)
+        self.client.force_authenticate(self.abakus_user)
+        pool_response = self.client.patch(
+            _get_pools_detail_url(1, 1),
+            {"activationDate": timezone.now().isoformat()},
+            format="json",
+        )
+        self.assertEqual(pool_response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn("activationDate", pool_response.json())
+
 
 @mock.patch("lego.apps.events.views.verify_captcha", return_value=True)
 class RegistrationsTransactionTestCase(BaseAPITransactionTestCase):

lego/apps/events/views.py

@@ -346,11 +346,18 @@ class PoolViewSet(
     serializer_class = PoolCreateAndUpdateSerializer
 
     def get_queryset(self):
-        event_id = self.kwargs.get("event_pk", None)
+        event_id = self.kwargs.get("event_pk")
         return Pool.objects.filter(event=event_id).prefetch_related(
             "permission_groups", "registrations"
         )
 
+    def get_serializer_context(self):
+        context = super().get_serializer_context()
+        event_id = self.kwargs.get("event_pk")
+        if event_id:
+            context["event"] = get_object_or_404(Event, pk=event_id)
+        return context
+
     def destroy(self, request, *args, **kwargs):
         try:
             return super().destroy(request, *args, **kwargs)