feat(organizations): add endpoints to handle organization members TAS…

…K-963 (#5235)

### 📣 Summary
- Enhance the organization member management API to support role updates
and member removal via PATCH and DELETE requests, excluding member
creation.


### 📖 Description
- This update introduces endpoints for managing organization members:

- GET `/api/v2/organizations/<org-id>/members/` to list all members in
an organization.
- PATCH `/api/v2/organizations/<org-id>/members/<username>/` to update a
member's role (e.g., promote to admin).
- DELETE `/api/v2/organizations/<org-id>/members/<username>/` to remove
a member from the organization.

Note: Creating members is not supported via this endpoint. Roles are
restricted to 'admin' or 'member'. Including the details of invited
members (those who have not yet joined the organization) is not covered
in this update.


### 👷 Description for instance maintainers
- This update provides new functionality to manage organization members:

- Role updates for members (promotion/demotion between 'admin' and
'member').
    - Member removal from an organization.
- The endpoints are optimized to use database joins to fetch member
roles efficiently without excessive database queries, ensuring minimal
load.



### 👀 Preview steps
1. ℹ️ Have an account and an organization with multiple members.
2. Use the GET method to list the members of the organization.
3. Use the PATCH method to update a member's role (e.g., promote a user
to admin).
4. Use the DELETE method to remove a member from the organization.
5. 🟢 Notice the endpoints behave as expected, with valid role updates
and member removal.


### 💭 Notes
- The code uses database joins to optimize role determination for
members, avoiding inefficient role lookups.
- Role validation restricts role assignments to 'admin' or 'member'
only.
- Including invited members' details is not covered in this update. This
will be implemented in a future update.

kobo/apps/organizations/permissions.py

@@ -1,14 +1,17 @@
 from django.http import Http404
 from rest_framework import permissions
+from rest_framework.permissions import IsAuthenticated
 
 from kobo.apps.organizations.constants import ORG_EXTERNAL_ROLE
+from kobo.apps.organizations.models import Organization
 from kpi.mixins.validation_password_permission import ValidationPasswordPermissionMixin
 from kpi.utils.object_permission import get_database_user
 
 
-class IsOrgAdmin(ValidationPasswordPermissionMixin, permissions.BasePermission):
+class IsOrgAdminPermission(ValidationPasswordPermissionMixin, IsAuthenticated):
     """
-    Object-level permission to only allow admin members of an object to access it.
+    Object-level permission to only allow admin (and owner) members of an object
+    to access it.
     Assumes the model instance has an `is_admin` attribute.
     """
 
@@ -26,18 +29,22 @@ def has_object_permission(self, request, view, obj):
         return obj.is_admin(user)
 
 
-class IsOrgAdminOrReadOnly(IsOrgAdmin):
-    """
-    Object-level permission to only allow admin members of an object to edit it.
-    Assumes the model instance has an `is_admin` attribute.
-    """
+class HasOrgRolePermission(IsOrgAdminPermission):
+    def has_permission(self, request, view):
+        if not super().has_permission(request, view):
+            return False
+
+        organization = Organization.objects.filter(
+            id=view.kwargs.get('organization_id')
+        ).first()
+        if organization and not self.has_object_permission(
+            request, view, organization
+        ):
+            return False
+        return True
 
     def has_object_permission(self, request, view, obj):
-
-        # Read permissions are allowed to any request,
-        # so we'll always allow GET, HEAD or OPTIONS requests.
-        if request.method in permissions.SAFE_METHODS:
+        obj = obj if isinstance(obj, Organization) else obj.organization
+        if super().has_object_permission(request, view, obj):
             return True
-
-        # Instance must have an attribute named `is_admin`
-        return obj.is_admin(request.user)
+        return request.method in permissions.SAFE_METHODS

kobo/apps/organizations/serializers.py

@@ -1,4 +1,7 @@
+from django.contrib.auth import get_user_model
+from django.utils.translation import gettext as t
 from rest_framework import serializers
+from rest_framework.reverse import reverse
 
 from kobo.apps.organizations.models import (
     Organization,
@@ -12,10 +15,62 @@
 
 
 class OrganizationUserSerializer(serializers.ModelSerializer):
+    user = serializers.HyperlinkedRelatedField(
+        queryset=get_user_model().objects.all(),
+        lookup_field='username',
+        view_name='user-kpi-detail',
+    )
+    role = serializers.CharField()
+    user__has_mfa_enabled = serializers.BooleanField(
+        source='has_mfa_enabled', read_only=True
+    )
+    url = serializers.SerializerMethodField()
+    date_joined = serializers.DateTimeField(
+        source='user.date_joined', format='%Y-%m-%dT%H:%M:%SZ'
+    )
+    user__username = serializers.ReadOnlyField(source='user.username')
+    user__extra_details__name = serializers.ReadOnlyField(
+        source='user.extra_details.data.name'
+    )
+    user__email = serializers.ReadOnlyField(source='user.email')
+    user__is_active = serializers.ReadOnlyField(source='user.is_active')
 
     class Meta:
         model = OrganizationUser
-        fields = ['user', 'organization']
+        fields = [
+            'url',
+            'user',
+            'user__username',
+            'user__email',
+            'user__extra_details__name',
+            'role',
+            'user__has_mfa_enabled',
+            'date_joined',
+            'user__is_active'
+        ]
+
+    def get_url(self, obj):
+        request = self.context.get('request')
+        return reverse(
+            'organization-members-detail',
+            kwargs={
+                'organization_id': obj.organization.id,
+                'user__username': obj.user.username
+            },
+            request=request
+        )
+
+    def update(self, instance, validated_data):
+        if role := validated_data.get('role', None):
+            validated_data['is_admin'] = role == 'admin'
+        return super().update(instance, validated_data)
+
+    def validate_role(self, role):
+        if role not in ['admin', 'member']:
+            raise serializers.ValidationError(
+                {'role': t("Invalid role. Only 'admin' or 'member' are allowed")}
+            )
+        return role
 
 
 class OrganizationOwnerSerializer(serializers.ModelSerializer):

kobo/apps/organizations/tests/test_organization_members_api.py

@@ -0,0 +1,131 @@
+from ddt import ddt, data, unpack
+from django.urls import reverse
+from rest_framework import status
+
+from kobo.apps.kobo_auth.shortcuts import User
+from kobo.apps.organizations.tests.test_organizations_api import (
+    BaseOrganizationAssetApiTestCase
+)
+from kpi.urls.router_api_v2 import URL_NAMESPACE
+
+
+@ddt
+class OrganizationMemberAPITestCase(BaseOrganizationAssetApiTestCase):
+    fixtures = ['test_data']
+    URL_NAMESPACE = URL_NAMESPACE
+
+    def setUp(self):
+        super().setUp()
+        self.organization = self.someuser.organization
+        self.owner_user = self.someuser
+        self.member_user = self.alice
+        self.admin_user = self.anotheruser
+        self.external_user = self.bob
+
+        self.list_url = reverse(
+            self._get_endpoint('organization-members-list'),
+            kwargs={'organization_id': self.organization.id},
+        )
+        self.detail_url = lambda username: reverse(
+            self._get_endpoint('organization-members-detail'),
+            kwargs={
+                'organization_id': self.organization.id,
+                'user__username': username
+            },
+        )
+
+    @data(
+        ('owner', status.HTTP_200_OK),
+        ('admin', status.HTTP_200_OK),
+        ('member', status.HTTP_200_OK),
+        ('external', status.HTTP_404_NOT_FOUND),
+        ('anonymous', status.HTTP_401_UNAUTHORIZED),
+    )
+    @unpack
+    def test_list_members_with_different_roles(self, user_role, expected_status):
+        if user_role == 'anonymous':
+            self.client.logout()
+        else:
+            user = getattr(self, f'{user_role}_user')
+            self.client.force_login(user)
+        response = self.client.get(self.list_url)
+        self.assertEqual(response.status_code, expected_status)
+
+    @data(
+        ('owner', status.HTTP_200_OK),
+        ('admin', status.HTTP_200_OK),
+        ('member', status.HTTP_200_OK),
+        ('external', status.HTTP_404_NOT_FOUND),
+        ('anonymous', status.HTTP_401_UNAUTHORIZED),
+    )
+    @unpack
+    def test_retrieve_member_details_with_different_roles(
+            self, user_role, expected_status
+    ):
+        if user_role == 'anonymous':
+            self.client.logout()
+        else:
+            user = getattr(self, f'{user_role}_user')
+            self.client.force_login(user)
+        response = self.client.get(self.detail_url(self.member_user))
+        self.assertEqual(response.status_code, expected_status)
+
+    @data(
+        ('owner', status.HTTP_200_OK),
+        ('admin', status.HTTP_200_OK),
+        ('member', status.HTTP_403_FORBIDDEN),
+        ('external', status.HTTP_404_NOT_FOUND),
+        ('anonymous', status.HTTP_401_UNAUTHORIZED),
+    )
+    @unpack
+    def test_update_member_role_with_different_roles(self, user_role, expected_status):
+        if user_role == 'anonymous':
+            self.client.logout()
+        else:
+            user = getattr(self, f'{user_role}_user')
+            self.client.force_login(user)
+        data = {'role': 'admin'}
+        response = self.client.patch(self.detail_url(self.member_user), data)
+        self.assertEqual(response.status_code, expected_status)
+
+    @data(
+        ('owner', status.HTTP_204_NO_CONTENT),
+        ('admin', status.HTTP_204_NO_CONTENT),
+        ('member', status.HTTP_403_FORBIDDEN),
+        ('external', status.HTTP_404_NOT_FOUND),
+        ('anonymous', status.HTTP_401_UNAUTHORIZED),
+    )
+    @unpack
+    def test_delete_member_with_different_roles(self, user_role, expected_status):
+        if user_role == 'anonymous':
+            self.client.logout()
+        else:
+            user = getattr(self, f'{user_role}_user')
+            self.client.force_login(user)
+        response = self.client.delete(self.detail_url(self.member_user))
+        self.assertEqual(response.status_code, expected_status)
+        if expected_status == status.HTTP_204_NO_CONTENT:
+            # Confirm deletion
+            response = self.client.get(self.detail_url(self.member_user))
+            self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+            self.assertFalse(
+                User.objects.filter(username=f'{user_role}_user').exists()
+            )
+
+    @data(
+        ('owner', status.HTTP_405_METHOD_NOT_ALLOWED),
+        ('admin', status.HTTP_405_METHOD_NOT_ALLOWED),
+        ('member', status.HTTP_403_FORBIDDEN),
+        ('external', status.HTTP_404_NOT_FOUND),
+        ('anonymous', status.HTTP_401_UNAUTHORIZED),
+    )
+    @unpack
+    def test_post_request_is_not_allowed(self, user_role, expected_status):
+        if user_role == 'anonymous':
+            self.client.logout()
+        else:
+            user = getattr(self, f'{user_role}_user')
+            self.client.force_login(user)
+        data = {'role': 'admin'}
+        response = self.client.post(self.list_url, data)
+        self.assertEqual(response.status_code, expected_status)

kobo/apps/organizations/tests/test_organizations_api.py

@@ -61,7 +61,12 @@ def test_anonymous_user(self):
     def test_create(self):
         data = {'name': 'my org'}
         res = self.client.post(self.url_list, data)
-        self.assertContains(res, data['name'], status_code=201)
+        self.assertEqual(res.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)
+
+    def test_delete(self):
+        self._insert_data()
+        res = self.client.delete(self.url_detail)
+        self.assertEqual(res.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)
 
     def test_list(self):
         self._insert_data()
@@ -349,7 +354,7 @@ def test_can_list(self, username, expected_status_code):
     def test_list_not_found_as_anonymous(self):
         self.client.logout()
         response = self.client.get(self.org_assets_list_url)
-        assert response.status_code == status.HTTP_404_NOT_FOUND
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED
 
     def test_list_only_organization_assets(self):
         # The organization's assets endpoint only returns assets where the `owner`