Refactor bulk upload logic for XLSX file processing

local_units/models.py

@@ -203,6 +203,7 @@ class HealthData(models.Model):
         null=True,
         blank=True,
     )
+
     ambulance_type_a = models.IntegerField(verbose_name=_("Ambulance Type A"), blank=True, null=True)
     ambulance_type_b = models.IntegerField(verbose_name=_("Ambulance Type B"), blank=True, null=True)
     ambulance_type_c = models.IntegerField(verbose_name=_("Ambulance Type C"), blank=True, null=True)

local_units/permissions.py

@@ -1,7 +1,7 @@
 from django.contrib.auth.models import Permission
 from rest_framework import permissions
 
-from api.models import Country
+from api.models import Country, Profile
 from local_units.models import LocalUnitType
 from local_units.utils import (
     get_local_unit_country_validators,
@@ -25,20 +25,46 @@ def has_object_permission(self, request, view, obj):
 
 
 class IsAuthenticatedForLocalUnit(permissions.BasePermission):
-    message = "Only Country Admins, Local Unit Validators, Region Admins, or Superusers are allowed to update Local Units."
+    message = (
+        "Only users with the correct organization type and country, "
+        "Region or Country Admins, Local Unit Validators, IFRC Admins, or Superusers "
+        "can update Local Units."
+    )
+
+    def user_has_permission(self, user_profile, obj) -> bool:
+        """NOTE:
+        Requirement:
+            Users whose profile is associated with the organization types NTLS, DLGN, or SCRT
+            should be able to update Local Units assigned to their country.
+
+        Purpose:
+            This permission enforces the requirement that users with specific organization types
+            (NTLS, DLGN, or SCRT) can update Local Units only within their assigned country.
+            Implementing it here avoids creating multiple permission groups for each organization
+            type and assigning them through the admin panel.
+        """
+        return (
+            user_profile.org_type
+            in [
+                Profile.OrgTypes.NTLS,
+                Profile.OrgTypes.DLGN,
+                Profile.OrgTypes.SCRT,
+            ]
+            and obj.country_id == user_profile.country_id
+        )
 
     def has_object_permission(self, request, view, obj):
         if request.method not in ["PUT", "PATCH"]:
             return True  # Only restrict update operations
 
         user = request.user
 
-        if user.is_superuser:
+        # IFRC Admin, Superuser, or org-type permission
+        if user.has_perm("api.ifrc_admin") or user.is_superuser or self.user_has_permission(user.profile, obj):
             return True
 
         country_id = obj.country_id
-        country = Country.objects.get(id=country_id)
-        region_id = country.region_id
+        region_id = obj.country.region_id
         # Country admin specific permissions
         country_admin_ids = [
             int(codename.replace("country_admin_", ""))

local_units/serializers.py

@@ -673,6 +673,7 @@ def update(self, instance, validated_data):
 
 
 class LocalUnitBulkUploadSerializer(serializers.ModelSerializer):
+    VALID_FILE_EXTENSIONS = (".xlsx", ".xlsm")
     country = serializers.PrimaryKeyRelatedField(
         queryset=Country.objects.filter(
             is_deprecated=False, independent=True, iso3__isnull=False, record_type=CountryType.COUNTRY
@@ -703,8 +704,8 @@ class Meta:
         )
 
     def validate_file(self, file):
-        if not file.name.endswith(".csv"):
-            raise serializers.ValidationError(gettext("File must be a CSV file."))
+        if not file.name.lower().endswith(self.VALID_FILE_EXTENSIONS):
+            raise serializers.ValidationError(gettext("The uploaded file must be an Excel document (.xlsx or .xlsm)."))
         if file.size > 10 * 1024 * 1024:
             raise serializers.ValidationError(gettext("File must be less than 10 MB."))
         return file
@@ -871,7 +872,7 @@ class LocalUnitBulkUploadDetailSerializer(serializers.ModelSerializer):
     visibility = serializers.CharField(required=True, allow_blank=True)
     date_of_data = serializers.CharField(required=False, allow_null=True)
     level = serializers.CharField(required=False, allow_null=True)
-    health = HealthDataBulkUploadSerializer(required=False)
+    health = serializers.PrimaryKeyRelatedField(queryset=HealthData.objects.all(), required=False, allow_null=True)
 
     class Meta:
         model = LocalUnit
@@ -952,10 +953,6 @@ def validate(self, validated_data):
         validated_data["status"] = LocalUnit.Status.EXTERNALLY_MANAGED
 
         # NOTE: Bulk upload doesn't call create() method
-        health_data = validated_data.pop("health", None)
-        if health_data:
-            health_instance = HealthData.objects.create(**health_data)
-            validated_data["health"] = health_instance
         return validated_data
 
 

local_units/test_views.py

@@ -12,7 +12,7 @@
 
 from api.factories.country import CountryFactory
 from api.factories.region import RegionFactory
-from api.models import Country, CountryGeoms, CountryType, Region
+from api.models import Country, CountryGeoms, CountryType, Profile, Region
 from deployments.factories.user import UserFactory
 from local_units.bulk_upload import BaseBulkUploadLocalUnit, BulkUploadHealthData
 from main import settings
@@ -120,7 +120,6 @@ def setUp(self):
         country_codename = f"local_unit_country_validator_{self.type_3.id}_{self.country2.id}"
         region_codename = f"local_unit_region_validator_{self.type_3.id}_{region.id}"
         global_codename = f"local_unit_global_validator_{self.type_3.id}"
-
         country_permission = Permission.objects.get(codename=country_codename)
         region_permission = Permission.objects.get(codename=region_codename)
         global_permission = Permission.objects.get(codename=global_codename)
@@ -882,7 +881,7 @@ def test_create_local_unit_with_externally_managed_country_and_type(self):
         self.assertEqual(response.status_code, 400)
         self.assertEqual(LocalUnitChangeRequest.objects.count(), 0)
 
-    def test_country_region_admin_permission_for_local_unit_update(self):
+    def test_local_unit_update(self):
         region1 = RegionFactory.create(name=2, label="Asia Pacific")
 
         region2 = RegionFactory.create(name=0, label="Africa")
@@ -895,9 +894,19 @@ def test_country_region_admin_permission_for_local_unit_update(self):
             independent=True,
             region=region1,
         )
+        country2 = CountryFactory.create(
+            name="Nepal",
+            iso3="NEP",
+            record_type=CountryType.COUNTRY,
+            is_deprecated=False,
+            independent=True,
+            region=region1,
+        )
         self.asia_admin = UserFactory.create(email="asia@admin.com")
         self.africa_admin = UserFactory.create(email="africa@admin.com")
         self.india_admin = UserFactory.create(email="india@admin.com")
+        self.ifrc_admin = UserFactory.create(email="ifrc@admin.test")
+        self.org_user = UserFactory.create(email="ifrc@admin.test")
         # India admin setup
         management.call_command("make_permissions")
         country_admin_codename = f"country_admin_{country.id}"
@@ -922,6 +931,20 @@ def test_country_region_admin_permission_for_local_unit_update(self):
         africa_admin_group.permissions.add(africa_admin_permission)
         self.africa_admin.groups.add(africa_admin_group)
 
+        # Ifrc admin
+        ifrc_admin_codename = "ifrc_admin"
+        ifrc_admin_permission = Permission.objects.get(codename=ifrc_admin_codename)
+        ifrc_admin_group_name = "IFRC Admins"
+        ifrc_admin_group = Group.objects.get(name=ifrc_admin_group_name)
+        ifrc_admin_group.permissions.add(ifrc_admin_permission)
+        self.ifrc_admin.groups.add(ifrc_admin_group)
+
+        # Set the user profile as organization type = NTLS for permission checks
+        profile = self.org_user.profile
+        profile.org_type = Profile.OrgTypes.NTLS
+        profile.country = country
+        profile.save()
+
         local_unit = LocalUnitFactory.create(
             country=country,
             type=self.local_unit_type,
@@ -936,6 +959,27 @@ def test_country_region_admin_permission_for_local_unit_update(self):
             status=LocalUnit.Status.VALIDATED,
             date_of_data="2023-08-08",
         )
+        local_unit3 = LocalUnitFactory.create(
+            country=country,
+            type=self.local_unit_type,
+            draft=False,
+            status=LocalUnit.Status.VALIDATED,
+            date_of_data="2023-08-08",
+        )
+        local_unit_4 = LocalUnitFactory.create(
+            country=country,
+            type=self.local_unit_type,
+            draft=False,
+            status=LocalUnit.Status.VALIDATED,
+            date_of_data="2023-08-08",
+        )
+        local_unit_5 = LocalUnitFactory.create(
+            country=country2,
+            type=self.local_unit_type,
+            draft=False,
+            status=LocalUnit.Status.VALIDATED,
+            date_of_data="2023-08-08",
+        )
         url = f"/api/v2/local-units/{local_unit.id}/"
         data = {
             "local_branch_name": "Updated local branch name",
@@ -969,6 +1013,26 @@ def test_country_region_admin_permission_for_local_unit_update(self):
         self.assert_200(response)
         self.assertEqual(response.data["local_branch_name"], "Updated local branch name")
 
+        # Test update as ifrc admin
+        url = f"/api/v2/local-units/{local_unit3.id}/"
+        self.authenticate(self.ifrc_admin)
+        response = self.client.patch(url, data=data, format="json")
+        self.assert_200(response)
+        self.assertEqual(response.data["local_branch_name"], "Updated local branch name")
+
+        # Test update as NTLS org type user with same local unit county
+        url = f"/api/v2/local-units/{local_unit_4.id}/"
+        self.authenticate(self.org_user)
+        response = self.client.patch(url, data=data, format="json")
+        self.assert_200(response)
+        self.assertEqual(response.data["local_branch_name"], "Updated local branch name")
+
+        # Test update as NTLS org type user with different  local unit county
+        url = f"/api/v2/local-units/{local_unit_5.id}/"
+        self.authenticate(self.org_user)
+        response = self.client.patch(url, data=data, format="json")
+        self.assert_403(response)
+
 
 class TestExternallyManagedLocalUnit(APITestCase):
     def setUp(self):
@@ -1177,15 +1241,15 @@ def setUp(self):
         global_group.permissions.add(global_permission)
         self.global_validator_user.groups.add(global_group)
 
-        file_path = os.path.join(settings.TEST_DIR, "local_unit/test.csv")
+        file_path = os.path.join(settings.TEST_DIR, "local_unit/test-admin.xlsx")
         with open(file_path, "rb") as f:
             self._file_content = f.read()
 
-    def create_upload_file(self, filename="test.csv"):
+    def create_upload_file(self, filename="test-admin.xlsx"):
         """
         Always return a new file instance to prevent stream exhaustion.
         """
-        return SimpleUploadedFile(filename, self._file_content, content_type="text/csv")
+        return SimpleUploadedFile(filename, self._file_content, content_type="text/xlsx")
 
     @mock.patch("local_units.tasks.process_bulk_upload_local_unit.delay")
     def test_bulk_upload_local_unit(self, mock_delay):
@@ -1330,16 +1394,16 @@ def setUpTestData(cls):
         cls.local_unit_type = LocalUnitType.objects.create(code=1, name="Administrative")
         cls.local_unit_type2 = LocalUnitType.objects.create(code=2, name="Health Care")
         cls.level = LocalUnitLevel.objects.create(level=0, name="National")
-        file_path = os.path.join(settings.TEST_DIR, "local_unit/test.csv")
+        file_path = os.path.join(settings.TEST_DIR, "local_unit/test-admin.xlsx")
         with open(file_path, "rb") as f:
             cls._file_content = f.read()
 
-    def create_upload_file(cls, filename="test.csv"):
-        return SimpleUploadedFile(filename, cls._file_content, content_type="text/csv")
+    def create_upload_file(cls, filename="test-admin.xlsx"):
+        return SimpleUploadedFile(filename, cls._file_content, content_type="text/xlsx")
 
     def test_bulk_upload_with_incorrect_country(cls):
         """
-        Test bulk upload fails when the country does not match CSV data.
+        Test bulk upload fails when the country does not match xlsx data.
         """
         cls.bulk_upload = LocalUnitBulkUploadFactory.create(
             country=cls.country1,
@@ -1362,13 +1426,13 @@ def test_bulk_upload_with_incorrect_country(cls):
 
     def test_bulk_upload_with_valid_country(cls):
         """
-        Test bulk upload succeeds when the country matches CSV data
+        Test bulk upload succeeds when the country matches xlsx data
         """
         cls.bulk_upload = LocalUnitBulkUploadFactory.create(
             country=cls.country2,  # Brazil
             local_unit_type=cls.local_unit_type,
             triggered_by=cls.user,
-            file=cls.create_upload_file(),  # CSV with Brazil rows
+            file=cls.create_upload_file(),  # xlsx with Brazil rows
             status=LocalUnitBulkUpload.Status.PENDING,
         )
         runner = BaseBulkUploadLocalUnit(cls.bulk_upload)
@@ -1381,7 +1445,7 @@ def test_bulk_upload_with_valid_country(cls):
 
     def test_bulk_upload_fails_and_delete(cls):
         """
-        Test bulk upload fails and delete when CSV has incorrect data.
+        Test bulk upload fails and delete when xlsx has incorrect data.
         """
         LocalUnitFactory.create_batch(
             5,
@@ -1410,7 +1474,7 @@ def test_bulk_upload_fails_and_delete(cls):
 
     def test_bulk_upload_deletes_old_and_creates_new_local_units(cls):
         """
-        Test bulk upload with correct CSV data.
+        Test bulk upload with correct data.
         """
         old_local_unit = LocalUnitFactory.create(
             country=cls.country2,
@@ -1442,10 +1506,14 @@ def test_empty_administrative_file(cls):
         Test bulk upload file is empty
         """
 
-        file_path = os.path.join(settings.STATICFILES_DIRS[0], "files", "local_units", "local-unit-bulk-upload-template.csv")
+        file_path = os.path.join(
+            settings.STATICFILES_DIRS[0], "files", "local_units", "Administrative Bulk Import Template - Local Units.xlsx"
+        )
         with open(file_path, "rb") as f:
             file_content = f.read()
-        empty_file = SimpleUploadedFile(name="local-unit-bulk-upload-template.csv", content=file_content, content_type="text/csv")
+        empty_file = SimpleUploadedFile(
+            name="Administrative Bulk Import Template - Local Units.xlsx", content=file_content, content_type="text/xlsx"
+        )
         LocalUnitFactory.create_batch(
             5,
             country=cls.country2,
@@ -1531,16 +1599,16 @@ def setUpTestData(cls):
         cls.professional_training_facilities = ProfessionalTrainingFacility.objects.create(code=1, name="Nurses")
         cls.general_medical_services = GeneralMedicalService.objects.create(code=1, name="Minor Trauma")
 
-        file_path = os.path.join(settings.TEST_DIR, "local_unit/test-health.csv")
+        file_path = os.path.join(settings.TEST_DIR, "local_unit/test-health.xlsm")
         with open(file_path, "rb") as f:
             cls._file_content = f.read()
 
-    def create_upload_file(cls, filename="test-health.csv"):
-        return SimpleUploadedFile(filename, cls._file_content, content_type="text/csv")
+    def create_upload_file(cls, filename="test-health.xlsm"):
+        return SimpleUploadedFile(filename, cls._file_content, content_type="text/xlsm")
 
     def test_bulk_upload_health_with_incorrect_country(cls):
         """
-        Should fail when CSV rows are not equal to bulk upload country.
+        Should fail when rows are not equal to bulk upload country.
         """
         cls.bulk_upload = LocalUnitBulkUploadFactory.create(
             country=cls.country1,
@@ -1561,7 +1629,7 @@ def test_bulk_upload_health_with_incorrect_country(cls):
 
     def test_bulk_upload_health_fails_and_does_not_delete(cls):
         """
-        Should fail and keep existing LocalUnits & HealthData when CSV invalid.
+        Should fail and keep existing LocalUnits & HealthData when file invalid.
         """
         health_data = HealthDataFactory.create_batch(
             5,
@@ -1640,12 +1708,12 @@ def test_empty_health_template_file(cls):
         """
 
         file_path = os.path.join(
-            settings.STATICFILES_DIRS[0], "files", "local_units", "local-unit-health-bulk-upload-template.csv"
+            settings.STATICFILES_DIRS[0], "files", "local_units", "Health-Care-Bulk-Import-Template-Local-Units.xlsm"
         )
         with open(file_path, "rb") as f:
             file_content = f.read()
         empty_file = SimpleUploadedFile(
-            name="local-unit-health-bulk-upload-template.csv", content=file_content, content_type="text/csv"
+            name="Health-Care-Bulk-Import-Template-Local-Units.xlsm", content=file_content, content_type="text/xlsm"
         )
         health_data = HealthDataFactory.create_batch(
             5,

local_units/utils.py

@@ -72,16 +72,13 @@ def get_model_field_names(
 
 
 def normalize_bool(value):
-    if isinstance(value, bool):
-        return value
     if not value:
         return False
     val = str(value).strip().lower()
-    if val in ("true", "1", "yes", "y"):
+    if val in ("yes"):
         return True
-    if val in ("false", "0", "no", "n"):
+    if val in ("no"):
         return False
-    return False
 
 
 def wash(string):
@@ -91,4 +88,4 @@ def wash(string):
 
 
 def numerize(value):
-    return value if value.isdigit() else 0
+    return value if value else 0