Refactor permission

Author: k9845

dref/models.py

@@ -662,6 +662,44 @@ def save(self, *args, **kwargs):
     def __str__(self):
         return f'{self.title} – {self.created_at.date()}, {self.get_status_display()}'
 
+    @staticmethod
+    def get_for(user):
+        user_id = user.id
+        print(Dref.objects.annotate(
+            created_user_list=models.F('created_by'),
+            users_list=models.F('users'),
+            op_users=models.Subquery(
+                DrefOperationalUpdate.objects.filter(
+                    dref=models.OuterRef('id')
+                ).order_by().values('id').annotate(
+                    c=models.F('users')
+                ).values('c')[:1]
+            ),
+        ).values("created_user_list", "users_list", "op_users"))
+        return Dref.objects.annotate(
+            created_user_list=models.F('created_by'),
+            users_list=models.F('users'),
+            op_users=models.Subquery(
+                DrefOperationalUpdate.objects.filter(
+                    dref=models.OuterRef('id')
+                ).order_by().values('id').annotate(
+                    c=models.F('users')
+                ).values('c')[:1]
+            ),
+            fr_users=models.Subquery(
+                DrefFinalReport.objects.filter(
+                    dref=models.OuterRef('id')
+                ).order_by().values('id').annotate(
+                    c=models.F('users')
+                ).values('c')[:1],
+            )
+        ).filter(
+            models.Q(created_user_list=user_id) |
+            models.Q(users_list=user_id) |
+            models.Q(op_users=user_id) |
+            models.Q(fr_users=user_id)
+        ).distinct()
+
 
 class DrefFile(models.Model):
     file = models.FileField(
@@ -1175,6 +1213,25 @@ class Meta:
         verbose_name = _('Dref Operational Update')
         verbose_name_plural = _('Dref Operational Updates')
 
+    @staticmethod
+    def get_for(user):
+        user_id = user.id
+        return DrefOperationalUpdate.objects.annotate(
+            created_user_list=models.F('created_by'),
+            users_list=models.F('users'),
+            dref_users=models.Subquery(
+                Dref.objects.filter(
+                    drefoperationalupdate=models.OuterRef('id')
+                ).order_by().values('drefoperationalupdate').annotate(
+                    c=models.F('users')
+                ).values('c')[:1]
+            )
+        ).filter(
+            models.Q(created_user_list=user_id) |
+            models.Q(users_list=user_id) |
+            models.Q(dref_users=user_id)
+        ).distinct()
+
 
 class DrefFinalReport(models.Model):
     created_at = models.DateTimeField(verbose_name=_('created at'), auto_now_add=True)

dref/permissions.py

@@ -6,6 +6,7 @@
     Dref,
     DrefOperationalUpdate,
 )
+from dref.utils import get_users_in_dref, get_users_in_dref_operational_update
 
 
 class IsSuperAdmin(permissions.BasePermission):
@@ -18,50 +19,16 @@ def has_object_permission(self, request, view, obj):
         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 DrefViewUpdatePermission(permissions.BasePermission):
+    message = "Can't view and update Dref"
+    pass
 
 
 class DrefOperationalUpdateCreatePermission(permissions.BasePermission):
     message = "Can create Operational Update for whom dref is shared with"
 
     def has_permission(self, request, view):
-        user = request.user
-        dref = request.data.get('dref')
-        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
+        pass
 
     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
+        pass

dref/test_views.py

@@ -746,7 +746,7 @@ def test_final_report_update_once_published(self):
         # try to publish this report
         url = f'/api/v2/dref-final-report/{final_report.id}/publish/'
         data = {}
-        self.client.force_authenticate(self.user)
+        self.client.force_authenticate(user1)
         response = self.client.post(url, data)
         self.assert_200(response)
         self.assertEqual(response.data['is_published'], True)
@@ -986,6 +986,7 @@ 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]',
@@ -1063,7 +1064,7 @@ def test_dref_permission(self):
         self.assertEqual(response.status_code, 200)
         self.assertEqual(len(response.data['results']), 0)
 
-    def test_dref_operational_update_permission(self):
+    def test_superuser_permisssion_operational_update(self):
         super_user = UserFactory.create(
             username='[email protected]',
             first_name='Test',
@@ -1072,56 +1073,105 @@ def test_dref_operational_update_permission(self):
             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)
+        # authenticate with superuser
+        self.authenticate(super_user)
         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,
+    def test_operational_update_view_permission(self):
+        Dref.objects.all().delete()
+        user1, user2, user3 = UserFactory.create_batch(3)
+        dref = DrefFactory.create(
+            title='Test Title',
+            is_published=True,
+            created_by=user1
         )
+        operational_update = DrefOperationalUpdateFactory.create(
+            dref=dref,
+        )
+        operational_update.users.set([user3])
+        self.country1 = Country.objects.create(name='abc')
+        self.district1 = District.objects.create(name='test district1', country=self.country1)
+
+        # Here user1 is able to see the operational_update
+        url = '/api/v2/dref-op-update/'
+        self.authenticate(user1)
+        response = self.client.get(url)
+        self.assert_200(response)
+
+        # authenticate with the user3
+        # here user3 should be able to view dref
+        # since user3 is assigned to op-update created from dref
+        dref_url = '/api/v2/dref/'
+        self.authenticate(user3)
+        response = self.client.get(dref_url)
+        self.assert_200(response)
+        self.assertEqual(len(response.data['results']), 1)
+        self.assertEqual(response.data['results'][0]['id'], dref.id)
+
+        # authenticate with user2
+        # here user2 is not able to view dref
+        self.authenticate(user2)
+        response = self.client.get(dref_url)
+        self.assert_200(response)
+        self.assertEqual(len(response.data['results']), 0)
+
+    def test_concurrent_dref_operational_update(self):
+        user1, user2, user3 = UserFactory.create_batch(3)
+        print(f'{user1.id} {user2.id} {user3.id}')
         dref = DrefFactory.create(
             title='Test Title',
             is_published=True,
+            created_by=user1
+        )
+        operational_update = DrefOperationalUpdateFactory.create(
+            dref=dref,
+            operational_update_number=1,
+            created_by=user3
         )
+        operational_update.users.set([user3, user2])
         self.country1 = Country.objects.create(name='abc')
         self.district1 = District.objects.create(name='test district1', country=self.country1)
+
+        # Here user1 is able to see the operational_update
         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)
+        self.authenticate(user1)
+        response = self.client.get(url)
+        self.assert_200(response)
+
+        # create another operational update from corresponding dref
+        operation_update2 =DrefOperationalUpdateFactory.create(
+            dref=dref,
+            operational_update_number=2
+        )
+        operation_update2.users.set([user2])
+
+        # authenticate with user2
+        url = '/api/v2/dref-op-update/'
+        self.authenticate(user2)
+        response = self.client.get(url)
+        self.assert_200(response)
+
+        # user2 should also be able to view dref
+        dref_url = '/api/v2/dref/'
+        self.authenticate(user2)
+        response = self.client.get(dref_url)
+        self.assert_200(response)
+        self.assertEqual(len(response.data['results']), 1)
+
+        # authenticate with user1
+        self.authenticate(user1)
+        response = self.client.get(dref_url)
+        self.assert_200(response)

dref/utils.py

@@ -1,5 +1,13 @@
 from django.conf import settings
 
+from dref.models import (
+    Dref,
+    DrefOperationalUpdate,
+    DrefFinalReport,
+)
+from django.db import models
+from django.contrib.postgres.aggregates import ArrayAgg
+
 
 def get_email_context(instance):
     from dref.serializers import DrefSerializer
@@ -11,3 +19,52 @@ def get_email_context(instance):
         'frontend_url': settings.FRONTEND_URL,
     }
     return email_context
+
+
+# get the list fo users in dref
+def get_users_in_dref():
+    data = Dref.objects.values('id').annotate(created_user_list=ArrayAgg(
+        'created_by', filter=models.Q(created_by__isnull=False))
+    ).annotate(
+        users_list=ArrayAgg('users', filter=models.Q(users__isnull=False))
+    ).annotate(
+        op_created_by=models.Subquery(
+            DrefOperationalUpdate.objects.filter(
+                dref=models.OuterRef('id')
+            ).order_by().values('id').annotate(
+                c=ArrayAgg('created_by', filter=models.Q(created_by__isnull=False))
+            ).values('c')[:1]
+        ),
+        op_users=models.Subquery(
+            DrefOperationalUpdate.objects.filter(
+                dref=models.OuterRef('id')
+            ).order_by().values('id').annotate(
+                c=ArrayAgg('users', filter=models.Q(users__isnull=False))
+            ).values('c')[:1]
+        )
+    )
+    return data
+
+
+# get the list fo users in dref
+def get_users_in_dref_operational_update():
+    data = DrefOperationalUpdate.objects.values('id').annotate(
+        created_user_list=ArrayAgg(
+            'created_by', filter=models.Q(users__isnull=False)
+        ),
+        users=ArrayAgg(
+            'users', filter=models.Q(users__isnull=False)
+        )
+    )
+    share_with_users = DrefOperationalUpdate.objects.values_list('users', flat=True)
+    created_by_users = DrefOperationalUpdate.objects.values_list('created_by', flat=True)
+    users = share_with_users.union(created_by_users)
+    return users
+
+
+# get the list fo users in dref
+def get_users_in_dref_final_report():
+    share_with_users = DrefFinalReport.objects.values_list('users', flat=True)
+    created_by_users = DrefFinalReport.objects.values_list('created_by', flat=True)
+    users = share_with_users.union(created_by_users)
+    return users