fix(dref): update dref approve permission

Author: sudip-khanal

dref/permissions.py

@@ -1,7 +1,7 @@
 from django.contrib.auth.models import Permission
 from rest_framework import permissions
 
-from dref.models import Dref, DrefFinalReport, DrefOperationalUpdate
+from dref.models import DrefFinalReport, DrefOperationalUpdate
 from dref.utils import get_dref_users
 
 
@@ -35,13 +35,24 @@ def has_object_permission(self, request, view, obj):
         return False
 
 
-class PublishDrefPermission(permissions.BasePermission):
-    message = "You need to be regional admin to publish dref"
+class ApproveDrefPermission(permissions.BasePermission):
+    message = "You need to be Superuser or Dref Regional admin to approve"
 
     def has_object_permission(self, request, view, obj):
-        region = obj.country.region.name
-        codename = f"dref_region_admin_{region}"
+
         user = request.user
-        if Permission.objects.filter(user=user, codename=codename).exists() and obj.status != Dref.Status.APPROVED:
+        region_id = obj.country.region_id
+
+        if user.is_superuser:
+            return True
+
+        dref_region_admins_ids = [
+            int(codename.replace("dref_region_admin_", ""))
+            for codename in Permission.objects.filter(
+                group__user=user,
+                codename__startswith="dref_region_admin_",
+            ).values_list("codename", flat=True)
+        ]
+        if region_id in dref_region_admins_ids:
             return True
         return False

dref/test_views.py

@@ -642,66 +642,71 @@ def test_dref_is_approved(self, mock_now):
         - Can only approve when status=FINALIZED
         - Approve sets status=APPROVED
         """
+        management.call_command("make_dref_regional_permission")
+
         initial_now = datetime.now()
         mock_now.return_value = initial_now
 
+        # Users
+        root_user = self.root_user
+        user1 = UserFactory.create(email="dref@test.com")
+
+        # Region and country
         region = Region.objects.create(name=RegionName.AFRICA)
         country = Country.objects.create(name="country1", region=region)
+
+        # Create DREF with DRAFT status
         dref = DrefFactory.create(
             title="test",
-            created_by=self.user,
+            created_by=root_user,
             country=country,
             status=Dref.Status.DRAFT,
             type_of_dref=Dref.DrefType.IMMINENT,
         )
-        # Normal PATCH
+
+        self.client.force_authenticate(root_user)
+        # ---- PATCH updates ----
         url = f"/api/v2/dref/{dref.id}/"
-        data = {
-            "title": "New Update Title",
-            "modified_at": initial_now + timedelta(days=1),
-        }
-        self.client.force_authenticate(self.user)
+        data = {"title": "Updated Title", "modified_at": initial_now + timedelta(days=1)}
         response = self.client.patch(url, data)
         self.assert_200(response)
-        # create new dref with status = DRAFT
-        not_approved_dref = DrefFactory.create(
-            title="test",
-            created_by=self.user,
-            country=country,
-            status=Dref.Status.DRAFT,
-        )
-        url = f"/api/v2/dref/{not_approved_dref.id}/"
-        self.client.force_authenticate(self.user)
-        data["modified_at"] = initial_now + timedelta(days=10)
-        response = self.client.patch(url, data)
-        self.assert_200(response)
-
-        data["modified_at"] = initial_now - timedelta(seconds=10)
-        response = self.client.patch(url, data)
-        self.assert_400(response)
-        # ---- Test publishing ----
-        publish_url = f"/api/v2/dref/{dref.id}/approve/"
-        data = {}
-        response = self.client.post(publish_url, data)
-        self.assert_403(response)
-        # Add permission to user
-        self.dref_permission = Permission.objects.create(
-            codename="dref_region_admin_0",
+        # ---- Add DREF regional admin permission to user1 ----
+        region_permission = Permission.objects.create(
+            codename=f"dref_region_admin_{region.id}",
             content_type=ContentType.objects.get_for_model(Region),
-            name="Dref Admin for 0",
+            name=f"Dref Admin for region {region.id}",
         )
-        self.user.user_permissions.add(self.dref_permission)
-        # Try again while DRAFT. Should fail(not finalized yet)
-        response = self.client.post(publish_url, data)
-        self.assert_400(response)
-        # Update status to FINALIZED, then publish should succeed
+        user1.user_permissions.add(region_permission)
+
+        # ---- Approve while DRAFT as superuser ----
+        approve_url = f"/api/v2/dref/{dref.id}/approve/"
+        self.authenticate(root_user)
+        response = self.client.post(approve_url, {})
+        self.assert_400(response)  # Not finalized yet
+
+        # ---- Update DREF to FINALIZED ----
         dref.status = Dref.Status.FINALIZED
         dref.save(update_fields=["status"])
         dref.refresh_from_db()
-        response = self.client.post(publish_url, data)
+
+        # ---- Approve as superuser ----
+        self.authenticate(root_user)
+        response = self.client.post(approve_url, {})
+        self.assert_200(response)
         dref.refresh_from_db()
+        self.assertEqual(dref.status, Dref.Status.APPROVED)
+
+        # ---- Reset to FINALIZED for regional admin test ----
+        dref.status = Dref.Status.FINALIZED
+        dref.save(update_fields=["status"])
+        dref.refresh_from_db()
+
+        # ---- Approve as regional admin ----
+        self.authenticate(user1)
+        response = self.client.post(approve_url, {})
         self.assert_200(response)
-        self.assertEqual(response.data["status"], Dref.Status.APPROVED)
+        dref.refresh_from_db()
+        self.assertEqual(dref.status, Dref.Status.APPROVED)
 
     def test_dref_operation_update_create(self):
         """
@@ -934,6 +939,8 @@ def test_final_report_update_once_published(self):
         user1 = UserFactory.create()
         region = Region.objects.create(name=RegionName.AFRICA)
         country = Country.objects.create(name="country1", region=region)
+        management.call_command("make_dref_regional_permission")
+        # Create DREF and final report
         dref = DrefFactory.create(
             title="Test Title",
             type_of_dref=Dref.DrefType.ASSESSMENT,
@@ -949,29 +956,25 @@ def test_final_report_update_once_published(self):
             status=Dref.Status.FINALIZED,
         )
         final_report.users.set([user1])
-        url = f"/api/v2/dref-final-report/{final_report.id}/approve/"
-        data = {}
-        self.client.force_authenticate(user1)
-        response = self.client.post(url, data)
-        self.assert_403(response)
-        # add permission to request user
-        self.dref_permission = Permission.objects.create(
-            codename="dref_region_admin_0",
-            content_type=ContentType.objects.get_for_model(Region),
-            name="Dref Admin for 0",
-        )
-        user1.user_permissions.add(self.dref_permission)
+
+        patch_url = f"/api/v2/dref-final-report/{final_report.id}/"
+
+        # ---- Normal PATCH update before approval ----
         self.client.force_authenticate(user1)
-        response = self.client.post(url, data)
-        self.assert_200(response)
-        self.assertEqual(response.data["status"], Dref.Status.APPROVED)
-        # now try to patch to the final report
-        url = f"/api/v2/dref-final-report/{final_report.id}/"
         data = {
-            "title": "New Field Report Title",
+            "title": "Updated Title Before Approval",
+            "modified_at": datetime.now(),
         }
-        response = self.client.patch(url, data)
-        self.assert_400(response)
+        response = self.client.patch(patch_url, data)
+        self.assert_200(response)
+        final_report.refresh_from_db()
+
+        self.client.force_authenticate(self.root_user)
+        approve_url = f"/api/v2/dref-final-report/{final_report.id}/approve/"
+        response = self.client.post(approve_url, {})
+        self.assert_200(response)
+        final_report.refresh_from_db()
+        self.assertEqual(final_report.status, Dref.Status.APPROVED)
 
     def test_dref_for_assessment_report(self):
         old_count = Dref.objects.count()
@@ -1960,7 +1963,7 @@ def test_dref_imminent(self):
         sct_2 = SectorFactory(
             title="health",
         )
-        national_society = Country.objects.create(name="abc")
+        national_society = Country.objects.create(name="abc123")
         disaster_type = DisasterType.objects.create(name="disaster 1")
         data = {
             "title": "Dref test title",

dref/views.py

@@ -30,7 +30,7 @@
     DrefShareUserFilterSet,
 )
 from dref.models import Dref, DrefFile, DrefFinalReport, DrefOperationalUpdate
-from dref.permissions import PublishDrefPermission
+from dref.permissions import ApproveDrefPermission
 from dref.serializers import (
     AddDrefUserSerializer,
     CompletedDrefOperationsSerializer,
@@ -90,14 +90,14 @@ def get_queryset(self):
         detail=True,
         url_path="approve",
         methods=["post"],
-        permission_classes=[permissions.IsAuthenticated, PublishDrefPermission, DenyGuestUserPermission],
+        permission_classes=[permissions.IsAuthenticated, ApproveDrefPermission, DenyGuestUserPermission],
     )
     def get_approved(self, request, pk=None, version=None):
         dref = self.get_object()
+        if dref.status == Dref.Status.APPROVED:
+            raise serializers.ValidationError(gettext("This Dref has already been approved."))
         if dref.status != Dref.Status.FINALIZED:
             raise serializers.ValidationError(gettext("Must be finalized before it can be approved"))
-        if dref.status == Dref.Status.APPROVED:
-            raise serializers.ValidationError(gettext("Dref %s is already approved" % dref))
         dref.status = Dref.Status.APPROVED
         dref.save(update_fields=["status"])
         serializer = DrefSerializer(dref, context={"request": request})
@@ -184,14 +184,16 @@ def get_queryset(self):
         detail=True,
         url_path="approve",
         methods=["post"],
-        permission_classes=[permissions.IsAuthenticated, PublishDrefPermission, DenyGuestUserPermission],
+        permission_classes=[permissions.IsAuthenticated, ApproveDrefPermission, DenyGuestUserPermission],
     )
     def get_approved(self, request, pk=None, version=None):
         operational_update = self.get_object()
+
+        if operational_update.status == Dref.Status.APPROVED:
+            raise serializers.ValidationError(gettext("This Operational update has already been approved."))
         if operational_update.status != Dref.Status.FINALIZED:
             raise serializers.ValidationError(gettext("Must be finalized before it can be approved."))
-        if operational_update.status == Dref.Status.APPROVED:
-            raise serializers.ValidationError(gettext("Operational update %s is already approved" % operational_update))
+
         operational_update.status = Dref.Status.APPROVED
         operational_update.save(update_fields=["status"])
         serializer = DrefOperationalUpdateSerializer(operational_update, context={"request": request})
@@ -247,14 +249,16 @@ def get_queryset(self):
         detail=True,
         url_path="approve",
         methods=["post"],
-        permission_classes=[permissions.IsAuthenticated, PublishDrefPermission, DenyGuestUserPermission],
+        permission_classes=[permissions.IsAuthenticated, ApproveDrefPermission, DenyGuestUserPermission],
     )
     def get_approved(self, request, pk=None, version=None):
         final_report = self.get_object()
+        if final_report.status == Dref.Status.APPROVED:
+            raise serializers.ValidationError(gettext("This Final Report has already been approved."))
+
         if final_report.status != Dref.Status.FINALIZED:
             raise serializers.ValidationError(gettext("Must be finalized before it can be approved."))
-        if final_report.status == Dref.Status.APPROVED:
-            raise serializers.ValidationError(gettext("Final Report %s is already approved" % final_report))
+
         final_report.status = Dref.Status.APPROVED
         final_report.save(update_fields=["status"])
         final_report.dref.is_active = False