feat(organizations): add endpoints to handle organization members TASK-963

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`