Promote/remove admins of a org

certego_saas/apps/organization/serializers.py

@@ -1,5 +1,6 @@
 from rest_flex_fields.serializers import FlexFieldsModelSerializer
 from rest_framework import serializers as rfs
+from rest_framework.exceptions import NotFound, PermissionDenied
 
 from certego_saas.apps.user.models import User
 from certego_saas.apps.user.serializers import UserSerializer
@@ -92,3 +93,40 @@ def create(self, validated_data) -> Invitation:
             send_email=True,
             request=request,
         )
+
+class AdminActionsSerializer(rfs.Serializer):
+    class Meta:
+        fields = ["username", "request_user_username"]
+
+    username = rfs.CharField()
+    request_user_username = rfs.CharField()
+
+    def validate(self, data): 
+        username = data.get("username")
+        try:
+            request_user = User.objects.get(username=data.get("request_user_username"))
+        except User.DoesNotExist:
+            raise NotFound()
+
+        try:
+            # request_user must have a membership
+            membership_request_user = request_user.membership
+        except Membership.DoesNotExist:
+            raise rfs.ValidationError({'detail': 'You are not a member of any organization.'})
+
+        try:
+            # user to promote/remove must be a member of request_user organization
+            membership_user = membership_request_user.organization.members.get(
+                user__username=username
+            )
+        except Membership.DoesNotExist:
+            raise rfs.ValidationError({'detail': 'User to promote/remove is not part of your organization.'})
+
+        if membership_user.is_owner:
+            raise PermissionDenied(
+                detail="You can't modify owner permission", code=403
+            )
+        # only the owner can promote/remove the user as admin
+        if not membership_request_user.is_owner:
+            raise PermissionDenied(detail="You are not the owner of the org", code=403)
+        return data

certego_saas/apps/organization/views.py

@@ -23,6 +23,7 @@
     InvitationsListSerializer,
     InviteCreateSerializer,
     OrganizationSerializer,
+    AdminActionsSerializer,
 )
 
 logger = logging.getLogger(__name__)
@@ -166,6 +167,40 @@ def leave(self, request, *args, **kwargs):
             raise Membership.OwnerCantLeaveException()
         request.user.membership.delete()
         return Response(status=status.HTTP_204_NO_CONTENT)
+
+    @action(detail=False, methods=["POST"])
+    def promote_admin(self, request, *args, **kwargs):
+        """
+        Promote user as an admin of the org
+
+        ``POST ~/organization/promote_admin``.
+        """
+        username_to_promote = request.data.get("username", None)
+        logger.info(f"promote {username_to_promote} as admin from user {request.user}")
+        org = self.get_object()
+        serializer = AdminActionsSerializer(data={"username": username_to_promote, "request_user_username": request.user.username})
+        serializer.is_valid(raise_exception=True)
+        membership_user_to_promote = org.members.get(user__username=username_to_promote)
+        membership_user_to_promote.is_admin = True
+        membership_user_to_promote.save()
+        return Response(status=status.HTTP_200_OK)
+
+    @action(detail=False, methods=["POST"])
+    def remove_admin(self, request, *args, **kwargs):
+        """
+        Remove user as admin of the org
+
+        ``POST ~/organization/remove_admin``.
+        """
+        username_to_remove = request.data.get("username", None)
+        logger.info(f"remove {username_to_remove} as admin from user {request.user}")
+        org = self.get_object()
+        serializer = AdminActionsSerializer(data={"username": username_to_remove, "request_user_username": request.user.username})
+        serializer.is_valid(raise_exception=True)
+        membership_user_to_remove = org.members.get(user__username=username_to_remove)
+        membership_user_to_remove.is_admin = False
+        membership_user_to_remove.save()
+        return Response(status=status.HTTP_204_NO_CONTENT)
 
 
 class InvitationViewSet(ListAndDeleteOnlyViewSet):

tests/apps/organization/test_organization.py

@@ -11,6 +11,8 @@
 org_leave_uri = reverse("user_organization-leave")
 org_invite_uri = reverse("user_organization-invite")
 org_remove_member_uri = reverse("user_organization-remove-member")
+org_remove_admin_uri = reverse("user_organization-remove-admin")
+org_promote_admin_uri = reverse("user_organization-promote-admin")
 
 
 @tag("apps", "organization")
@@ -357,3 +359,149 @@ def test_error_invite(self):
             org_invite_uri, {"username": self.user_owner_org_1.username}
         )
         self.assertEqual(404, response.status_code)
+
+    def test_correct_remove_admin_from_org(self):
+        """Only owner can remove user as admin from their org"""
+        self.client.force_authenticate(self.user_owner_org_1)
+        response = self.client.post(
+            org_remove_admin_uri, {"username": self.user_admin2_org_1.username}
+        )
+        self.assertEqual(204, response.status_code)
+
+    def test_error_remove_admin_from_org(self):
+        """
+        1 - common user cannot remove admins
+        2 - an admin cannot remove other admin
+        3 - request without a username
+        4 - request with a username not existing
+        5 - request with a valid username, but it's not a member
+        6 - request with a valid username and member of another org
+        7 - user with no org
+        """
+        # 1 - common user cannot remove admin
+        self.client.force_authenticate(self.user_common_org_1)
+        response = self.client.post(
+            org_remove_admin_uri, {"username": self.user_owner_org_1.username}
+        )
+        self.assertEqual(403, response.status_code)
+        response = self.client.post(
+            org_remove_admin_uri, {"username": self.user_admin_org_1.username}
+        )
+        self.assertEqual(403, response.status_code)
+        # 2 - an admin cannot remove other admin
+        self.client.force_authenticate(self.user_admin_org_1)
+        response = self.client.post(
+            org_remove_admin_uri, {"username": self.user_owner_org_1.username}
+        )
+        self.assertEqual(403, response.status_code)
+        self.assertEqual("You can't modify owner permission", response.json()["detail"])
+        response = self.client.post(
+            org_remove_admin_uri, {"username": self.user_admin2_org_1.username}
+        )
+        self.assertEqual(403, response.status_code)
+        self.assertEqual("You are not the owner of the org", response.json()["detail"])
+        # 3 - request without a username
+        self.client.force_authenticate(self.user_admin_org_1)
+        response = self.client.post(org_remove_admin_uri, {"username": ""})
+        self.assertEqual(400, response.status_code)
+        self.assertIn("This field may not be blank.", response.json()["errors"]["username"])
+        # 4 - request with a username not existing
+        self.client.force_authenticate(self.user_admin_org_1)
+        response = self.client.post(
+            org_remove_admin_uri, {"username": "not_existing_user"}
+        )
+        self.assertEqual(400, response.status_code)
+        self.assertIn("User to promote/remove is not part of your organization.", response.json()["errors"]["detail"])
+        # 5 - request with a valid username, but it's not a member
+        self.client.force_authenticate(self.user_admin_org_1)
+        response = self.client.post(
+            org_remove_admin_uri, {"username": self.user_no_org.username}
+        )
+        self.assertEqual(400, response.status_code)
+        self.assertIn("User to promote/remove is not part of your organization.", response.json()["errors"]["detail"])
+        # 6 - request with a valid username and member of another org
+        self.client.force_authenticate(self.user_admin_org_1)
+        response = self.client.post(
+            org_remove_admin_uri, {"username": self.user_common_org_2.username}
+        )
+        self.assertEqual(400, response.status_code)
+        self.assertIn("User to promote/remove is not part of your organization.", response.json()["errors"]["detail"])
+        # 7 - user with no org
+        self.client.force_authenticate(self.user_no_org)
+        response = self.client.post(
+            org_remove_admin_uri, {"username": self.user_common_org_2.username}
+        )
+        self.assertEqual(404, response.status_code)
+
+    def test_correct_promote_admin_from_org(self):
+        """Only owner can promote user as admin from their org"""
+        self.client.force_authenticate(self.user_owner_org_1)
+        response = self.client.post(
+            org_promote_admin_uri, {"username": self.user_common_org_1.username}
+        )
+        self.assertEqual(200, response.status_code)
+
+    def test_error_promote_admin_from_org(self):
+        """
+        1 - common user cannot promote admins
+        2 - an admin cannot promote other admin
+        3 - request without a username
+        4 - request with a username not existing
+        5 - request with a valid username, but it's not a member
+        6 - request with a valid username and member of another org
+        7 - user with no org
+        """
+        # 1 - common user cannot promote admin
+        self.client.force_authenticate(self.user_common_org_1)
+        response = self.client.post(
+            org_promote_admin_uri, {"username": self.user_owner_org_1.username}
+        )
+        self.assertEqual(403, response.status_code)
+        response = self.client.post(
+            org_promote_admin_uri, {"username": self.user_admin_org_1.username}
+        )
+        self.assertEqual(403, response.status_code)
+        # 2 - an admin cannot promote other admin
+        self.client.force_authenticate(self.user_admin_org_1)
+        response = self.client.post(
+            org_promote_admin_uri, {"username": self.user_owner_org_1.username}
+        )
+        self.assertEqual(403, response.status_code)
+        self.assertEqual("You can't modify owner permission", response.json()["detail"])
+        response = self.client.post(
+            org_promote_admin_uri, {"username": self.user_admin2_org_1.username}
+        )
+        self.assertEqual(403, response.status_code)
+        self.assertEqual("You are not the owner of the org", response.json()["detail"])
+        # 3 - request without a username
+        self.client.force_authenticate(self.user_admin_org_1)
+        response = self.client.post(org_promote_admin_uri, {"username": ""})
+        self.assertEqual(400, response.status_code)
+        self.assertIn("This field may not be blank.", response.json()["errors"]["username"])
+        # 4 - request with a username not existing
+        self.client.force_authenticate(self.user_admin_org_1)
+        response = self.client.post(
+            org_promote_admin_uri, {"username": "not_existing_user"}
+        )
+        self.assertEqual(400, response.status_code)
+        self.assertIn("User to promote/remove is not part of your organization.", response.json()["errors"]["detail"])
+        # 5 - request with a valid username, but it's not a member
+        self.client.force_authenticate(self.user_admin_org_1)
+        response = self.client.post(
+            org_promote_admin_uri, {"username": self.user_no_org.username}
+        )
+        self.assertEqual(400, response.status_code)
+        self.assertIn("User to promote/remove is not part of your organization.", response.json()["errors"]["detail"])
+        # 6 - request with a valid username and member of another org
+        self.client.force_authenticate(self.user_owner_org_1)
+        response = self.client.post(
+            org_promote_admin_uri, {"username": self.user_common_org_2.username}
+        )
+        self.assertEqual(400, response.status_code)
+        self.assertIn("User to promote/remove is not part of your organization.", response.json()["errors"]["detail"])
+        # 7 - user with no org
+        self.client.force_authenticate(self.user_no_org)
+        response = self.client.post(
+            org_promote_admin_uri, {"username": self.user_common_org_2.username}
+        )
+        self.assertEqual(404, response.status_code)