Restrict pool updates to active pools

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

@@ -454,24 +454,43 @@ 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()
+                    perm_ids = [getattr(g, "id", g) for g in permission_groups]
+                    pool_instance = None
+                    if pool_id:
+                        pool_instance = (
+                            Pool.objects.filter(id=pool_id, event=instance)
+                            .select_for_update()
+                            .first()
+                        )
+                        existing_ids.discard(pool_id)
+                    pool_data = {
+                        "name": pool.get("name", getattr(pool_instance, "name", None)),
+                        "capacity": pool.get(
+                            "capacity", getattr(pool_instance, "capacity", 0)
+                        ),
+                        "activation_date": pool.get(
+                            "activation_date",
+                            getattr(pool_instance, "activation_date", None),
+                        ),
+                        "permission_groups": perm_ids,
+                    }
+                    ser = PoolCreateAndUpdateSerializer(
+                        instance=pool_instance,
+                        data=pool_data,
+                        context={**self.context, "event": instance},
+                        partial=True,
+                    )
+                    ser.is_valid(raise_exception=True)
+                    ser.save()
+                if existing_ids:
+                    for p in Pool.objects.filter(
+                        event=instance, id__in=existing_ids
+                    ).iterator():
+                        p.delete()
             return super().update(instance, validated_data)
 
 

lego/apps/events/serializers/pools.py

@@ -90,10 +90,29 @@ class Meta:
             "registrations": {"read_only": True},
         }
 
+    def validate(self, attrs):
+        instance = getattr(self, "instance", None)
+        if not instance:
+            return attrs
+        if "permission_groups" in attrs and instance.is_activated:
+            new_ids = {getattr(g, "id", g) for g in attrs["permission_groups"]}
+            old_ids = instance.permission_group_ids()
+            if new_ids != old_ids:
+                raise serializers.ValidationError(
+                    {
+                        "permission_groups": "Permission control is disabled for active pools."
+                    }
+                )
+        if "activation_date" in attrs and instance.is_activated:
+            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

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,23 @@ 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 perform_create(self, serializer):
+        event_id = self.kwargs.get("event_pk")
+        event = get_object_or_404(Event, pk=event_id)
+        serializer.save(event=event)
+
     def destroy(self, request, *args, **kwargs):
         try:
             return super().destroy(request, *args, **kwargs)