feat: Add public API endpoint to fetch user sessions (#840)

api/public/v2/owner/serializers.py

@@ -21,3 +21,13 @@ class UserSerializer(OwnerSerializer):
     class Meta:
         model = Owner
         fields = OwnerSerializer.Meta.fields + ("activated", "is_admin", "email")
+
+
+class UserSessionSerializer(serializers.ModelSerializer):
+    has_active_session = serializers.BooleanField()
+    expiry_date = serializers.DateTimeField()
+
+    class Meta:
+        model = Owner
+        fields = ("username", "name", "has_active_session", "expiry_date")
+        read_only_fields = fields

api/public/v2/owner/views.py

@@ -1,14 +1,22 @@
-from django.db.models import Q
+from typing import Any
+
+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.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.shared.owner.mixins import OwnerViewSetMixin, UserViewSetMixin
+from api.shared.owner.mixins import (
+    OwnerViewSetMixin,
+    UserSessionViewSetMixin,
+    UserViewSetMixin,
+)
 from codecov_auth.models import Owner, Service
 
-from .serializers import OwnerSerializer, UserSerializer
+from .serializers import OwnerSerializer, UserSerializer, UserSessionSerializer
 
 
 @extend_schema(parameters=owner_parameters, tags=["Users"])
@@ -19,7 +27,7 @@ class OwnerViewSet(
     queryset = Owner.objects.none()
 
     @extend_schema(summary="Owner detail")
-    def retrieve(self, request, *args, **kwargs):
+    def retrieve(self, request: Request, *args: Any, **kwargs: Any) -> Owner:
         """
         Returns a single owner by name
         """
@@ -32,20 +40,33 @@ class UserViewSet(UserViewSetMixin, mixins.ListModelMixin, mixins.RetrieveModelM
     queryset = Owner.objects.none()
 
     @extend_schema(summary="User list")
-    def list(self, request, *args, **kwargs):
+    def list(self, request: Request, *args: Any, **kwargs: Any) -> Response:
         """
         Returns a paginated list of users for the specified owner (org)
         """
         return super().list(request, *args, **kwargs)
 
     @extend_schema(summary="User detail")
-    def retrieve(self, request, *args, **kwargs):
+    def retrieve(self, request: Request, *args: Any, **kwargs: Any) -> Owner:
         """
         Returns a user for the specified owner_username or ownerid
         """
         return super().retrieve(request, *args, **kwargs)
 
 
+@extend_schema(parameters=owner_parameters, tags=["Users"])
+class UserSessionViewSet(UserSessionViewSetMixin, mixins.ListModelMixin):
+    serializer_class = UserSessionSerializer
+    queryset = Owner.objects.none()
+
+    @extend_schema(summary="User session list")
+    def list(self, request: Request, *args: Any, **kwargs: Any) -> Response:
+        """
+        Returns a paginated list of users' login session for the specified owner (org)
+        """
+        return super().list(request, *args, **kwargs)
+
+
 @extend_schema(
     parameters=[
         service_parameter,
@@ -56,7 +77,7 @@ class OwnersViewSet(viewsets.GenericViewSet, mixins.ListModelMixin):
     serializer_class = OwnerSerializer
     permission_classes = [IsAuthenticated]
 
-    def get_queryset(self):
+    def get_queryset(self) -> QuerySet:
         service = self.kwargs.get("service")
         try:
             Service(service)
@@ -70,7 +91,7 @@ def get_queryset(self):
         )
 
     @extend_schema(summary="Service owners")
-    def list(self, request, *args, **kwargs):
+    def list(self, request: Request, *args: Any, **kwargs: Any) -> Response:
         """
         Returns all owners to which the currently authenticated user has access
         """

api/public/v2/tests/test_api_owner_viewset.py

@@ -1,9 +1,16 @@
+from datetime import timedelta
+
+from django.utils import timezone
 from rest_framework import status
 from rest_framework.exceptions import ErrorDetail
 from rest_framework.reverse import reverse
 from rest_framework.test import APITestCase
 
-from codecov_auth.tests.factories import OwnerFactory
+from codecov_auth.tests.factories import (
+    DjangoSessionFactory,
+    OwnerFactory,
+    SessionFactory,
+)
 from utils.test_utils import APIClient
 
 
@@ -174,3 +181,160 @@ def test_retrieve_cannot_get_details_if_not_member_of_org(self):
             "is_admin": False,
             "email": another_user.email,
         }
+
+
+class UserSessionViewSetTests(APITestCase):
+    def _list(self, kwargs):
+        return self.client.get(reverse("api-v2-user-sessions-list", kwargs=kwargs))
+
+    def setUp(self):
+        self.org = OwnerFactory(service="github")
+        self.admin_owner = OwnerFactory(service="github", organizations=[self.org.pk])
+        self.org.admins = [self.admin_owner.pk]
+        self.org.save()
+        self.client = APIClient()
+
+    def test_not_part_of_org(self):
+        self.current_owner = OwnerFactory(service="github", organizations=[])
+        self.client.force_login_owner(self.current_owner)
+
+        response = self._list(
+            kwargs={"service": self.org.service, "owner_username": self.org.username}
+        )
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+    def test_not_admin_of_org(self):
+        self.not_in_org_owner = OwnerFactory(
+            service="github", organizations=[self.org.pk]
+        )
+        self.client.force_login_owner(self.not_in_org_owner)
+
+        response = self._list(
+            kwargs={"service": self.org.service, "owner_username": self.org.username}
+        )
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+    def test_no_sessions(self):
+        self.client.force_login_owner(self.admin_owner)
+
+        response = self._list(
+            kwargs={"service": self.org.service, "owner_username": self.org.username}
+        )
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {
+            "count": 1,
+            "next": None,
+            "previous": None,
+            "results": [
+                {
+                    "username": self.admin_owner.username,
+                    "name": self.admin_owner.name,
+                    "has_active_session": False,
+                    "expiry_date": None,
+                }
+            ],
+            "total_pages": 1,
+        }
+
+    def test_has_active_session(self):
+        expiry_date = timezone.now() + timedelta(days=1)
+        expiry_date_response = str(expiry_date).replace(" ", "T").replace("+00:00", "Z")
+
+        self.session = SessionFactory(
+            owner=self.admin_owner,
+            login_session=DjangoSessionFactory(expire_date=expiry_date),
+        )
+        self.client.force_login_owner(self.admin_owner)
+
+        response = self._list(
+            kwargs={"service": self.org.service, "owner_username": self.org.username}
+        )
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {
+            "count": 1,
+            "next": None,
+            "previous": None,
+            "results": [
+                {
+                    "username": self.admin_owner.username,
+                    "name": self.admin_owner.name,
+                    "has_active_session": True,
+                    "expiry_date": expiry_date_response,
+                }
+            ],
+            "total_pages": 1,
+        }
+
+    def test_multiple_sessions_one(self):
+        expiry_date = timezone.now() + timedelta(days=1)
+        expiry_date_response = str(expiry_date).replace(" ", "T").replace("+00:00", "Z")
+
+        self.session_1 = SessionFactory(
+            owner=self.admin_owner,
+            login_session=DjangoSessionFactory(expire_date=expiry_date),
+        )
+        self.session_2 = SessionFactory(
+            owner=self.admin_owner,
+            login_session=DjangoSessionFactory(
+                expire_date=timezone.now() - timedelta(days=1)
+            ),
+        )
+        self.client.force_login_owner(self.admin_owner)
+
+        response = self._list(
+            kwargs={"service": self.org.service, "owner_username": self.org.username}
+        )
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {
+            "count": 1,
+            "next": None,
+            "previous": None,
+            "results": [
+                {
+                    "username": self.admin_owner.username,
+                    "name": self.admin_owner.name,
+                    "has_active_session": True,
+                    "expiry_date": expiry_date_response,
+                }
+            ],
+            "total_pages": 1,
+        }
+
+    def test_multiple_sessions_two(self):
+        expiry_date = timezone.now()
+        expiry_date_response = str(expiry_date).replace(" ", "T").replace("+00:00", "Z")
+
+        self.session_1 = SessionFactory(
+            owner=self.admin_owner,
+            login_session=DjangoSessionFactory(expire_date=expiry_date),
+        )
+        self.session_2 = SessionFactory(
+            owner=self.admin_owner,
+            login_session=DjangoSessionFactory(
+                expire_date=timezone.now() - timedelta(days=1)
+            ),
+        )
+        self.client.force_login_owner(self.admin_owner)
+
+        response = self._list(
+            kwargs={"service": self.org.service, "owner_username": self.org.username}
+        )
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {
+            "count": 1,
+            "next": None,
+            "previous": None,
+            "results": [
+                {
+                    "username": self.admin_owner.username,
+                    "name": self.admin_owner.name,
+                    "has_active_session": False,
+                    "expiry_date": expiry_date_response,
+                }
+            ],
+            "total_pages": 1,
+        }

api/public/v2/urls.py

@@ -12,7 +12,7 @@
 from .component.views import ComponentViewSet
 from .coverage.views import CoverageViewSet, FlagCoverageViewSet
 from .flag.views import FlagViewSet
-from .owner.views import OwnersViewSet, OwnerViewSet, UserViewSet
+from .owner.views import OwnersViewSet, OwnerViewSet, UserSessionViewSet, UserViewSet
 from .pull.views import PullViewSet
 from .repo.views import RepositoryConfigView, RepositoryViewSet
 from .report.views import FileReportViewSet, ReportViewSet, TotalsViewSet
@@ -26,6 +26,9 @@
 
 owner_artifacts_router = OptionalTrailingSlashRouter()
 owner_artifacts_router.register(r"users", UserViewSet, basename="api-v2-users")
+owner_artifacts_router.register(
+    r"user-sessions", UserSessionViewSet, basename="api-v2-user-sessions"
+)
 
 repository_router = OptionalTrailingSlashRouter()
 repository_router.register(r"repos", RepositoryViewSet, basename="api-v2-repos")

api/shared/owner/mixins.py

@@ -1,11 +1,12 @@
-from django.db.models import Q
+from django.db.models import BooleanField, Case, Max, Q, QuerySet, When
 from django.shortcuts import get_object_or_404
+from django.utils import timezone
 from django_filters import rest_framework as django_filters
 from rest_framework import filters, viewsets
 from rest_framework.exceptions import NotFound
 
 from api.shared.mixins import OwnerPropertyMixin
-from api.shared.permissions import MemberOfOrgPermissions
+from api.shared.permissions import MemberOfOrgPermissions, UserIsAdminPermissions
 from codecov_auth.models import Owner, Service
 
 from .filters import UserFilters
@@ -15,15 +16,15 @@ class OwnerViewSetMixin(viewsets.GenericViewSet):
     lookup_field = "owner_username"
     lookup_value_regex = "[^/]+"
 
-    def get_queryset(self):
+    def get_queryset(self) -> QuerySet:
         service = self.kwargs.get("service")
         try:
             Service(service)
         except ValueError:
             raise NotFound(f"Service not found: {service}")
         return Owner.objects.filter(service=self.kwargs.get("service"))
 
-    def get_object(self):
+    def get_object(self) -> Owner:
         return get_object_or_404(
             self.get_queryset(),
             username=self.kwargs.get("owner_username"),
@@ -45,14 +46,14 @@ class UserViewSetMixin(
     lookup_field = "user_username_or_ownerid"
     search_fields = ["name", "username", "email"]
 
-    def get_queryset(self):
+    def get_queryset(self) -> QuerySet:
         return (
             Owner.objects.users_of(owner=self.owner)
             .annotate_activated_in(owner=self.owner)
             .annotate_is_admin_in(owner=self.owner)
         )
 
-    def get_object(self):
+    def get_object(self) -> Owner:
         username_or_ownerid = self.kwargs.get("user_username_or_ownerid")
         try:
             ownerid = int(username_or_ownerid)
@@ -63,3 +64,22 @@ def get_object(self):
             self.get_queryset(),
             (Q(username=username_or_ownerid) | Q(ownerid=ownerid)),
         )
+
+
+class UserSessionViewSetMixin(
+    viewsets.GenericViewSet,
+    OwnerPropertyMixin,
+):
+    permission_classes = [UserIsAdminPermissions]
+    ordering_fields = ("name", "username")
+
+    def get_queryset(self) -> QuerySet:
+        return Owner.objects.users_of(owner=self.owner).annotate(
+            expiry_date=Max("session__login_session__expire_date"),
+            has_active_session=Case(
+                When(expiry_date__isnull=True, then=False),
+                When(expiry_date__gt=timezone.now(), then=True),
+                default=False,
+                output_field=BooleanField(),
+            ),
+        )