Merge pull request #1554 from IFRCGo/feature/dref-superuser-access

Allow superuser to view all dref

Author: szabozoltan69

deployments/snapshots/snap_tests.py

@@ -873,7 +873,7 @@
                 'is_deprecated': False,
                 'iso': 'HA',
                 'iso3': 'xcS',
-                'name': 'country-jYLXlNQGqkURvDMLeoyyigbmHGRAjMglENMcYIGWhfEQiMIaXR',
+                'name': 'country-AMjYLXlNQGqkURvDMLeoyyigbmHGRAjMglENMcYIGWhfEQiMIa',
                 'record_type': 4,
                 'record_type_display': 'Country Office',
                 'region': 3,

dref/migrations/0044_alter_dref_modified_at.py

@@ -0,0 +1,19 @@
+# Generated by Django 3.2.16 on 2022-10-18 07:32
+
+import datetime
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('dref', '0043_delete_dreffileupload'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='dref',
+            name='modified_at',
+            field=models.DateTimeField(blank=True, default=datetime.datetime.now, verbose_name='modified at'),
+        ),
+    ]

dref/migrations/0045_alter_dref_modified_at.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.2.16 on 2022-10-24 06:41
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('dref', '0044_alter_dref_modified_at'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='dref',
+            name='modified_at',
+            field=models.DateTimeField(auto_now=True, verbose_name='modified at'),
+        ),
+    ]

dref/models.py

@@ -1,6 +1,7 @@
 import reversion
 import os
 import copy
+from datetime import datetime
 
 from pdf2image import convert_from_bytes
 
@@ -227,7 +228,11 @@ class Status(models.IntegerChoices):
         COMPLETED = 1, _('Completed')
 
     created_at = models.DateTimeField(verbose_name=_('created at'), auto_now_add=True)
-    modified_at = models.DateTimeField(verbose_name=_('modified at'), auto_now=True)
+    modified_at = models.DateTimeField(
+        verbose_name=_('modified at'),
+        auto_now=True,
+        blank=True
+    )
     created_by = models.ForeignKey(
         settings.AUTH_USER_MODEL, verbose_name=_('created by'), on_delete=models.SET_NULL,
         null=True, related_name='created_by_dref'

dref/serializers.py

@@ -184,6 +184,7 @@ class DrefSerializer(
     ALLOWED_ASSESSMENT_REPORT_EXTENSIONS = ["pdf", "docx", "pptx"]
     MAX_OPERATION_TIMEFRAME = 30
     ASSESSMENT_REPORT_MAX_OPERATION_TIMEFRAME = 2
+    DREF_UPDATE_ERROR_MESSAGE = "OBSOLETE_PAYLOAD"
     national_society_actions = NationalSocietyActionSerializer(many=True, required=False)
     needs_identified = IdentifiedNeedSerializer(many=True, required=False)
     planned_interventions = PlannedInterventionSerializer(many=True, required=False)
@@ -206,6 +207,7 @@ class DrefSerializer(
     assessment_report_details = DrefFileSerializer(source='assessment_report', read_only=True)
     supporting_document_details = DrefFileSerializer(read_only=True, source='supporting_document')
     risk_security = RiskSecuritySerializer(many=True, required=False)
+    modified_at = serializers.DateTimeField(required=False)
 
     class Meta:
         model = Dref
@@ -349,6 +351,9 @@ def create(self, validated_data):
     def update(self, instance, validated_data):
         validated_data['modified_by'] = self.context['request'].user
         is_assessment_report = validated_data.get('is_assessment_report')
+        modified_at = validated_data.pop('modified_at', None)
+        if modified_at is None:
+            raise serializers.ValidationError({ 'modified_at': 'Modified At is required!' })
         if is_assessment_report:
             # Previous Operations
             validated_data['lessons_learned'] = None
@@ -374,12 +379,15 @@ def update(self, instance, validated_data):
             dref_assessment_report = super().update(instance, validated_data)
             dref_assessment_report.needs_identified.clear()
             return dref_assessment_report
+
         # we don't send notification again to the already notified users:
         if 'users' in validated_data:
             to = {u.email for u in validated_data['users']
                   if u.email not in {t.email for t in instance.users.iterator()}}
         else:
             to = None
+        if modified_at and instance.modified_at and modified_at < instance.modified_at:
+            raise serializers.ValidationError({ 'modified_at': self.DREF_UPDATE_ERROR_MESSAGE })
         dref = super().update(instance, validated_data)
         if to:
             transaction.on_commit(

dref/test_views.py

@@ -1,5 +1,6 @@
 import os
 from unittest import mock
+from datetime import datetime, timedelta
 
 
 from django.conf import settings
@@ -355,7 +356,8 @@ def test_update_dref_image(self):
                     "file": file2.id,
                     "caption": "Test Caption"
                 }
-            ]
+            ],
+            "modified_at": datetime.now()
         }
         self.client.force_authenticate(self.user)
         response = self.client.patch(url, data=data, format="json")
@@ -372,7 +374,8 @@ def test_update_dref_image(self):
                     "file": file4.id,
                     "caption": "Test Caption"
                 }
-            ]
+            ],
+            "modified_at": datetime.now()
         }
         self.client.force_authenticate(self.ifrc_user)
         response = self.client.patch(url, data, format='multipart')
@@ -385,7 +388,8 @@ def test_update_dref_image(self):
                     "file": file4.id,
                     "caption": "Test Caption"
                 }
-            ]
+            ],
+            "modified_at": datetime.now()
         }
         self.client.force_authenticate(self.ifrc_user)
         response = self.client.patch(url, data)
@@ -398,7 +402,8 @@ def test_update_dref_image(self):
                     "file": file5.id,
                     "caption": "Test Caption"
                 }
-            ]
+            ],
+            "modified_at": datetime.now()
         }
         self.client.force_authenticate(self.ifrc_user)
         response = self.client.patch(url, data)
@@ -436,31 +441,42 @@ def test_dref_options(self):
         self.assertIn('needs_identified', response.data)
         self.assertIn('national_society_actions', response.data)
 
-    def test_dref_is_published(self):
+    @mock.patch('django.utils.timezone.now')
+    def test_dref_is_published(self, mock_now):
         """
         Test for dref if is_published = True
         """
+        initial_now = datetime(2011, 11, 11)
+        mock_now.return_value = initial_now
+
         dref = DrefFactory.create(
-            title='test', created_by=self.user,
-            is_published=True
+            title='test',
+            created_by=self.user,
+            is_published=True,
         )
         url = f'/api/v2/dref/{dref.id}/'
         data = {
-            "title" : "New Update Title"
+            "title": "New Update Title",
+            "modified_at": initial_now,
         }
         self.client.force_authenticate(self.user)
         response = self.client.patch(url, data)
         self.assert_400(response)
 
         # create new dref with is_published = False
         not_published_dref = DrefFactory.create(
-            title='test', created_by=self.user,
+            title='test',
+            created_by=self.user,
         )
         url = f'/api/v2/dref/{not_published_dref.id}/'
         self.client.force_authenticate(self.user)
         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 dref published endpoint
         url = f'/api/v2/dref/{not_published_dref.id}/publish/'
         data = {}
@@ -795,3 +811,83 @@ def test_dref_for_assessment_report(self):
         response = self.client.post(url, data)
         self.assertEqual(response.status_code, 201)
         self.assertEqual(Dref.objects.count(), old_count + 1)
+
+    def test_dref_for_super_user(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]',
+        )
+        dref1 = DrefFactory.create(
+            title='Test Title',
+            created_by=user2,
+        )
+        DrefFactory.create(
+            title='Test Title New',
+        )
+
+        # authenticate with user1(superuser)
+        # user1 should be able to view all dref
+        url = '/api/v2/dref/'
+        self.client.force_authenticate(user1)
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data['results']), 2)
+
+        # authenticate with User2
+        self.client.force_authenticate(user2)
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data['results']), 1)
+        self.assertEqual(response.data['results'][0]['id'], dref1.id)
+
+        # authenticate with User3
+        self.client.force_authenticate(user3)
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data['results']), 0)
+
+    def test_dref_latest_update(self):
+        dref = DrefFactory.create(
+            title='Test Title',
+            created_by=self.user,
+            modified_at=datetime(2022, 4, 18, 2, 29, 39, 793615)
+        )
+        url = f'/api/v2/dref/{dref.id}/'
+        data = {
+            'title': "New title",
+        }
+
+        # without `modified_at`
+        self.client.force_authenticate(self.user)
+        response = self.client.patch(url, data=data)
+        self.assertEqual(response.status_code, 400)
+
+        # with `modified_at` less than instance `modified_at`
+        data['modified_at'] = datetime(2022, 2, 18, 2, 29, 39, 793615)
+        self.client.force_authenticate(self.user)
+        response = self.client.patch(url, data=data)
+        self.assertEqual(response.status_code, 400)
+
+        data['modified_at'] = datetime.now()
+        response = self.client.patch(url, data=data)
+        self.assertEqual(response.status_code, 200)
+        # Title should be latest since modified_at is greater than modified_at in database
+        self.assertEqual(response.data['title'], "New title")

dref/views.py

@@ -40,17 +40,17 @@ class DrefViewSet(viewsets.ModelViewSet):
     filterset_class = DrefFilter
 
     def get_queryset(self):
-        return Dref.objects\
-            .filter(
-                models.Q(created_by=self.request.user) |
-                models.Q(users=self.request.user)
-            )\
-            .prefetch_related(
-                'planned_interventions',
-                'needs_identified',
-                'national_society_actions',
-                'users'
-            ).order_by('-created_at').distinct()
+        user = self.request.user
+        queryset = Dref.objects.prefetch_related(
+            'planned_interventions',
+            'needs_identified',
+            'national_society_actions',
+            'users'
+        ).order_by('-created_at').distinct()
+        if user.is_superuser:
+            return queryset
+        else:
+            return queryset.filter(models.Q(created_by=user) | models.Q(users=user))
 
     @action(
         detail=True,
@@ -74,10 +74,8 @@ class DrefOperationalUpdateViewSet(viewsets.ModelViewSet):
     filterset_class = DrefOperationalUpdateFilter
 
     def get_queryset(self):
-        return DrefOperationalUpdate.objects.filter(
-            models.Q(created_by=self.request.user) |
-            models.Q(users=self.request.user)
-        ).select_related(
+        user = self.request.user
+        queryset = DrefOperationalUpdate.objects.select_related(
             'national_society',
             'national_society',
             'disaster_type',
@@ -94,6 +92,10 @@ def get_queryset(self):
             'images',
             'photos',
         ).order_by('-created_at').distinct()
+        if user.is_superuser:
+            return queryset
+        else:
+            return queryset.filter(models.Q(created_by=user) | models.Q(users=user))
 
     @action(
         detail=True,
@@ -121,7 +123,6 @@ def get_queryset(self):
             'dref__needs_identified',
         ).order_by('-created_at').distinct()
 
-
     @action(
         detail=True,
         url_path='publish',