Merge pull request #2252 from coronasafe/rithviknishad/fix/disable-as…

…set-create-for-ro-users

Disallow read only users from performing writes to asset

care/facility/api/viewsets/asset.py

@@ -280,7 +280,7 @@ class AssetViewSet(
     lookup_field = "external_id"
     filter_backends = (filters.DjangoFilterBackend, drf_filters.SearchFilter)
     search_fields = ["name", "serial_number", "qr_code_id"]
-    permission_classes = [IsAuthenticated]
+    permission_classes = (IsAuthenticated, DRYPermissions)
     filterset_class = AssetFilter
 
     def get_queryset(self):

care/facility/models/asset.py

@@ -178,6 +178,26 @@ def delete(self, *args, **kwargs):
         AssetBed.objects.filter(asset=self).update(deleted=True)
         super().delete(*args, **kwargs)
 
+    @staticmethod
+    def has_write_permission(request):
+        if request.user.asset or request.user.user_type in User.READ_ONLY_TYPES:
+            return False
+        return (
+            request.user.is_superuser
+            or request.user.verified
+            and request.user.user_type >= User.TYPE_VALUE_MAP["Staff"]
+        )
+
+    def has_object_write_permission(self, request):
+        return self.has_write_permission(request)
+
+    @staticmethod
+    def has_read_permission(request):
+        return request.user.is_superuser or request.user.verified
+
+    def has_object_read_permission(self, request):
+        return self.has_read_permission(request)
+
     def __str__(self):
         return self.name
 

care/facility/tests/test_asset_api.py

@@ -3,6 +3,7 @@
 from rest_framework.test import APITestCase
 
 from care.facility.models import Asset, Bed
+from care.users.models import User
 from care.utils.assetintegration.asset_classes import AssetClasses
 from care.utils.tests.test_utils import TestUtils
 
@@ -17,6 +18,11 @@ def setUpTestData(cls) -> None:
         cls.facility = cls.create_facility(cls.super_user, cls.district, cls.local_body)
         cls.asset_location = cls.create_asset_location(cls.facility)
         cls.user = cls.create_user("staff", cls.district, home_facility=cls.facility)
+        cls.state_admin_ro = cls.create_user(
+            "stateadmin-ro",
+            cls.district,
+            user_type=User.TYPE_VALUE_MAP["StateReadOnlyAdmin"],
+        )
         cls.patient = cls.create_patient(
             cls.district, cls.facility, local_body=cls.local_body
         )
@@ -38,6 +44,16 @@ def test_create_asset(self):
         response = self.client.post("/api/v1/asset/", sample_data)
         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
 
+    def test_create_asset_read_only(self):
+        sample_data = {
+            "name": "Test Asset",
+            "asset_type": 50,
+            "location": self.asset_location.external_id,
+        }
+        self.client.force_authenticate(self.state_admin_ro)
+        response = self.client.post("/api/v1/asset/", sample_data)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
     def test_create_asset_with_warranty_past(self):
         sample_data = {
             "name": "Test Asset",