WIP✨(backend) add bulk delete route

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to
 
 ## Added
 
+- Add admin endpoint to bulk delete objects
 - Remove djmoney and Moneyfields
 - Add admin endpoints to create/update/delete organization/course accesses
 - Add admin endpoint to search users

src/backend/joanie/core/api/admin.py

@@ -1,14 +1,16 @@
 """
 Admin API Endpoints
 """
+
 import django_filters.rest_framework
 from rest_framework import authentication, mixins, permissions, viewsets
 
 from joanie.core import filters, models, serializers
+from joanie.core.mixins import admin as admin_mixins
 
 
 # pylint: disable=too-many-ancestors
-class OrganizationViewSet(viewsets.ModelViewSet):
+class OrganizationViewSet(viewsets.ModelViewSet, admin_mixins.BulkDeleteMixin):
     """
     Admin Organization ViewSet
     """
@@ -29,7 +31,7 @@ def get_serializer_class(self):
         return self.serializer_class
 
 
-class ProductViewSet(viewsets.ModelViewSet):
+class ProductViewSet(viewsets.ModelViewSet, admin_mixins.BulkDeleteMixin):
     """
     Admin Product ViewSet
     """
@@ -41,7 +43,7 @@ class ProductViewSet(viewsets.ModelViewSet):
     filterset_class = filters.ProductAdminFilterSet
 
 
-class CourseViewSet(viewsets.ModelViewSet):
+class CourseViewSet(viewsets.ModelViewSet, admin_mixins.BulkDeleteMixin):
     """
     Admin Course ViewSet
     """
@@ -62,7 +64,7 @@ def get_serializer_class(self):
         return self.serializer_class
 
 
-class CourseRunViewSet(viewsets.ModelViewSet):
+class CourseRunViewSet(viewsets.ModelViewSet, admin_mixins.BulkDeleteMixin):
     """
     Admin CourseRun ViewSet
     """
@@ -75,7 +77,7 @@ class CourseRunViewSet(viewsets.ModelViewSet):
     filterset_class = filters.CourseRunAdminFilterSet
 
 
-class CertificateDefinitionViewSet(viewsets.ModelViewSet):
+class CertificateDefinitionViewSet(viewsets.ModelViewSet, admin_mixins.BulkDeleteMixin):
     """
     Admin Certificate ViewSet
     """

src/backend/joanie/core/mixins/__init__.py

@@ -0,0 +1,5 @@
+"""Mixins for Joanie Core app."""
+# flake8: noqa
+# pylint: disable=wildcard-import
+
+from .admin import *

src/backend/joanie/core/mixins/admin.py

@@ -0,0 +1,70 @@
+"""
+Mixins for admin api
+"""
+
+from django.db import IntegrityError
+from django.utils.translation import gettext_lazy as _
+
+from rest_framework import mixins
+from rest_framework.response import Response
+
+
+class BulkDeleteMixin(mixins.DestroyModelMixin):
+    """
+    Mixin for classes that need to be able to bulk delete
+    """
+
+    def delete(self, request, *args, **kwargs):
+        """
+        Performs a deletion. Handles cases where an id is present
+        in the url (single deletion), if no id is present in url
+        and ids are in the body, performs bulk delete
+        """
+
+        if self.kwargs.get("id", None):
+            return super().delete(request, *args, **kwargs)
+
+        user = request.user
+        id_to_delete = request.data.get("id", [])
+        elements_to_delete = self.queryset.filter(id__in=id_to_delete)
+        (deleted_elements, error_elements) = self.perform_bulk_delete(
+            elements_to_delete, user
+        )
+
+        response = {}
+        if len(deleted_elements) > 0:
+            response["deleted"] = deleted_elements
+        if len(error_elements) > 0:
+            response["error"] = error_elements
+
+        return Response(response)
+
+    def perform_bulk_delete(self, instances, user):
+        """
+        Bulk deletion of model
+        data = {"id": [<id_0>, <id_1>]}
+        """
+        error_elements = []
+        deleted_elements = []
+        for element in instances.all():
+            if element.get_abilities(user)["delete"]:
+                try:
+                    deleted_element_id = element.id
+                    element.delete()
+                    deleted_elements.append(deleted_element_id)
+                except IntegrityError as error:
+                    error_elements.append(
+                        {
+                            "id": element.id,
+                            "error": str(error),
+                        }
+                    )
+            else:
+                error_elements.append(
+                    {
+                        "id": element.id,
+                        "error": _("Not authorized to delete this element"),
+                    }
+                )
+
+        return (deleted_elements, error_elements)

src/backend/joanie/core/models/certifications.py

@@ -45,6 +45,21 @@ class Meta:
 
     def __str__(self):
         return self.safe_translation_getter("title", any_language=True)
+
+    def get_abilities(self, user):
+        """
+        Compute and return abilities for a given user taking into account
+        the current state of the object.
+        """
+
+        # TODO : Who can delete a CertificateDefinition?
+        return {
+            "delete": True,
+            "get": True,
+            "patch": True,
+            "put": True,
+            "set_role_to": True,
+        }
 
 
 class Certificate(BaseModel):

src/backend/joanie/core/models/courses.py

@@ -613,6 +613,21 @@ class CourseRun(parler_models.TranslatableModel, BaseModel):
         ),
     )
 
+    def get_abilities(self, user):
+        """
+        Compute and return abilities for a given user taking into account
+        the current state of the object.
+        """
+
+        # TODO : Who can delete a CourseRun?
+        return {
+            "delete": True,
+            "get": True,
+            "patch": True,
+            "put": True,
+            "set_role_to": True,
+        }
+
     class Meta:
         db_table = "joanie_course_run"
         verbose_name = _("Course run")

src/backend/joanie/core/models/products.py

@@ -230,6 +230,21 @@ def clean(self):
             )
         super().clean()
 
+    def get_abilities(self, user):
+        """
+        Compute and return abilities for a given user taking into account
+        the current state of the object.
+        """
+
+        # TODO : Who can delete a Product?
+        return {
+            "delete": True,
+            "get": True,
+            "patch": True,
+            "put": True,
+            "set_role_to": True,
+        }
+
 
 class ProductTargetCourseRelation(BaseModel):
     """

src/backend/joanie/tests/core/test_api_admin_certificate_definitions.py

@@ -5,7 +5,7 @@
 
 from django.test import TestCase
 
-from joanie.core import factories
+from joanie.core import factories, models
 
 
 class CertificateDefinitionAdminApiTest(TestCase):
@@ -250,3 +250,63 @@ def test_admin_api_certificate_definition_delete(self):
         )
 
         self.assertEqual(response.status_code, 204)
+
+    def test_admin_api_certificate_definition_bulk_delete(self):
+        """
+        Staff user should be able to delete multiple certificate_definitions.
+        """
+        admin = factories.UserFactory(is_staff=True, is_superuser=True)
+        self.client.login(username=admin.username, password="password")
+        certificate_definitions = factories.CertificateDefinitionFactory.create_batch(3)
+        factories.CertificateDefinitionFactory()
+        data = {
+            "id": [
+                str(certificate_definition.id)
+                for certificate_definition in certificate_definitions
+            ]
+        }
+        response = self.client.delete(
+            "/api/v1.0/admin/certificate-definitions/",
+            data=data,
+            content_type="application/json",
+        )
+        content = response.json()
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(models.CertificateDefinition.objects.count(), 1)
+        self.assertEqual(len(content["deleted"]), 3)
+        self.assertFalse("error" in content)
+
+    def test_admin_api_certificate_definition_bulk_delete_error(self):
+        """
+        CertificateDefinition linked to a Certificate cannot be deleted.
+        """
+        admin = factories.UserFactory(is_staff=True, is_superuser=True)
+        self.client.login(username=admin.username, password="password")
+        certificate_definitions = factories.CertificateDefinitionFactory.create_batch(2)
+        factories.CertificateFactory(certificate_definition=certificate_definitions[0])
+        data = {
+            "id": [
+                str(certificate_definition.id)
+                for certificate_definition in certificate_definitions
+            ]
+        }
+        response = self.client.delete(
+            "/api/v1.0/admin/certificate-definitions/",
+            data=data,
+            content_type="application/json",
+        )
+        content = response.json()
+        self.assertEqual(response.status_code, 200)
+        self.assertTrue(
+            models.CertificateDefinition.objects.filter(
+                id=certificate_definitions[0].id
+            ).exists()
+        )
+        self.assertFalse(
+            models.CertificateDefinition.objects.filter(
+                id=certificate_definitions[1].id
+            ).exists()
+        )
+        self.assertEqual(len(content["deleted"]), 1)
+        self.assertEqual(content["deleted"][0], str(certificate_definitions[1].id))
+        self.assertEqual(len(content["error"]), 1)

src/backend/joanie/tests/core/test_api_admin_course_runs.py

@@ -5,7 +5,7 @@
 
 from django.test import TestCase
 
-from joanie.core import factories
+from joanie.core import factories, models
 
 
 class CourseRunAdminApiTest(TestCase):
@@ -233,3 +233,21 @@ def test_admin_api_course_runs_delete(self):
         response = self.client.delete(f"/api/v1.0/admin/course-runs/{course_run.id}/")
 
         self.assertEqual(response.status_code, 204)
+
+    def test_admin_api_course_runs_bulk_delete(self):
+        """
+        Staff user should be able to delete multiple course runs.
+        """
+        admin = factories.UserFactory(is_staff=True, is_superuser=True)
+        self.client.login(username=admin.username, password="password")
+        course_runs = factories.CourseRunFactory.create_batch(3)
+        factories.CourseRunFactory()
+        data = {"id": [str(course_run.id) for course_run in course_runs]}
+        response = self.client.delete(
+            "/api/v1.0/admin/course-runs/", data=data, content_type="application/json"
+        )
+        content = response.json()
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(models.CourseRun.objects.count(), 1)
+        self.assertEqual(len(content["deleted"]), 3)
+        self.assertFalse("error" in content)

src/backend/joanie/tests/core/test_api_admin_courses.py

@@ -5,7 +5,7 @@
 
 from django.test import TestCase
 
-from joanie.core import factories
+from joanie.core import factories, models
 
 
 class CourseAdminApiTest(TestCase):
@@ -309,3 +309,64 @@ def test_admin_api_course_delete(self):
         response = self.client.delete(f"/api/v1.0/admin/courses/{course.id}/")
 
         self.assertEqual(response.status_code, 204)
+
+    def test_admin_api_course_bulk_delete(self):
+        """
+        Staff user should be able to delete multiple courses.
+        """
+        admin = factories.UserFactory(is_staff=True, is_superuser=True)
+        self.client.login(username=admin.username, password="password")
+        courses = factories.CourseFactory.create_batch(3)
+        for course in courses:
+            factories.UserCourseAccessFactory(user=admin, course=course, role="owner")
+        factories.CourseFactory()
+        data = {"id": [str(course.id) for course in courses]}
+        response = self.client.delete(
+            "/api/v1.0/admin/courses/", data=data, content_type="application/json"
+        )
+        content = response.json()
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(models.Course.objects.count(), 1)
+        self.assertEqual(len(content["deleted"]), 3)
+        self.assertFalse("error" in content)
+
+    def test_admin_api_course_bulk_delete_no_authorization(self):
+        """
+        Unauthorized user should be able to delete multiple courses.
+        """
+        admin = factories.UserFactory(is_staff=True, is_superuser=True)
+        self.client.login(username=admin.username, password="password")
+        courses = factories.CourseFactory.create_batch(3)
+        factories.CourseFactory()
+        data = {"id": [str(course.id) for course in courses]}
+        response = self.client.delete(
+            "/api/v1.0/admin/courses/", data=data, content_type="application/json"
+        )
+        content = response.json()
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(models.Course.objects.count(), 4)
+        self.assertEqual(len(content["error"]), 3)
+
+    def test_admin_api_course_bulk_delete_error(self):
+        """
+        Courses with linked CourseRun cannot be deleted.
+        """
+        admin = factories.UserFactory(is_staff=True, is_superuser=True)
+        self.client.login(username=admin.username, password="password")
+        courses = factories.CourseFactory.create_batch(3)
+        factories.CourseRunFactory(course=courses[0])
+        for course in courses:
+            factories.UserCourseAccessFactory(user=admin, course=course, role="owner")
+        data = {"id": [str(course.id) for course in courses]}
+        response = self.client.delete(
+            "/api/v1.0/admin/courses/", data=data, content_type="application/json"
+        )
+        content = response.json()
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(models.Course.objects.count(), 1)
+        self.assertEqual(len(content["deleted"]), 2)
+        self.assertEqual(len(content["error"]), 1)
+        self.assertEqual(content["error"][0]["id"], str(courses[0].id))
+        self.assertTrue(
+            "through protected foreign keys" in content["error"][0]["error"]
+        )