Merge pull request #1722 from BLSQ/IA-3560-out-restriction-user-roles…

…-back

IA-3560 Org Unit Type restrictions in `UserRole` (backend)

iaso/api/org_unit_change_request_configurations/views_mobile.py

@@ -71,12 +71,10 @@ def list(self, request: Request, *args, **kwargs) -> Response:
         app_id = AppIdSerializer(data=self.request.query_params).get_app_id(raise_exception=True)
         queryset = self.get_queryset()
 
-        user_editable_org_unit_type_ids = set(
-            self.request.user.iaso_profile.editable_org_unit_types.values_list("id", flat=True)
-        )
+        user_editable_org_unit_type_ids = self.request.user.iaso_profile.get_editable_org_unit_type_ids()
 
         if user_editable_org_unit_type_ids:
-            project_org_unit_types = set(Project.objects.get(app_id=app_id).unit_types.values_list("id", flat=True))
+            project_org_unit_types = set(Project.objects.filter(app_id=app_id).values_list("unit_types__id", flat=True))
             non_editable_org_unit_type_ids = project_org_unit_types - user_editable_org_unit_type_ids
 
             dynamic_configurations = [

iaso/api/profiles/profiles.py

@@ -220,7 +220,7 @@ class ProfilesViewSet(viewsets.ViewSet):
 
     def get_queryset(self):
         account = self.request.user.iaso_profile.account
-        return Profile.objects.filter(account=account).prefetch_related("editable_org_unit_types")
+        return Profile.objects.filter(account=account).with_editable_org_unit_types()
 
     def list(self, request):
         limit = request.GET.get("limit", None)
@@ -260,7 +260,7 @@ def list(self, request):
             teams=teams,
             managed_users_only=managed_users_only,
             ids=ids,
-        )
+        ).order_by("id")
 
         queryset = queryset.prefetch_related(
             "user",

iaso/api/user_roles.py

@@ -1,13 +1,13 @@
-from typing import Any
-from django.conf import settings
-from django.shortcuts import get_object_or_404
-from rest_framework.response import Response
 from rest_framework import permissions, serializers, status
+from rest_framework.response import Response
+
 from django.contrib.auth.models import Permission, Group
 from django.db.models import Q, QuerySet
-from iaso.models import UserRole
+from django.shortcuts import get_object_or_404
+
 from .common import TimestampField, ModelViewSet
 from hat.menupermissions import models as permission
+from iaso.models import Project, OrgUnitType, UserRole
 
 
 class HasUserRolePermission(permissions.BasePermission):
@@ -23,45 +23,43 @@ class Meta:
         fields = ("id", "name", "codename")
 
 
+class OrgUnitTypeNestedReadSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = OrgUnitType
+        fields = ["id", "name", "short_name"]
+
+
 class UserRoleSerializer(serializers.ModelSerializer):
     permissions = serializers.SerializerMethodField("get_permissions")
     name = serializers.CharField(source="group.name")
+    created_at = TimestampField(read_only=True)
+    updated_at = TimestampField(read_only=True)
 
     class Meta:
         model = UserRole
-        fields = ["id", "name", "permissions", "created_at", "updated_at"]
+        fields = ["id", "name", "permissions", "editable_org_unit_types", "created_at", "updated_at"]
 
     def to_representation(self, instance):
         user_role = super().to_representation(instance)
         account_id = user_role["name"].split("_")[0]
-        user_role["name"] = self.remove_prefix_from_str(user_role["name"], account_id + "_")
+        user_role["name"] = user_role["name"].removeprefix(f"{account_id}_")
+        user_role["editable_org_unit_types"] = OrgUnitTypeNestedReadSerializer(
+            instance.editable_org_unit_types.only("id", "name", "short_name").order_by("id"), many=True
+        ).data
         return user_role
 
-    created_at = TimestampField(read_only=True)
-    updated_at = TimestampField(read_only=True)
-
-    # This method will remove a given prefix from a string
-    def remove_prefix_from_str(self, str, prefix):
-        if str.startswith(prefix):
-            return str[len(prefix) :]
-        return str
-
     def get_permissions(self, obj):
         return [permission["codename"] for permission in PermissionSerializer(obj.group.permissions, many=True).data]
 
     def create(self, validated_data):
-        account = self.context["request"].user.iaso_profile.account
         request = self.context["request"]
+        account = request.user.iaso_profile.account
         group_name = str(account.id) + "_" + request.data.get("name")
         permissions = request.data.get("permissions", [])
-
-        # check if the user role name has been given
-        if not group_name:
-            raise serializers.ValidationError({"name": "User role name is required"})
+        editable_org_unit_types = validated_data.get("editable_org_unit_types", [])
 
         # check if a user role with the same name already exists
-        group_exists = Group.objects.filter(name__iexact=group_name)
-        if group_exists:
+        if Group.objects.filter(name__iexact=group_name).exists():
             raise serializers.ValidationError({"name": "User role already exists"})
 
         group = Group(name=group_name)
@@ -73,21 +71,20 @@ def create(self, validated_data):
                 group.permissions.add(permission)
             group.save()
 
-        userRole = UserRole.objects.create(group=group, account=account)
-        userRole.save()
-        return userRole
+        user_role = UserRole.objects.create(group=group, account=account)
+        user_role.save()
+        user_role.editable_org_unit_types.set(editable_org_unit_types)
+        return user_role
 
     def update(self, user_role, validated_data):
-        account = self.context["request"].user.iaso_profile.account
-        group_name = str(account.id) + "_" + self.context["request"].data.get("name", None)
-        permissions = self.context["request"].data.get("permissions", None)
+        request = self.context["request"]
+        account = request.user.iaso_profile.account
         group = user_role.group
+        group_name = str(account.id) + "_" + validated_data.get("group", {}).get("name")
+        permissions = request.data.get("permissions", None)
+        editable_org_unit_types = validated_data.get("editable_org_unit_types", [])
 
-        if group_name is not None:
-            group.name = group_name
-        # check if a user role with the same name already exists other than the current user role
-        group_exists = Group.objects.filter(~Q(pk=group.id), name__iexact=group_name)
-        if group_exists:
+        if Group.objects.filter(~Q(pk=group.id), name__iexact=group_name).exists():
             raise serializers.ValidationError({"name": "User role already exists"})
 
         if permissions is not None:
@@ -98,8 +95,19 @@ def update(self, user_role, validated_data):
 
         group.save()
         user_role.save()
+        user_role.editable_org_unit_types.set(editable_org_unit_types)
         return user_role
 
+    def validate_editable_org_unit_types(self, editable_org_unit_types):
+        account = self.context.get("request").user.iaso_profile.account
+        project_org_unit_types = set(Project.objects.filter(account=account).values_list("unit_types__id", flat=True))
+        for org_unit_type in editable_org_unit_types:
+            if org_unit_type.pk not in project_org_unit_types:
+                raise serializers.ValidationError(
+                    f"`{org_unit_type.name} ({org_unit_type.pk})` is not a valid Org Unit Type for this account."
+                )
+        return editable_org_unit_types
+
 
 class UserRolesViewSet(ModelViewSet):
     f"""Roles API
@@ -120,7 +128,9 @@ class UserRolesViewSet(ModelViewSet):
 
     def get_queryset(self) -> QuerySet[UserRole]:
         user = self.request.user
-        queryset = UserRole.objects.filter(account=user.iaso_profile.account)  # type: ignore
+        queryset = UserRole.objects.filter(account=user.iaso_profile.account).prefetch_related(
+            "group__permissions", "editable_org_unit_types"
+        )
         search = self.request.GET.get("search", None)
         orders = self.request.GET.get("order", "group__name").split(",")
         if search:

iaso/migrations/0305_userrole_editable_org_unit_types.py

@@ -0,0 +1,17 @@
+# Generated by Django 4.2.16 on 2024-10-16 13:06
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("iaso", "0304_profile_org_unit_types"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="userrole",
+            name="editable_org_unit_types",
+            field=models.ManyToManyField(blank=True, related_name="editable_by_user_role_set", to="iaso.orgunittype"),
+        ),
+    ]

iaso/models/base.py

@@ -1405,6 +1405,21 @@ def as_dict(self):
         }
 
 
+class ProfileQuerySet(models.QuerySet):
+    def with_editable_org_unit_types(self):
+        qs = self
+        return qs.annotate(
+            annotated_editable_org_unit_types_ids=ArrayAgg(
+                "editable_org_unit_types__id", distinct=True, filter=Q(editable_org_unit_types__isnull=False)
+            ),
+            annotated_user_roles_editable_org_unit_type_ids=ArrayAgg(
+                "user_roles__editable_org_unit_types__id",
+                distinct=True,
+                filter=Q(user_roles__editable_org_unit_types__isnull=False),
+            ),
+        )
+
+
 class Profile(models.Model):
     account = models.ForeignKey(Account, on_delete=models.CASCADE)
     user = models.OneToOneField(User, on_delete=models.CASCADE, related_name="iaso_profile")
@@ -1424,6 +1439,8 @@ class Profile(models.Model):
         "OrgUnitType", related_name="editable_by_iaso_profile_set", blank=True
     )
 
+    objects = models.Manager.from_queryset(ProfileQuerySet)()
+
     class Meta:
         constraints = [models.UniqueConstraint(fields=["dhis2_id", "account"], name="dhis2_id_constraint")]
 
@@ -1440,6 +1457,16 @@ def as_dict(self, small=False):
         )
         all_permissions = user_group_permissions + user_permissions
         permissions = list(set(all_permissions))
+        try:
+            editable_org_unit_type_ids = self.annotated_editable_org_unit_types_ids
+        except AttributeError:
+            editable_org_unit_type_ids = [out.pk for out in self.editable_org_unit_types.all()]
+        try:
+            user_roles_editable_org_unit_type_ids = self.annotated_user_roles_editable_org_unit_type_ids
+        except AttributeError:
+            user_roles_editable_org_unit_type_ids = (
+                list(self.user_roles.values_list("editable_org_unit_types", flat=True)),
+            )
         if not small:
             return {
                 "id": self.id,
@@ -1462,7 +1489,8 @@ def as_dict(self, small=False):
                 "phone_number": self.phone_number.as_e164 if self.phone_number else None,
                 "country_code": region_code_for_number(self.phone_number).lower() if self.phone_number else None,
                 "projects": [p.as_dict() for p in self.projects.all().order_by("name")],
-                "editable_org_unit_type_ids": [out.pk for out in self.editable_org_unit_types.all()],
+                "editable_org_unit_type_ids": editable_org_unit_type_ids,
+                "user_roles_editable_org_unit_type_ids": user_roles_editable_org_unit_type_ids,
             }
         else:
             return {
@@ -1485,10 +1513,21 @@ def as_dict(self, small=False):
                 "phone_number": self.phone_number.as_e164 if self.phone_number else None,
                 "country_code": region_code_for_number(self.phone_number).lower() if self.phone_number else None,
                 "projects": [p.as_dict() for p in self.projects.all()],
-                "editable_org_unit_type_ids": [out.pk for out in self.editable_org_unit_types.all()],
+                "editable_org_unit_type_ids": editable_org_unit_type_ids,
+                "user_roles_editable_org_unit_type_ids": user_roles_editable_org_unit_type_ids,
             }
 
     def as_short_dict(self):
+        try:
+            editable_org_unit_type_ids = self.annotated_editable_org_unit_types_ids
+        except AttributeError:
+            editable_org_unit_type_ids = [out.pk for out in self.editable_org_unit_types.all()]
+        try:
+            user_roles_editable_org_unit_type_ids = self.annotated_user_roles_editable_org_unit_type_ids
+        except AttributeError:
+            user_roles_editable_org_unit_type_ids = (
+                list(self.user_roles.values_list("editable_org_unit_types", flat=True)),
+            )
         return {
             "id": self.id,
             "first_name": self.user.first_name,
@@ -1499,7 +1538,8 @@ def as_short_dict(self):
             "user_id": self.user.id,
             "phone_number": self.phone_number.as_e164 if self.phone_number else None,
             "country_code": region_code_for_number(self.phone_number).lower() if self.phone_number else None,
-            "editable_org_unit_type_ids": [out.pk for out in self.editable_org_unit_types.all()],
+            "editable_org_unit_type_ids": editable_org_unit_type_ids,
+            "user_roles_editable_org_unit_type_ids": user_roles_editable_org_unit_type_ids,
         }
 
     def has_a_team(self):
@@ -1508,12 +1548,15 @@ def has_a_team(self):
             return True
         return False
 
+    def get_editable_org_unit_type_ids(self) -> set[int]:
+        ids_in_user_roles = set(self.user_roles.values_list("editable_org_unit_types", flat=True))
+        ids_in_user_profile = set(self.editable_org_unit_types.values_list("id", flat=True))
+        return ids_in_user_profile.union(ids_in_user_roles)
+
     def has_org_unit_write_permission(
-        self, org_unit_type_id: int, prefetched_editable_org_unit_type_ids: list = None
+        self, org_unit_type_id: int, prefetched_editable_org_unit_type_ids: set[int] = None
     ) -> bool:
-        editable_org_unit_type_ids = prefetched_editable_org_unit_type_ids or list(
-            self.editable_org_unit_types.values_list("id", flat=True)
-        )
+        editable_org_unit_type_ids = prefetched_editable_org_unit_type_ids or self.get_editable_org_unit_type_ids()
         if not editable_org_unit_type_ids:
             return True
         return org_unit_type_id in editable_org_unit_type_ids
@@ -1674,6 +1717,11 @@ class UserRole(models.Model):
     group = models.OneToOneField(auth.models.Group, on_delete=models.CASCADE, related_name="iaso_user_role")
     created_at = models.DateTimeField(auto_now_add=True)
     updated_at = models.DateTimeField(auto_now=True)
+    # Each user can have restricted write access to OrgUnits, based on their type.
+    # By default, empty `editable_org_unit_types` means access to everything.
+    editable_org_unit_types = models.ManyToManyField(
+        "OrgUnitType", related_name="editable_by_user_role_set", blank=True
+    )
 
     def __str__(self) -> str:
         return self.group.name

iaso/tasks/org_units_bulk_update.py

@@ -81,7 +81,7 @@ def org_units_bulk_update(
         raise Exception("Modification on read only source are not allowed")
 
     total = queryset.count()
-    editable_org_unit_type_ids = list(user.iaso_profile.editable_org_unit_types.values_list("id", flat=True))
+    editable_org_unit_type_ids = user.iaso_profile.get_editable_org_unit_type_ids()
     skipped_messages = []
 
     # FIXME Task don't handle rollback properly if task is killed by user or other error

iaso/tests/api/org_unit_change_request_configurations/test_org_unit_change_request_configs_mobile.py

@@ -1,7 +1,7 @@
-import json
-
 from rest_framework import status
 
+from django.contrib.auth.models import Group
+
 from iaso import models as m
 from iaso.tests.api.org_unit_change_request_configurations.common_base_with_setup import OUCRCAPIBase
 
@@ -15,16 +15,17 @@ class MobileOrgUnitChangeRequestConfigurationAPITestCase(OUCRCAPIBase):
 
     def test_list_ok(self):
         self.client.force_authenticate(self.user_ash_ketchum)
-        with self.assertNumQueries(8):
+        with self.assertNumQueries(9):
             # get_queryset
             #   1. SELECT user_editable_org_unit_type_ids
-            #   2. COUNT(*) OrgUnitChangeRequestConfiguration
-            #   3. SELECT OrgUnitChangeRequestConfiguration
-            #   4. PREFETCH OrgUnitChangeRequestConfiguration.possible_types
-            #   5. PREFETCH OrgUnitChangeRequestConfiguration.possible_parent_types
-            #   6. PREFETCH OrgUnitChangeRequestConfiguration.group_sets
-            #   7. PREFETCH OrgUnitChangeRequestConfiguration.editable_reference_forms
-            #   8. PREFETCH OrgUnitChangeRequestConfiguration.other_groups
+            #   2. SELECT user_roles_editable_org_unit_type_ids
+            #   3. COUNT(*) OrgUnitChangeRequestConfiguration
+            #   4. SELECT OrgUnitChangeRequestConfiguration
+            #   5. PREFETCH OrgUnitChangeRequestConfiguration.possible_types
+            #   6. PREFETCH OrgUnitChangeRequestConfiguration.possible_parent_types
+            #   7. PREFETCH OrgUnitChangeRequestConfiguration.group_sets
+            #   8. PREFETCH OrgUnitChangeRequestConfiguration.editable_reference_forms
+            #   9. PREFETCH OrgUnitChangeRequestConfiguration.other_groups
             response = self.client.get(f"{self.MOBILE_OUCRC_API_URL}?app_id={self.app_id}")
             self.assertJSONResponse(response, status.HTTP_200_OK)
             self.assertEqual(3, len(response.data["results"]))  # the 3 OUCRCs from setup
@@ -39,10 +40,20 @@ def test_list_ok_with_restricted_write_permission_for_user(self):
         new_org_unit_type_3.projects.add(self.project_johto)
         self.assertEqual(self.project_johto.unit_types.count(), 6)
 
+        # Restrict write permissions on Org Units at the "Profile" level.
         self.user_ash_ketchum.iaso_profile.editable_org_unit_types.set(
             # Only org units of this type are now writable for this user.
-            [self.ou_type_fire_pokemons, new_org_unit_type_3]
+            [self.ou_type_fire_pokemons]
+        )
+
+        # Restrict write permissions on Org Units at the "Role" level.
+        group = Group.objects.create(name="Group")
+        user_role = m.UserRole.objects.create(group=group, account=self.account_pokemon)
+        user_role.editable_org_unit_types.set(
+            # Only org units of this type are now writable for this user.
+            [new_org_unit_type_3]
         )
+        self.user_ash_ketchum.iaso_profile.user_roles.set([user_role])
 
         self.client.force_authenticate(self.user_ash_ketchum)