Creating crud endpoint for user and facility flag

care/facility/api/serializers/facility_flag.py

@@ -0,0 +1,13 @@
+from rest_framework import serializers
+
+from care.facility.models import Facility, FacilityFlag
+from care.utils.serializers.fields import ExternalIdSerializerField
+
+
+class FacilityFlagSerializer(serializers.ModelSerializer):
+    id = serializers.UUIDField(source="external_id", read_only=True)
+    facility = ExternalIdSerializerField(queryset=Facility.objects.all(), required=True)
+
+    class Meta:
+        model = FacilityFlag
+        exclude = ["external_id", "deleted", "modified_date", "created_date"]

care/facility/api/viewsets/facility_flag.py

@@ -0,0 +1,12 @@
+from rest_framework import viewsets
+
+from care.facility.api.serializers.facility_flag import FacilityFlagSerializer
+from care.facility.models import FacilityFlag
+from care.utils.custom_permissions import IsSuperUser
+
+
+class FacilityFlagViewSet(viewsets.ModelViewSet):
+    queryset = FacilityFlag.objects.all()
+    serializer_class = FacilityFlagSerializer
+    permission_classes = [IsSuperUser]
+    lookup_field = "external_id"

care/facility/tests/test_facility_flags_api.py

@@ -0,0 +1,127 @@
+from rest_framework import status
+from rest_framework.test import APITestCase
+
+from care.utils.registries.feature_flag import FlagRegistry, FlagType
+from care.utils.tests.test_utils import TestUtils
+
+
+class FacilityFlagsViewSetTestCase(TestUtils, APITestCase):
+    @classmethod
+    def setUpTestData(cls):
+        FlagRegistry.register(FlagType.FACILITY, "TEST_FLAG")
+        FlagRegistry.register(FlagType.FACILITY, "TEST_FLAG_2")
+
+        cls.state = cls.create_state()
+        cls.district = cls.create_district(cls.state)
+        cls.local_body = cls.create_local_body(cls.district)
+        cls.super_user = cls.create_super_user("su", cls.district)
+        cls.facility = cls.create_facility(cls.super_user, cls.district, cls.local_body)
+        cls.facility2 = cls.create_facility(
+            cls.super_user, cls.district, cls.local_body
+        )
+        cls.user = cls.create_user("staff", cls.district, home_facility=cls.facility)
+        cls.facility_flag_2 = cls.create_facility_flag("TEST_FLAG_2", cls.facility)
+
+    def setUp(self):
+        self.facility_flag_1 = self.create_facility_flag("TEST_FLAG", self.facility)
+
+    def get_url(self, facility_flag_id=None):
+        base_url = "/api/v1/facility_flags/"
+        if facility_flag_id is not None:
+            base_url += f"{facility_flag_id}/"
+        return base_url
+
+    def test_access_with_non_super_user(self):
+        self.client.force_authenticate(user=self.user)
+        response = self.client.get(self.get_url())
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_access_with_super_user(self):
+        self.client.force_authenticate(user=self.super_user)
+        response = self.client.get(self.get_url())
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+    def test_list_facility_flags(self):
+        self.client.force_authenticate(user=self.super_user)
+        response = self.client.get(self.get_url())
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        data = response.json()
+        self.assertEqual(data["count"], 2)
+
+    def test_create_facility_flag(self):
+        self.client.force_authenticate(user=self.super_user)
+
+        # Attempting to create a duplicate flag
+        response = self.client.post(
+            self.get_url(), {"flag": "TEST_FLAG", "facility": self.facility.external_id}
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+        # Creating a new facility flag
+        response = self.client.post(
+            self.get_url(),
+            {"flag": "TEST_FLAG", "facility": self.facility2.external_id},
+        )
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+
+    def test_retrieve_facility_flag(self):
+        self.client.force_authenticate(user=self.super_user)
+        response = self.client.get(self.get_url(self.facility_flag_1.external_id))
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        data = response.json()
+        self.assertEqual(data["flag"], "TEST_FLAG")
+        self.assertEqual(data["facility"], str(self.facility.external_id))
+
+    def test_update_facility_flag(self):
+        self.client.force_authenticate(user=self.super_user)
+
+        # Confirming original values
+        response = self.client.get(self.get_url(self.facility_flag_1.external_id))
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        data = response.json()
+        self.assertEqual(data["flag"], "TEST_FLAG")
+        self.assertEqual(data["facility"], str(self.facility.external_id))
+
+        # Update the facility flag
+        response = self.client.put(
+            self.get_url(self.facility_flag_1.external_id),
+            {"flag": "TEST_FLAG", "facility": self.facility2.external_id},
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.facility_flag_1.refresh_from_db()
+        self.assertEqual(self.facility_flag_1.flag, "TEST_FLAG")
+        self.assertEqual(
+            self.facility_flag_1.facility.external_id, self.facility2.external_id
+        )
+
+    def test_patch_facility_flag(self):
+        self.client.force_authenticate(user=self.super_user)
+
+        # Confirming original values
+        response = self.client.get(self.get_url(self.facility_flag_1.external_id))
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        data = response.json()
+        self.assertEqual(data["flag"], "TEST_FLAG")
+        self.assertEqual(data["facility"], str(self.facility.external_id))
+
+        # Patch the facility flag
+        response = self.client.patch(
+            self.get_url(self.facility_flag_1.external_id),
+            {"facility": self.facility2.external_id},
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.facility_flag_1.refresh_from_db()
+        self.assertEqual(self.facility_flag_1.flag, "TEST_FLAG")
+        self.assertEqual(
+            self.facility_flag_1.facility.external_id, self.facility2.external_id
+        )
+
+    def test_creating_facility_flag_with_non_existing_flag(self):
+        self.client.force_authenticate(user=self.super_user)
+
+        response = self.client.post(
+            self.get_url(),
+            {"flag": "TEST_FLAG_NON_EXISTING", "facility": self.facility2.external_id},
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(response.json()["detail"], "Flag not registered")

care/users/api/serializers/user_flag.py

@@ -0,0 +1,13 @@
+from rest_framework import serializers
+
+from care.users.models import User, UserFlag
+from care.utils.serializers.fields import ExternalIdSerializerField
+
+
+class UserFlagSerializer(serializers.ModelSerializer):
+    id = serializers.UUIDField(source="external_id", read_only=True)
+    user = ExternalIdSerializerField(queryset=User.objects.all(), required=True)
+
+    class Meta:
+        model = UserFlag
+        exclude = ["external_id", "deleted", "modified_date", "created_date"]

care/users/api/viewsets/user_flag.py

@@ -0,0 +1,12 @@
+from rest_framework import viewsets
+
+from care.users.api.serializers.user_flag import UserFlagSerializer
+from care.users.models import UserFlag
+from care.utils.custom_permissions import IsSuperUser
+
+
+class UserFlagViewSet(viewsets.ModelViewSet):
+    queryset = UserFlag.objects.all()
+    serializer_class = UserFlagSerializer
+    permission_classes = [IsSuperUser]
+    lookup_field = "external_id"

care/users/tests/test_user_flags_api.py

@@ -0,0 +1,120 @@
+from rest_framework import status
+from rest_framework.test import APITestCase
+
+from care.utils.registries.feature_flag import FlagRegistry, FlagType
+from care.utils.tests.test_utils import TestUtils
+
+
+class UserFlagsViewSetTestCase(TestUtils, APITestCase):
+    @classmethod
+    def setUpTestData(cls):
+        FlagRegistry.register(FlagType.USER, "TEST_FLAG")
+        FlagRegistry.register(FlagType.USER, "TEST_FLAG_2")
+
+        cls.state = cls.create_state()
+        cls.district = cls.create_district(cls.state)
+        cls.local_body = cls.create_local_body(cls.district)
+        cls.super_user = cls.create_super_user("su", cls.district)
+        cls.facility = cls.create_facility(cls.super_user, cls.district, cls.local_body)
+        cls.user = cls.create_user("staff", cls.district, home_facility=cls.facility)
+        cls.user_2 = cls.create_user("user2", cls.district, home_facility=cls.facility)
+        cls.user_flag_2 = cls.create_user_flag("TEST_FLAG_2", cls.user_2)
+
+    def setUp(self):
+        self.user_flag_1 = self.create_user_flag("TEST_FLAG", self.user)
+
+    def get_url(self, user_flag_id=None):
+        base_url = "/api/v1/user_flags/"
+        if user_flag_id is not None:
+            base_url += f"{user_flag_id}/"
+        return base_url
+
+    def test_access_with_non_super_user(self):
+        self.client.force_authenticate(user=self.user)
+        response = self.client.get(self.get_url())
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_access_with_super_user(self):
+        self.client.force_authenticate(user=self.super_user)
+        response = self.client.get(self.get_url())
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+    def test_list_user_flags(self):
+        self.client.force_authenticate(user=self.super_user)
+        response = self.client.get(self.get_url())
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        data = response.json()
+        self.assertEqual(data["count"], 2)
+
+    def test_create_user_flag(self):
+        self.client.force_authenticate(user=self.super_user)
+
+        # Attempting to create a duplicate flag
+        response = self.client.post(
+            self.get_url(), {"flag": "TEST_FLAG", "user": self.user.external_id}
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+        # Creating a new user flag
+        response = self.client.post(
+            self.get_url(), {"flag": "TEST_FLAG", "user": self.user_2.external_id}
+        )
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+
+    def test_retrieve_user_flag(self):
+        self.client.force_authenticate(user=self.super_user)
+        response = self.client.get(self.get_url(self.user_flag_1.external_id))
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        data = response.json()
+        self.assertEqual(data["flag"], "TEST_FLAG")
+        self.assertEqual(data["user"], str(self.user.external_id))
+
+    def test_update_user_flag(self):
+        self.client.force_authenticate(user=self.super_user)
+
+        # Confirm original values
+        response = self.client.get(self.get_url(self.user_flag_1.external_id))
+        data = response.json()
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(data["flag"], "TEST_FLAG")
+        self.assertEqual(data["user"], str(self.user.external_id))
+
+        # Update the user flag
+        response = self.client.put(
+            self.get_url(self.user_flag_1.external_id),
+            {"flag": "TEST_FLAG", "user": self.user_2.external_id},
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.user_flag_1.refresh_from_db()
+        self.assertEqual(self.user_flag_1.flag, "TEST_FLAG")
+        self.assertEqual(self.user_flag_1.user.external_id, self.user_2.external_id)
+
+    def test_patch_user_flag(self):
+        self.client.force_authenticate(user=self.super_user)
+
+        # Confirm original values
+        response = self.client.get(self.get_url(self.user_flag_1.external_id))
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        data = response.json()
+        self.assertEqual(data["flag"], "TEST_FLAG")
+        self.assertEqual(data["user"], str(self.user.external_id))
+
+        # Patch the user flag
+        response = self.client.patch(
+            self.get_url(self.user_flag_1.external_id),
+            {"user": self.user_2.external_id},
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.user_flag_1.refresh_from_db()
+        self.assertEqual(self.user_flag_1.flag, "TEST_FLAG")
+        self.assertEqual(self.user_flag_1.user.external_id, self.user_2.external_id)
+
+    def test_creating_user_flag_with_non_existing_flag(self):
+        self.client.force_authenticate(user=self.super_user)
+
+        response = self.client.post(
+            self.get_url(),
+            {"flag": "TEST_FLAG_NON_EXISTING", "user": self.user.external_id},
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(response.json()["detail"], "Flag not registered")

care/utils/custom_permissions.py

@@ -0,0 +1,6 @@
+from rest_framework.permissions import BasePermission
+
+
+class IsSuperUser(BasePermission):
+    def has_permission(self, request, view):
+        return request.user and request.user.is_superuser

care/utils/tests/test_utils.py

@@ -22,6 +22,7 @@
     DiseaseStatusEnum,
     EncounterSymptom,
     Facility,
+    FacilityFlag,
     InvestigationSession,
     InvestigationValue,
     LocalBody,
@@ -51,7 +52,7 @@
     PatientCodeStatusType,
     PatientConsent,
 )
-from care.users.models import District, State
+from care.users.models import District, State, UserFlag
 
 fake = Faker()
 
@@ -728,6 +729,26 @@ def create_prescription(
         data.update(**kwargs)
         return Prescription.objects.create(**data)
 
+    @classmethod
+    def create_facility_flag(
+        cls, flag: str, facility: Facility, **kwargs
+    ) -> FacilityFlag:
+        data = {
+            "facility": facility,
+            "flag": flag,
+        }
+        data.update(**kwargs)
+        return FacilityFlag.objects.create(**data)
+
+    @classmethod
+    def create_user_flag(cls, flag: str, user: User, **kwargs) -> UserFlag:
+        data = {
+            "user": user,
+            "flag": flag,
+        }
+        data.update(**kwargs)
+        return UserFlag.objects.create(**data)
+
     def get_list_representation(self, obj) -> dict:
         """
         Returns the dict representation of the obj in list API

config/api_router.py

@@ -40,6 +40,7 @@
     FacilityViewSet,
 )
 from care.facility.api.viewsets.facility_capacity import FacilityCapacityViewSet
+from care.facility.api.viewsets.facility_flag import FacilityFlagViewSet
 from care.facility.api.viewsets.facility_users import FacilityUserViewSet
 from care.facility.api.viewsets.file_upload import FileUploadViewSet
 from care.facility.api.viewsets.hospital_doctor import HospitalDoctorViewSet
@@ -101,6 +102,7 @@
 )
 from care.users.api.viewsets.plug_config import PlugConfigViewset
 from care.users.api.viewsets.skill import SkillViewSet
+from care.users.api.viewsets.user_flag import UserFlagViewSet
 from care.users.api.viewsets.users import UserViewSet
 from care.users.api.viewsets.userskill import UserSkillViewSet
 
@@ -315,6 +317,9 @@
 
 router.register("medibase", MedibaseViewSet, basename="medibase")
 
+router.register(r"facility_flags", FacilityFlagViewSet, basename="facility-flags")
+router.register(r"user_flags", UserFlagViewSet, basename="user-flags")
+
 # Public endpoints
 router.register("public/asset", AssetPublicViewSet, basename="public-asset")
 router.register("public/asset_qr", AssetPublicQRViewSet, basename="public-asset-qr")