DRY Permissions added for assetLocation viewset

care/facility/api/viewsets/asset.py

@@ -4,10 +4,17 @@
 from django_filters import rest_framework as filters
 from drf_yasg.utils import swagger_auto_schema
 from rest_framework import filters as drf_filters
+from dry_rest_permissions.generics import DRYPermissions
+from rest_framework.permissions import IsAuthenticated
 from rest_framework import status
 from rest_framework.decorators import action
 from rest_framework.exceptions import ValidationError
-from rest_framework.mixins import CreateModelMixin, ListModelMixin, RetrieveModelMixin, UpdateModelMixin
+from rest_framework.mixins import (
+    CreateModelMixin,
+    ListModelMixin,
+    RetrieveModelMixin,
+    UpdateModelMixin,
+)
 from rest_framework.response import Response
 from rest_framework.serializers import CharField, JSONField, Serializer, UUIDField
 from rest_framework.viewsets import GenericViewSet
@@ -18,7 +25,12 @@
     AssetTransactionSerializer,
     UserDefaultAssetLocationSerializer,
 )
-from care.facility.models.asset import Asset, AssetLocation, AssetTransaction, UserDefaultAssetLocation
+from care.facility.models.asset import (
+    Asset,
+    AssetLocation,
+    AssetTransaction,
+    UserDefaultAssetLocation,
+)
 from care.users.models import User
 from care.utils.assetintegration.asset_classes import AssetClasses
 from care.utils.cache.cache_allowed_facilities import get_accessible_facilities
@@ -37,7 +49,13 @@ class AssetLocationViewSet(
     UpdateModelMixin,
     GenericViewSet,
 ):
-    queryset = AssetLocation.objects.all().select_related("facility").order_by("-created_date")
+    queryset = (
+        AssetLocation.objects.all().select_related("facility").order_by("-created_date")
+    )
+    permission_classes = (
+        IsAuthenticated,
+        DRYPermissions,
+    )
     serializer_class = AssetLocationSerializer
     lookup_field = "external_id"
     filter_backends = (drf_filters.SearchFilter,)
@@ -56,11 +74,15 @@ def get_queryset(self):
             allowed_facilities = get_accessible_facilities(user)
             queryset = queryset.filter(facility__id__in=allowed_facilities)
 
-        return queryset.filter(facility__external_id=self.kwargs["facility_external_id"])
+        return queryset.filter(
+            facility__external_id=self.kwargs["facility_external_id"]
+        )
 
     def get_facility(self):
         facilities = get_facility_queryset(self.request.user)
-        return get_object_or_404(facilities.filter(external_id=self.kwargs["facility_external_id"]))
+        return get_object_or_404(
+            facilities.filter(external_id=self.kwargs["facility_external_id"])
+        )
 
     def perform_create(self, serializer):
         serializer.save(facility=self.get_facility())
@@ -82,7 +104,9 @@ class AssetViewSet(
     GenericViewSet,
 ):
     queryset = (
-        Asset.objects.all().select_related("current_location", "current_location__facility").order_by("-created_date")
+        Asset.objects.all()
+        .select_related("current_location", "current_location__facility")
+        .order_by("-created_date")
     )
     serializer_class = AssetSerializer
     lookup_field = "external_id"
@@ -98,16 +122,22 @@ def get_queryset(self):
         elif user.user_type >= User.TYPE_VALUE_MAP["StateLabAdmin"]:
             queryset = queryset.filter(current_location__facility__state=user.state)
         elif user.user_type >= User.TYPE_VALUE_MAP["DistrictLabAdmin"]:
-            queryset = queryset.filter(current_location__facility__district=user.district)
+            queryset = queryset.filter(
+                current_location__facility__district=user.district
+            )
         else:
             allowed_facilities = get_accessible_facilities(user)
-            queryset = queryset.filter(current_location__facility__id__in=allowed_facilities)
+            queryset = queryset.filter(
+                current_location__facility__id__in=allowed_facilities
+            )
         return queryset
 
     @swagger_auto_schema(responses={200: UserDefaultAssetLocationSerializer()})
     @action(detail=False, methods=["GET"])
     def get_default_user_location(self, request, *args, **kwargs):
-        obj = get_object_or_404(UserDefaultAssetLocation.objects.filter(user=request.user))
+        obj = get_object_or_404(
+            UserDefaultAssetLocation.objects.filter(user=request.user)
+        )
         return Response(UserDefaultAssetLocationSerializer(obj).data)
 
     class DummyAssetSerializer(Serializer):  # Dummy for Spec
@@ -200,11 +230,13 @@ def get_queryset(self):
             pass
         elif user.user_type >= User.TYPE_VALUE_MAP["StateLabAdmin"]:
             queryset = queryset.filter(
-                Q(from_location__facility__state=user.state) | Q(to_location__facility__state=user.state)
+                Q(from_location__facility__state=user.state)
+                | Q(to_location__facility__state=user.state)
             )
         elif user.user_type >= User.TYPE_VALUE_MAP["DistrictLabAdmin"]:
             queryset = queryset.filter(
-                Q(from_location__facility__district=user.district) | Q(to_location__facility__district=user.district)
+                Q(from_location__facility__district=user.district)
+                | Q(to_location__facility__district=user.district)
             )
         else:
             allowed_facilities = get_accessible_facilities(user)

care/facility/models/asset.py

@@ -6,6 +6,7 @@
 from django.db.models import Q
 
 from care.facility.models.facility import Facility
+from care.facility.models.mixins.permissions.asset import AssetsPermissionMixin
 from care.users.models import User, phone_number_regex
 from care.utils.assetintegration.asset_classes import AssetClasses
 from care.utils.models.base import BaseModel
@@ -15,7 +16,7 @@ def get_random_asset_id():
     return str(uuid.uuid4())
 
 
-class AssetLocation(BaseModel):
+class AssetLocation(BaseModel, AssetsPermissionMixin):
     """
     This model is also used to store rooms that the assets are in, Since these rooms are mapped to
     actual rooms in the hospital, Beds are also connected to this model to remove duplication of efforts
@@ -29,8 +30,12 @@ class RoomType(enum.Enum):
 
     name = models.CharField(max_length=1024, blank=False, null=False)
     description = models.TextField(default="", null=True, blank=True)
-    location_type = models.IntegerField(choices=RoomTypeChoices, default=RoomType.OTHER.value)
-    facility = models.ForeignKey(Facility, on_delete=models.PROTECT, null=False, blank=False)
+    location_type = models.IntegerField(
+        choices=RoomTypeChoices, default=RoomType.OTHER.value
+    )
+    facility = models.ForeignKey(
+        Facility, on_delete=models.PROTECT, null=False, blank=False
+    )
 
 
 class Asset(BaseModel):
@@ -50,10 +55,16 @@ class Status(enum.Enum):
 
     name = models.CharField(max_length=1024, blank=False, null=False)
     description = models.TextField(default="", null=True, blank=True)
-    asset_type = models.IntegerField(choices=AssetTypeChoices, default=AssetType.INTERNAL.value)
-    asset_class = models.CharField(choices=AssetClassChoices, default=None, null=True, blank=True, max_length=20)
+    asset_type = models.IntegerField(
+        choices=AssetTypeChoices, default=AssetType.INTERNAL.value
+    )
+    asset_class = models.CharField(
+        choices=AssetClassChoices, default=None, null=True, blank=True, max_length=20
+    )
     status = models.IntegerField(choices=StatusChoices, default=Status.ACTIVE.value)
-    current_location = models.ForeignKey(AssetLocation, on_delete=models.PROTECT, null=False, blank=False)
+    current_location = models.ForeignKey(
+        AssetLocation, on_delete=models.PROTECT, null=False, blank=False
+    )
     is_working = models.BooleanField(default=None, null=True, blank=True)
     not_working_reason = models.CharField(max_length=1024, blank=True, null=True)
     serial_number = models.CharField(max_length=1024, blank=True, null=True)
@@ -62,14 +73,18 @@ class Status(enum.Enum):
     # Vendor Details
     vendor_name = models.CharField(max_length=1024, blank=True, null=True)
     support_name = models.CharField(max_length=1024, blank=True, null=True)
-    support_phone = models.CharField(max_length=14, validators=[phone_number_regex], default="")
+    support_phone = models.CharField(
+        max_length=14, validators=[phone_number_regex], default=""
+    )
     support_email = models.EmailField(blank=True, null=True)
     qr_code_id = models.CharField(max_length=1024, blank=True, default=None, null=True)
 
     class Meta:
         constraints = [
             models.UniqueConstraint(
-                fields=["qr_code_id"], name="qr_code_unique_when_not_null", condition=Q(qr_code_id__isnull=False)
+                fields=["qr_code_id"],
+                name="qr_code_unique_when_not_null",
+                condition=Q(qr_code_id__isnull=False),
             ),
         ]
 
@@ -79,20 +94,36 @@ def __str__(self):
 
 class UserDefaultAssetLocation(BaseModel):
     user = models.ForeignKey(User, on_delete=models.PROTECT, null=False, blank=False)
-    location = models.ForeignKey(AssetLocation, on_delete=models.PROTECT, null=False, blank=False)
+    location = models.ForeignKey(
+        AssetLocation, on_delete=models.PROTECT, null=False, blank=False
+    )
 
 
 class FacilityDefaultAssetLocation(BaseModel):
-    facility = models.ForeignKey(Facility, on_delete=models.PROTECT, null=False, blank=False)
-    location = models.ForeignKey(AssetLocation, on_delete=models.PROTECT, null=False, blank=False)
+    facility = models.ForeignKey(
+        Facility, on_delete=models.PROTECT, null=False, blank=False
+    )
+    location = models.ForeignKey(
+        AssetLocation, on_delete=models.PROTECT, null=False, blank=False
+    )
 
 
 class AssetTransaction(BaseModel):
     asset = models.ForeignKey(Asset, on_delete=models.PROTECT, null=False, blank=False)
     from_location = models.ForeignKey(
-        AssetLocation, on_delete=models.PROTECT, related_name="from_location", null=False, blank=False,
+        AssetLocation,
+        on_delete=models.PROTECT,
+        related_name="from_location",
+        null=False,
+        blank=False,
     )
     to_location = models.ForeignKey(
-        AssetLocation, on_delete=models.PROTECT, related_name="to_location", null=False, blank=False,
+        AssetLocation,
+        on_delete=models.PROTECT,
+        related_name="to_location",
+        null=False,
+        blank=False,
+    )
+    performed_by = models.ForeignKey(
+        User, on_delete=models.PROTECT, null=False, blank=False
     )
-    performed_by = models.ForeignKey(User, on_delete=models.PROTECT, null=False, blank=False)

care/facility/models/mixins/permissions/asset.py

@@ -0,0 +1,24 @@
+from care.facility.models.mixins.permissions.base import BasePermissionMixin
+from care.users.models import User
+
+
+class AssetsPermissionMixin(BasePermissionMixin):
+    def has_object_read_permission(self, request):
+        return True
+
+    def has_object_write_permission(self, request):
+        if (
+            request.user.user_type == User.TYPE_VALUE_MAP["DistrictReadOnlyAdmin"]
+            or request.user.user_type == User.TYPE_VALUE_MAP["StateReadOnlyAdmin"]
+            or request.user.user_type == User.TYPE_VALUE_MAP["StaffReadOnly"]
+        ):
+            return False
+
+        return True
+
+
+    def has_object_update_permission(self, request):
+        return self.has_object_write_permission(request)
+
+    def has_object_destroy_permission(self, request):
+        return self.has_object_write_permission(request)