Restrict pool updates to active pools

Author: tindevw

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

@@ -423,57 +423,75 @@ def create(self, validated_data):
         event_status_type = validated_data.get(
             "event_status_type", Event._meta.get_field("event_status_type").default
         )
+
+        pools = self._pools_for_status(event_status_type, pools)
         require_auth = validated_data.get("require_auth", False)
         validated_data["require_auth"] = require_auth
-        if event_status_type == constants.TBA:
-            pools = []
-        elif event_status_type == constants.OPEN:
-            pools = []
-        elif event_status_type == constants.INFINITE:
-            pools = [pools[0]]
-            pools[0]["capacity"] = 0
+
         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)
+                pool_data = PoolCreateAndUpdateSerializer.extract_pool_data(pool)
+                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):
         pools = validated_data.pop("pools", None)
         event_status_type = validated_data.get(
             "event_status_type", Event._meta.get_field("event_status_type").default
         )
-        if event_status_type == constants.TBA:
-            pools = []
-        elif event_status_type == constants.OPEN:
-            pools = []
-        elif event_status_type == constants.INFINITE:
-            pools = [pools[0]]
-            pools[0]["capacity"] = 0
+
+        pools = self._pools_for_status(event_status_type, pools)
+
         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)
+
+                    pool_data = PoolCreateAndUpdateSerializer.extract_pool_data(pool)
+                    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)
 
+    def _pools_for_status(self, event_status_type, pools):
+        if pools is None:
+            return None
+        if event_status_type in (constants.TBA, constants.OPEN):
+            return []
+        if event_status_type == constants.INFINITE:
+            if not pools:
+                return []
+            pools = [pools[0]]
+            pools[0]["capacity"] = 0
+        return pools
+
 
 class FrontpageEventSerializer(serializers.ModelSerializer):
     cover = ImageField(required=False, options={"height": 500})

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,38 @@ class Meta:
             "registrations": {"read_only": True},
         }
 
+    def validate(self, attrs):
+        instance = getattr(self, "instance", None)
+        if not instance or not instance.is_activated:
+            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 active pools."}
+                )
+        if "activation_date" in attrs:
+            if attrs["activation_date"] != instance.activation_date:
+                raise serializers.ValidationError(
+                    {"activation_date": "Time travel is disabled for active pools."}
+                )
+        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
+
+    @staticmethod
+    def extract_pool_data(pool):
+        permission_groups = pool.get("permission_groups", [])
+        return {
+            "name": pool.get("name"),
+            "capacity": pool.get("capacity"),
+            "activation_date": pool.get("activation_date"),
+            "permission_groups": [getattr(g, "id", g) for g in permission_groups],
+        }

lego/apps/events/tests/test_events_api.py

@@ -1000,6 +1000,33 @@ 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_after_activation(self):
+        """Test that change of permission group is not possible in active pool"""
+        AbakusGroup.objects.get(name="Bedkom").add_user(self.abakus_user)
+        self.client.force_authenticate(self.abakus_user)
+        pool = Event.objects.get(pk=1).pools.first()
+        new_group = AbakusGroup.objects.get(name="Webkom")
+        pool_response = self.client.patch(
+            _get_pools_detail_url(1, pool.id),
+            {"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_after_activation(self):
+        """Test that change of activation date is not possible in active pool"""
+        AbakusGroup.objects.get(name="Bedkom").add_user(self.abakus_user)
+        self.client.force_authenticate(self.abakus_user)
+        pool = Event.objects.get(pk=1).pools.first()
+        pool_response = self.client.patch(
+            _get_pools_detail_url(1, pool.id),
+            {"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)