v2 API: add endpoint to activate and deactivate users

api/public/v2/owner/serializers.py

@@ -31,3 +31,11 @@ class Meta:
         model = Owner
         fields = ("username", "name", "has_active_session", "expiry_date")
         read_only_fields = fields
+
+
+class UserUpdateActivationSerializer(serializers.ModelSerializer):
+    activated = serializers.BooleanField()
+
+    class Meta:
+        model = Owner
+        fields = ("activated",)

api/public/v2/owner/views.py

@@ -3,20 +3,28 @@
 from django.db.models import Q, QuerySet
 from drf_spectacular.utils import extend_schema
 from rest_framework import mixins, viewsets
-from rest_framework.exceptions import NotFound
+from rest_framework.exceptions import NotFound, PermissionDenied
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.request import Request
 from rest_framework.response import Response
 
-from api.public.v2.schema import owner_parameters, service_parameter
+from api.public.v2.schema import (
+    owner_parameters,
+    service_parameter,
+)
 from api.shared.owner.mixins import (
     OwnerViewSetMixin,
     UserSessionViewSetMixin,
     UserViewSetMixin,
 )
 from codecov_auth.models import Owner, Service
 
-from .serializers import OwnerSerializer, UserSerializer, UserSessionSerializer
+from .serializers import (
+    OwnerSerializer,
+    UserSerializer,
+    UserSessionSerializer,
+    UserUpdateActivationSerializer,
+)
 
 
 @extend_schema(parameters=owner_parameters, tags=["Users"])
@@ -53,6 +61,34 @@ def retrieve(self, request: Request, *args: Any, **kwargs: Any) -> Owner:
         """
         return super().retrieve(request, *args, **kwargs)
 
+    @extend_schema(summary="Update a user", request=UserUpdateActivationSerializer)
+    def partial_update(self, request: Request, *args: Any, **kwargs: Any) -> Response:
+        """
+        Updates a user for the specified owner_username or ownerid
+
+        Allowed fields
+          - activated: boolean value to activate or deactivate the use
+        """
+        instance = self.get_object()
+        serializer = UserUpdateActivationSerializer(
+            instance,
+            data=request.data,
+        )
+        serializer.is_valid(raise_exception=True)
+
+        if serializer.validated_data["activated"]:
+            if self.owner.can_activate_user(instance):
+                print("can activate")
+                self.owner.activate_user(instance)
+            else:
+                raise PermissionDenied(
+                    f"Cannot activate user {self.owner.username} -- not enough seats left."
+                )
+        else:
+            self.owner.deactivate_user(instance)
+
+        return super().retrieve(request, *args, **kwargs)
+
 
 @extend_schema(parameters=owner_parameters, tags=["Users"])
 class UserSessionViewSet(UserSessionViewSetMixin, mixins.ListModelMixin):

api/public/v2/tests/test_api_owner_viewset.py

@@ -63,6 +63,11 @@ def _list(self, kwargs):
     def _detail(self, kwargs):
         return self.client.get(reverse("api-v2-users-detail", kwargs=kwargs))
 
+    def _patch(self, kwargs, data):
+        return self.client.patch(
+            reverse("api-v2-users-detail", kwargs=kwargs), data=data
+        )
+
     def setUp(self):
         self.org = OwnerFactory(service="github")
         self.current_owner = OwnerFactory(service="github", organizations=[self.org.pk])
@@ -179,6 +184,302 @@ def test_retrieve_cannot_get_details_if_not_member_of_org(self):
             "email": another_user.email,
         }
 
+    def test_update_activate_by_username(self):
+        another_user = OwnerFactory(service="github", organizations=[self.org.pk])
+
+        # Activate user
+        response = self._patch(
+            kwargs={
+                "service": self.org.service,
+                "owner_username": self.org.username,
+                "user_username_or_ownerid": another_user.username,
+            },
+            data={"activated": True},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {
+            "service": "github",
+            "username": another_user.username,
+            "name": another_user.name,
+            "activated": True,
+            "is_admin": False,
+            "email": another_user.email,
+        }
+
+        # Deactivate user
+        response = self._patch(
+            kwargs={
+                "service": self.org.service,
+                "owner_username": self.org.username,
+                "user_username_or_ownerid": another_user.username,
+            },
+            data={"activated": False},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {
+            "service": "github",
+            "username": another_user.username,
+            "name": another_user.name,
+            "activated": False,
+            "is_admin": False,
+            "email": another_user.email,
+        }
+
+    def test_update_activate_by_ownerid(self):
+        another_user = OwnerFactory(service="github", organizations=[self.org.pk])
+
+        # Activate user
+        response = self._patch(
+            kwargs={
+                "service": self.org.service,
+                "owner_username": self.org.username,
+                "user_username_or_ownerid": another_user.ownerid,
+            },
+            data={"activated": True},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {
+            "service": "github",
+            "username": another_user.username,
+            "name": another_user.name,
+            "activated": True,
+            "is_admin": False,
+            "email": another_user.email,
+        }
+
+        # Deactivate user
+        response = self._patch(
+            kwargs={
+                "service": self.org.service,
+                "owner_username": self.org.username,
+                "user_username_or_ownerid": another_user.ownerid,
+            },
+            data={"activated": False},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {
+            "service": "github",
+            "username": another_user.username,
+            "name": another_user.name,
+            "activated": False,
+            "is_admin": False,
+            "email": another_user.email,
+        }
+
+    def test_update_activate_unauthorized_members_of_other_orgs(self):
+        another_org = OwnerFactory(service="github")
+        another_user = OwnerFactory(service="github", organizations=[another_org.pk])
+
+        # Activate user - not allowed
+        response = self._patch(
+            kwargs={
+                "service": self.org.service,
+                "owner_username": self.org.username,
+                "user_username_or_ownerid": another_user.username,
+            },
+            data={"activated": True},
+        )
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+
+        # Deactivate user - not allowed
+        response = self._patch(
+            kwargs={
+                "service": self.org.service,
+                "owner_username": self.org.username,
+                "user_username_or_ownerid": another_user.username,
+            },
+            data={"activated": False},
+        )
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+
+        # Request allowed after user joins the org
+        another_user.organizations.append(self.org.pk)
+        another_user.save()
+
+        # Activate user
+        response = self._patch(
+            kwargs={
+                "service": self.org.service,
+                "owner_username": self.org.username,
+                "user_username_or_ownerid": another_user.username,
+            },
+            data={"activated": True},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {
+            "service": "github",
+            "username": another_user.username,
+            "name": another_user.name,
+            "activated": True,
+            "is_admin": False,
+            "email": another_user.email,
+        }
+
+        # Deactivate user
+        response = self._patch(
+            kwargs={
+                "service": self.org.service,
+                "owner_username": self.org.username,
+                "user_username_or_ownerid": another_user.username,
+            },
+            data={"activated": False},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {
+            "service": "github",
+            "username": another_user.username,
+            "name": another_user.name,
+            "activated": False,
+            "is_admin": False,
+            "email": another_user.email,
+        }
+
+    def test_update_activate_unauthorized_not_member_of_org(self):
+        another_org = OwnerFactory(service="github")
+        another_user = OwnerFactory(service="github", organizations=[another_org.pk])
+
+        # Activate user - not allowed
+        response = self._patch(
+            kwargs={
+                "service": another_org.service,
+                "owner_username": another_org.username,
+                "user_username_or_ownerid": another_user.username,
+            },
+            data={"activated": True},
+        )
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+
+        # Deactivate user - not allowed
+        response = self._patch(
+            kwargs={
+                "service": another_org.service,
+                "owner_username": another_org.username,
+                "user_username_or_ownerid": another_user.username,
+            },
+            data={"activated": False},
+        )
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+
+        # Request owner now joins the other org and thus is allowed to activate/deactivate
+        self.current_owner.organizations.append(another_org.pk)
+        self.current_owner.save()
+
+        # Activate user
+        response = self._patch(
+            kwargs={
+                "service": another_org.service,
+                "owner_username": another_org.username,
+                "user_username_or_ownerid": another_user.username,
+            },
+            data={"activated": True},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {
+            "service": "github",
+            "username": another_user.username,
+            "name": another_user.name,
+            "activated": True,
+            "is_admin": False,
+            "email": another_user.email,
+        }
+
+        # Deactivate user
+        response = self._patch(
+            kwargs={
+                "service": another_org.service,
+                "owner_username": another_org.username,
+                "user_username_or_ownerid": another_user.username,
+            },
+            data={"activated": False},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {
+            "service": "github",
+            "username": another_user.username,
+            "name": another_user.name,
+            "activated": False,
+            "is_admin": False,
+            "email": another_user.email,
+        }
+
+    def test_update_activate_no_seats_left(self):
+        another_user = OwnerFactory(service="github", organizations=[self.org.pk])
+        another_user_2 = OwnerFactory(service="github", organizations=[self.org.pk])
+
+        # Activate user 1
+        response = self._patch(
+            kwargs={
+                "service": self.org.service,
+                "owner_username": self.org.username,
+                "user_username_or_ownerid": another_user.username,
+            },
+            data={"activated": True},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {
+            "service": "github",
+            "username": another_user.username,
+            "name": another_user.name,
+            "activated": True,
+            "is_admin": False,
+            "email": another_user.email,
+        }
+
+        # Activate user 2
+        response = self._patch(
+            kwargs={
+                "service": self.org.service,
+                "owner_username": self.org.username,
+                "user_username_or_ownerid": another_user_2.username,
+            },
+            data={"activated": True},
+        )
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+        assert response.data == {
+            "detail": ErrorDetail(
+                string=f"Cannot activate user {self.org.username} -- not enough seats left.",
+                code="permission_denied",
+            )
+        }
+
+        # Deactivate user 1 to make room for user 2
+        response = self._patch(
+            kwargs={
+                "service": self.org.service,
+                "owner_username": self.org.username,
+                "user_username_or_ownerid": another_user.username,
+            },
+            data={"activated": False},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {
+            "service": "github",
+            "username": another_user.username,
+            "name": another_user.name,
+            "activated": False,
+            "is_admin": False,
+            "email": another_user.email,
+        }
+
+        # Activate user 2 now that there's room
+        response = self._patch(
+            kwargs={
+                "service": self.org.service,
+                "owner_username": self.org.username,
+                "user_username_or_ownerid": another_user_2.username,
+            },
+            data={"activated": True},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {
+            "service": "github",
+            "username": another_user_2.username,
+            "name": another_user_2.name,
+            "activated": True,
+            "is_admin": False,
+            "email": another_user_2.email,
+        }
+
 
 class UserSessionViewSetTests(APITestCase):
     def _list(self, kwargs):