Merge pull request #22 from ProjetSigma/new_routes_groupes

Implement PUT /group-member/{pk}/rank and DELETE /group-member/{pk}/

sigma_core/tests/test_group_member.py

@@ -11,6 +11,111 @@
 from sigma_core.serializers.user import DetailedUserSerializer as UserSerializer
 from sigma_core.tests.factories import UserFactory, AdminUserFactory, GroupFactory, GroupMemberFactory
 
+# Test /rank, /kick
+class GroupMemberPermissionTests(APITestCase):
+    @classmethod
+    def setUpTestData(self):
+        super().setUpTestData()
+        self.member_rank_url = "/group-member/%d/rank/"
+        self.member_url = "/group-member/%d/"
+        self.users = UserFactory.create_batch(6)
+        self.group = GroupFactory(req_rank_promote=3, req_rank_demote=4, req_rank_kick=5)
+        self.mships = [
+            None,
+            GroupMemberFactory(user=self.users[1], group=self.group, perm_rank=1),
+            GroupMemberFactory(user=self.users[2], group=self.group, perm_rank=2),
+            GroupMemberFactory(user=self.users[3], group=self.group, perm_rank=3),
+            GroupMemberFactory(user=self.users[4], group=self.group, perm_rank=4),
+            GroupMemberFactory(user=self.users[5], group=self.group, perm_rank=5)
+        ]
+
+    def try_rank(self, userId, targetId, newPermRank, expectedHttpResponseCode):
+        """
+        This function attempts to set $targetId membership rank to $newPermRank
+        using $userId permissions
+        PUT /group-member/{targetId}/rank with perm_rank=newPermRank
+        """
+        if userId >= 0:
+            self.client.force_authenticate(user=self.users[userId])
+        response = self.client.put(self.member_rank_url % (self.mships[targetId].id), {"perm_rank": newPermRank})
+        self.assertEqual(response.status_code, expectedHttpResponseCode)
+
+    def try_delete(self, userId, targetId, expectedHttpResponseCode):
+        """
+        This function attempts to remove $targetId membership from Group
+        using $userId permissions
+        DELETE /group-member/{targetId}/
+        """
+        if userId >= 0:
+            self.client.force_authenticate(user=self.users[userId])
+        response = self.client.delete(self.member_url % targetId)
+        self.assertEqual(response.status_code, expectedHttpResponseCode)
+
+    # /rank
+    def test_rank_not_authed(self):
+        self.try_rank(-1, 1, 4, status.HTTP_401_UNAUTHORIZED)
+
+    def test_rank_not_group_member(self):
+        self.try_rank(0, 1, 4, status.HTTP_404_NOT_FOUND)
+
+    def test_rank_demote_no_permission(self):
+        self.try_rank(1, 2, 1, status.HTTP_403_FORBIDDEN)
+
+    def test_rank_demote_no_permission2(self):
+        self.try_rank(3, 2, 1, status.HTTP_403_FORBIDDEN)
+        self.assertEqual(GroupMember.objects.get(pk=self.mships[2].id).perm_rank, 2)
+
+    def test_rank_promote_no_permission(self):
+        self.try_rank(1, 1, 2, status.HTTP_403_FORBIDDEN)
+
+    def test_rank_no_rank_change(self):
+        self.try_rank(5, 1, 1, status.HTTP_400_BAD_REQUEST)
+
+    def test_rank_not_a_number(self):
+        self.try_rank(5, 1, "hi!", status.HTTP_400_BAD_REQUEST)
+
+    def test_rank_set_rank_to_zero(self):
+        self.try_rank(5, 1, 0, status.HTTP_400_BAD_REQUEST)
+
+    def test_rank_bad_rank(self):
+        self.try_rank(5, 1, -1, status.HTTP_400_BAD_REQUEST)
+        self.try_rank(5, 1, Group.ADMINISTRATOR_RANK + 1, status.HTTP_400_BAD_REQUEST)
+
+    def test_rank_too_high_rank(self):
+        self.try_rank(5, 1, 5, status.HTTP_403_FORBIDDEN)
+
+    def test_rank_promote_ok(self):
+        self.try_rank(3, 1, 2, status.HTTP_200_OK)
+        self.assertEqual(GroupMember.objects.get(pk=self.mships[1].id).perm_rank, 2)
+
+    def test_rank_demote_ok(self):
+        self.try_rank(5, 3, 2, status.HTTP_200_OK)
+        self.assertEqual(GroupMember.objects.get(pk=self.mships[3].id).perm_rank, 2)
+
+    def test_rank_demote_self_ok(self):
+        self.try_rank(3, 3, 2, status.HTTP_200_OK)
+        self.assertEqual(GroupMember.objects.get(pk=self.mships[3].id).perm_rank, 2)
+
+    # delete
+    def test_delete_not_authed(self):
+        self.try_delete(-1, 1, status.HTTP_401_UNAUTHORIZED)
+
+    def test_delete_not_group_member(self):
+        self.try_delete(0, 1, status.HTTP_404_NOT_FOUND)
+
+    def test_delete_no_permission(self):
+        self.try_delete(3, 2, status.HTTP_403_FORBIDDEN)
+
+    def test_delete_user_not_in_group(self):
+        self.try_delete(4, 0, status.HTTP_404_NOT_FOUND)
+
+    def test_delete_self_ok(self):
+        self.try_delete(1, 1, status.HTTP_204_NO_CONTENT)
+        self.assertFalse(GroupMember.objects.filter(pk=self.mships[1].id).exists())
+
+    def test_delete_ok(self):
+        self.try_delete(5, 1, status.HTTP_204_NO_CONTENT)
+        self.assertFalse(GroupMember.objects.filter(pk=self.mships[1].id).exists())
 
 class OpenGroupMemberCreationTests(APITestCase):
     @classmethod

sigma_core/views/group_member.py

@@ -24,16 +24,66 @@ def create(self, request):
         mem = serializer.save()
         return Response(serializer.data, status=status.HTTP_201_CREATED)
 
-    # def update(self, request, pk=None):
-    #     pass
-    #
-    # @decorators.detail_route(methods=['put'])
-    # def promote(self, request, pk=None):
-    #     pass
-    #
-    # @decorators.detail_route(methods=['put'])
-    # def demote(self, request, pk=None):
-    #     pass
+    def destroy(self, request, pk=None):
+        from sigma_core.models.group import Group
+        try:
+            modified_mship = GroupMember.objects.all().select_related('group').get(pk=pk)
+            group = modified_mship.group
+            my_mship = GroupMember.objects.all().get(group=group, user=request.user)
+        except GroupMember.DoesNotExist:
+            raise Http404()
+
+        # You can always quit the Group (ie destroy YOUR membership)
+        if my_mship.id != modified_mship.id:
+            # Can't modify someone higher than you
+            if my_mship.perm_rank <= modified_mship.perm_rank:
+                return Response(status=status.HTTP_403_FORBIDDEN)
+
+            # Check permission
+            if group.req_rank_kick > my_mship.perm_rank:
+                return Response(status=status.HTTP_403_FORBIDDEN)
+
+        modified_mship.delete()
+        return Response(status=status.HTTP_204_NO_CONTENT)
+
+    @decorators.detail_route(methods=['put'])
+    def rank(self, request, pk=None):
+        from sigma_core.models.group import Group
+        try:
+            modified_mship = GroupMember.objects.all().select_related('group').get(pk=pk)
+            group = modified_mship.group
+            my_mship = GroupMember.objects.all().get(group=group, user=request.user)
+        except GroupMember.DoesNotExist:
+            raise Http404()
+
+        perm_rank_new = request.data.get('perm_rank', None)
+
+        try:
+            if perm_rank_new > Group.ADMINISTRATOR_RANK or perm_rank_new < 1 or perm_rank_new == modified_mship.perm_rank:
+                return Response(status=status.HTTP_400_BAD_REQUEST)
+        except TypeError:
+            return Response(status=status.HTTP_400_BAD_REQUEST)
+
+        # You can demote yourself:
+        demote_self = modified_mship.id == my_mship.id and perm_rank_new < my_mship.perm_rank
+
+        # Can't modify someone higher than you, or set rank to higher than you
+        if (my_mship.perm_rank <= modified_mship.perm_rank or my_mship.perm_rank <= perm_rank_new) and not demote_self:
+            return Response(status=status.HTTP_403_FORBIDDEN)
+
+        # promote
+        if perm_rank_new > modified_mship.perm_rank:
+            if group.req_rank_promote > my_mship.perm_rank:
+                return Response(status=status.HTTP_403_FORBIDDEN)
+        # demote
+        else:
+            if group.req_rank_demote > my_mship.perm_rank and not demote_self:
+                return Response(status=status.HTTP_403_FORBIDDEN)
+
+        modified_mship.perm_rank = perm_rank_new
+        modified_mship.save()
+
+        return Response(GroupMemberSerializer(modified_mship).data, status=status.HTTP_200_OK)
 
     @decorators.detail_route(methods=['put'])
     def accept_join_request(self, request, pk=None):
@@ -52,8 +102,3 @@ def accept_join_request(self, request, pk=None):
 
         s = GroupMemberSerializer(gm)
         return Response(s.data, status=status.HTTP_200_OK)
-
-    # @decorators.detail_route(methods=['put'])
-    # def kick(self, request, pk=None):
-    #     pass
-    #