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

@@ -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

@@ -798,6 +798,24 @@ def test_event_partial_update_pool_deletion(self):
         self.assertEqual(0, event.can_view_groups.count())
         self.assertEqual(res_event["pools"], [])
 
+    def test_event_update_pool_permission_groups_after_activation(self):
+        """Test updating permission group after pool activation is not allowed"""
+        event = self.client.get(_get_detail_url(self.event_id)).json()
+        permission_group = AbakusGroup.objects.get(name="Abakus").id
+        pool = event["pools"][0]
+        pool["permissionGroups"] = [permission_group]
+        event_update_response = self.client.put(_get_detail_url(self.event_id), event)
+        self.assertEqual(event_update_response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_event_update_pool_activation_date_after_activation(self):
+        """Test updating date after pool activation is not allowed"""
+        event = self.client.get(_get_detail_url(self.event_id)).json()
+        activation_date = timezone.now() + timedelta(days=2)
+        pool = event["pools"][0]
+        pool["activationDate"] = activation_date.isoformat()
+        event_update_response = self.client.put(_get_detail_url(self.event_id), event)
+        self.assertEqual(event_update_response.status_code, status.HTTP_400_BAD_REQUEST)
+
     def test_event_correct_youtube_url(self):
         test_event = _test_event_data[0].copy()
         test_event["youtubeUrl"] = "https://www.youtube.com/watch?v=KrzIaRwAMvc"