Add basic permission in dref

Author: k9845

api/drf_views.py

@@ -1095,6 +1095,7 @@ class GoHistoricalViewSet(viewsets.ReadOnlyModelViewSet):
     def get_queryset(self):
         return Event.objects.filter(appeals__isnull=False)
 
+
 class CountryOfFieldReportToReviewViewset(viewsets.ReadOnlyModelViewSet):
     queryset = CountryOfFieldReportToReview.objects.order_by('country')
     serializer_class = CountryOfFieldReportToReviewSerializer

dref/permissions.py

@@ -2,20 +2,66 @@
 
 from rest_framework import permissions
 
-from dref.models import Dref
+from dref.models import (
+    Dref,
+    DrefOperationalUpdate,
+)
+
+
+class IsSuperAdmin(permissions.BasePermission):
+    """
+    Allow super user to view and perform all actions
+    """
+    def has_object_permission(self, request, view, obj):
+        if request.method in permissions.SAFE_METHODS:
+            return True
+        return request.user.is_superuser
+
+
+class DrefViewUpdatePermission(IsSuperAdmin):
+    message = "Can view and update Dref"
+
+    def has_permission(self, request, view):
+        if request.method == "POST":
+            return True
+        return True
+
+    def has_object_permission(self, request, view, obj):
+        user = request.user
+        if user.is_superuser:
+            return True
+        if request.method in ["PUT", "DELETE", "PATCH"]:
+            return Dref.objects.filter(
+                models.Q(id=obj.id, users=user) |
+                models.Q(id=obj.id, created_by=user)
+            ).exists()
+        return True
 
 
 class DrefOperationalUpdateCreatePermission(permissions.BasePermission):
     message = "Can create Operational Update for whom dref is shared with"
 
     def has_permission(self, request, view):
-        if request.method != "POST":
-            return True
         user = request.user
         dref = request.data.get('dref')
-        if dref:
-            return Dref.objects.filter(
-                models.Q(id=dref, users=user) |
-                models.Q(id=dref, created_by=user)
+        if request.method == "POST":
+            if user.is_superuser:
+                return True
+            if dref:
+                return Dref.objects.filter(
+                    models.Q(id=dref, users=user) |
+                    models.Q(id=dref, created_by=user),
+                    is_published=True
+                ).exists()
+            return True
+
+    def has_object_permission(self, request, view, obj):
+        user = request.user
+        if user.is_superuser:
+            return True
+        if request.method in ["PUT", "DELETE", "PATCH"]:
+            return DrefOperationalUpdate.objects.filter(
+                models.Q(id=obj.id, users=user) |
+                models.Q(id=obj.id, created_by=user)
             ).exists()
         return True

dref/test_views.py

@@ -986,3 +986,142 @@ def test_optimistic_lock_in_final_report(self):
         data['modified_at'] = datetime.now() + timedelta(days=2)
         response = self.client.patch(url, data=data)
         self.assert_200(response)
+    def test_dref_permission(self):
+        user1 = UserFactory.create(
+            username='[email protected]',
+            first_name='Test',
+            last_name='User1',
+            password='admin123',
+            email='[email protected]',
+            is_superuser=True,
+        )
+        user2 = UserFactory.create(
+            username='[email protected]',
+            first_name='Test',
+            last_name='User2',
+            password='admin123',
+            email='[email protected]',
+        )
+        user3 = UserFactory.create(
+            username='[email protected]',
+            first_name='Test',
+            last_name='User3',
+            password='admin123',
+            email='[email protected]',
+        )
+        user4 = UserFactory.create(
+            username='[email protected]',
+            first_name='Test',
+            last_name='User4',
+            password='admin123',
+            email='[email protected]',
+        )
+        dref1 = DrefFactory.create(
+            title='Test Title',
+            created_by=user1,
+        )
+        dref1.users.add(user2)
+        DrefFactory.create(
+            title='Test Title New',
+            created_by=user3
+        )
+        get_url = '/api/v2/dref/'
+        # authenticate with superuser
+        # should be able to view all drefs
+        self.client.force_authenticate(user1)
+        response = self.client.get(get_url)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data['results']), 2)
+
+        # # let superuser patch the dref
+        patch_url = f'/api/v2/dref/{dref1.id}/'
+        data = {
+            "title": "New test title",
+            "modified_at": datetime.now()
+        }
+        response = self.client.patch(patch_url, data=data)
+        self.assertEqual(response.status_code, 200)
+
+        # # lets authenticate with user for whom dref is shared with
+        self.client.force_authenticate(user2)
+        response = self.client.get(get_url)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data['results']), 1)
+
+        # # try to patch by user
+        self.client.force_authenticate(user2)
+        data = {
+            "title": "New test title",
+            "modified_at": datetime.now() + timedelta(days=1),
+        }
+        response = self.client.patch(patch_url, data=data)
+        self.assertEqual(response.status_code, 200)
+
+        # # try to authenticate with user who is neither assigned nor created_by
+        self.client.force_authenticate(user4)
+        response = self.client.get(get_url)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data['results']), 0)
+
+    def test_dref_operational_update_permission(self):
+        super_user = UserFactory.create(
+            username='[email protected]',
+            first_name='Test',
+            last_name='User1',
+            password='admin123',
+            email='[email protected]',
+            is_superuser=True,
+        )
+        user1, user2 = UserFactory.create_batch(2)
+        dref = DrefFactory.create(
+            title='Test Title',
+            is_published=True,
+        )
+        dref.users.add(user1)
+        self.country1 = Country.objects.create(name='abc')
+        self.district1 = District.objects.create(name='test district1', country=self.country1)
+        old_op_count = DrefOperationalUpdate.objects.filter(dref=dref).count()
+        url = '/api/v2/dref-op-update/'
+        data = {
+            'dref': dref.id,
+            'country': self.country1.id,
+            'district': [self.district1.id],
+        }
+        # authenticate with user for whom dref is not shared
+        self.authenticate(self.user)
+        response = self.client.post(url, data=data)
+        self.assert_403(response)
+        # authenticate with user for whom dref is shared
+        self.authenticate(user1)
+        response = self.client.post(url, data=data)
+        self.assert_201(response)
+        self.assertEqual(
+            DrefOperationalUpdate.objects.filter(dref=dref).count(),
+            old_op_count + 1
+        )
+
+    def test_superuser_permisssion_operational_update(self):
+        super_user = UserFactory.create(
+            username='[email protected]',
+            first_name='Test',
+            last_name='User1',
+            password='admin123',
+            email='[email protected]',
+            is_superuser=True,
+        )
+        dref = DrefFactory.create(
+            title='Test Title',
+            is_published=True,
+        )
+        self.country1 = Country.objects.create(name='abc')
+        self.district1 = District.objects.create(name='test district1', country=self.country1)
+        url = '/api/v2/dref-op-update/'
+        data = {
+            'dref': dref.id,
+            'country': self.country1.id,
+            'district': [self.district1.id],
+        }
+        # authenticate with superuser
+        self.authenticate(super_user)
+        response = self.client.post(url, data=data)
+        self.assert_201(response)

dref/views.py

@@ -32,11 +32,15 @@
     DrefFilter,
     DrefOperationalUpdateFilter
 )
+from dref.permissions import (
+    DrefViewUpdatePermission,
+    DrefOperationalUpdateCreatePermission,
+)
 
 
 class DrefViewSet(RevisionMixin, viewsets.ModelViewSet):
     serializer_class = DrefSerializer
-    permission_classes = [permissions.IsAuthenticated]
+    permission_classes = [permissions.IsAuthenticated, DrefViewUpdatePermission]
     filterset_class = DrefFilter
 
     def get_queryset(self):