feat: [AXM-33] create enrollments filtering by course completion statuses

common/djangoapps/student/models/course_enrollment.py

@@ -129,11 +129,71 @@ class UnenrollmentNotAllowed(CourseEnrollmentException):
     pass
 
 
+class CourseEnrollmentQuerySet(models.QuerySet):
+    """
+    Custom queryset for CourseEnrollment with Table-level filter methods.
+    """
+
+    def active(self):
+        """
+        Returns a queryset of CourseEnrollment objects for courses that are currently active.
+        """
+        return self.filter(is_active=True)
+
+    def without_certificates(self, user_username):
+        """
+        Returns a queryset of CourseEnrollment objects for courses that do not have a certificate.
+        """
+        from lms.djangoapps.certificates.models import GeneratedCertificate  # pylint: disable=import-outside-toplevel
+        course_ids_with_certificates = GeneratedCertificate.objects.filter(
+            user__username=user_username
+        ).values_list('course_id', flat=True)
+        return self.exclude(course_id__in=course_ids_with_certificates)
+
+    def with_certificates(self, user_username):
+        """
+        Returns a queryset of CourseEnrollment objects for courses that have a certificate.
+        """
+        from lms.djangoapps.certificates.models import GeneratedCertificate  # pylint: disable=import-outside-toplevel
+        course_ids_with_certificates = GeneratedCertificate.objects.filter(
+            user__username=user_username
+        ).values_list('course_id', flat=True)
+        return self.filter(course_id__in=course_ids_with_certificates)
+
+    def in_progress(self, user_username, time_zone=UTC):
+        """
+        Returns a queryset of CourseEnrollment objects for courses that are currently in progress.
+        """
+        now = datetime.now(time_zone)
+        return self.active().without_certificates(user_username).filter(
+            Q(course__start__lte=now, course__end__gte=now)
+            | Q(course__start__isnull=True, course__end__isnull=True)
+            | Q(course__start__isnull=True, course__end__gte=now)
+            | Q(course__start__lte=now, course__end__isnull=True),
+        )
+
+    def completed(self, user_username):
+        """
+        Returns a queryset of CourseEnrollment objects for courses that have been completed.
+        """
+        return self.active().with_certificates(user_username)
+
+    def expired(self, user_username, time_zone=UTC):
+        """
+        Returns a queryset of CourseEnrollment objects for courses that have expired.
+        """
+        now = datetime.now(time_zone)
+        return self.active().without_certificates(user_username).filter(course__end__lt=now)
+
+
 class CourseEnrollmentManager(models.Manager):
     """
     Custom manager for CourseEnrollment with Table-level filter methods.
     """
 
+    def get_queryset(self):
+        return CourseEnrollmentQuerySet(self.model, using=self._db)
+
     def is_small_course(self, course_id):
         """
         Returns false if the number of enrollments are one greater than 'max_enrollments' else true

lms/djangoapps/mobile_api/users/enums.py

@@ -0,0 +1,22 @@
+"""
+Enums for mobile_api users app.
+"""
+from enum import Enum
+
+
+class EnrollmentStatuses(Enum):
+    """
+    Enum for enrollment statuses.
+    """
+
+    ALL = 'all'
+    IN_PROGRESS = 'in_progress'
+    COMPLETED = 'completed'
+    EXPIRED = 'expired'
+
+    @classmethod
+    def values(cls):
+        """
+        Returns string representation of all enum values.
+        """
+        return [e.value for e in cls]

lms/djangoapps/mobile_api/users/serializers.py

@@ -110,7 +110,7 @@ def get_audit_access_expires(self, model):
         """
         Returns expiration date for a course audit expiration, if any or null
         """
-        return get_user_course_expiration_date(model.user, model.course)
+        return get_user_course_expiration_date(model.user, model.course, model)
 
     def get_certificate(self, model):
         """Returns the information about the user's certificate in the course."""

lms/djangoapps/mobile_api/users/tests.py

@@ -36,6 +36,7 @@
     MobileAuthUserTestMixin,
     MobileCourseAccessTestMixin
 )
+from lms.djangoapps.mobile_api.users.enums import EnrollmentStatuses
 from lms.djangoapps.mobile_api.utils import API_V1, API_V05, API_V2, API_V3, API_V4
 from openedx.core.lib.courses import course_image_url
 from openedx.core.djangoapps.discussions.models import DiscussionsConfiguration
@@ -706,6 +707,201 @@ def test_course_status_in_primary_obj_when_student_have_progress(
         self.assertEqual(response.data['primary']['course_status'], expected_course_status)
         get_last_completed_block_mock.assert_called_once_with(self.user, course.id)
 
+    @patch('lms.djangoapps.mobile_api.users.serializers.cache.set', return_value=None)
+    def test_user_enrollment_api_v4_in_progress_status(self, cache_mock: MagicMock):
+        """
+        Testing
+        """
+        self.login()
+        old_course = CourseFactory.create(
+            org="edx",
+            mobile_available=True,
+            start=self.THREE_YEARS_AGO,
+            end=self.LAST_WEEK
+        )
+        actual_course = CourseFactory.create(
+            org="edx",
+            mobile_available=True,
+            start=self.LAST_WEEK,
+            end=self.NEXT_WEEK
+        )
+        infinite_course = CourseFactory.create(
+            org="edx",
+            mobile_available=True,
+            start=self.LAST_WEEK,
+            end=None
+        )
+
+        self.enroll(old_course.id)
+        self.enroll(actual_course.id)
+        self.enroll(infinite_course.id)
+
+        response = self.api_response(api_version=API_V4, data={'status': EnrollmentStatuses.IN_PROGRESS.value})
+        enrollments = response.data['enrollments']
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(enrollments['count'], 2)
+        self.assertEqual(enrollments['results'][1]['course']['id'], str(actual_course.id))
+        self.assertEqual(enrollments['results'][0]['course']['id'], str(infinite_course.id))
+        self.assertNotIn('primary', response.data)
+
+    def test_user_enrollment_api_v4_completed_status(self):
+        """
+        Testing
+        """
+        self.login()
+        old_course = CourseFactory.create(
+            org="edx",
+            mobile_available=True,
+            start=self.THREE_YEARS_AGO,
+            end=self.LAST_WEEK
+        )
+        actual_course = CourseFactory.create(
+            org="edx",
+            mobile_available=True,
+            start=self.LAST_WEEK,
+            end=self.NEXT_WEEK
+        )
+        infinite_course = CourseFactory.create(
+            org="edx",
+            mobile_available=True,
+            start=self.LAST_WEEK,
+            end=None
+        )
+        GeneratedCertificateFactory.create(
+            user=self.user,
+            course_id=infinite_course.id,
+            status=CertificateStatuses.downloadable,
+            mode='verified',
+        )
+
+        self.enroll(old_course.id)
+        self.enroll(actual_course.id)
+        self.enroll(infinite_course.id)
+
+        response = self.api_response(api_version=API_V4, data={'status': EnrollmentStatuses.COMPLETED.value})
+        enrollments = response.data['enrollments']
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(enrollments['count'], 1)
+        self.assertEqual(enrollments['results'][0]['course']['id'], str(infinite_course.id))
+        self.assertNotIn('primary', response.data)
+
+    def test_user_enrollment_api_v4_expired_status(self):
+        """
+        Testing
+        """
+        self.login()
+        old_course = CourseFactory.create(
+            org="edx",
+            mobile_available=True,
+            start=self.THREE_YEARS_AGO,
+            end=self.LAST_WEEK
+        )
+        actual_course = CourseFactory.create(
+            org="edx",
+            mobile_available=True,
+            start=self.LAST_WEEK,
+            end=self.NEXT_WEEK
+        )
+        infinite_course = CourseFactory.create(
+            org="edx",
+            mobile_available=True,
+            start=self.LAST_WEEK,
+            end=None
+        )
+        self.enroll(old_course.id)
+        self.enroll(actual_course.id)
+        self.enroll(infinite_course.id)
+
+        response = self.api_response(api_version=API_V4, data={'status': EnrollmentStatuses.EXPIRED.value})
+        enrollments = response.data['enrollments']
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(enrollments['count'], 1)
+        self.assertEqual(enrollments['results'][0]['course']['id'], str(old_course.id))
+        self.assertNotIn('primary', response.data)
+
+    def test_user_enrollment_api_v4_expired_course_with_certificate(self):
+        """
+        Testing that the API returns a course with
+        an expiration date in the past if the user has a certificate for this course.
+        """
+        self.login()
+        expired_course = CourseFactory.create(
+            org="edx",
+            mobile_available=True,
+            start=self.THREE_YEARS_AGO,
+            end=self.LAST_WEEK
+        )
+        expired_course_with_cert = CourseFactory.create(
+            org="edx",
+            mobile_available=True,
+            start=self.THREE_YEARS_AGO,
+            end=self.LAST_WEEK
+        )
+        GeneratedCertificateFactory.create(
+            user=self.user,
+            course_id=expired_course_with_cert.id,
+            status=CertificateStatuses.downloadable,
+            mode='verified',
+        )
+
+        self.enroll(expired_course_with_cert.id)
+        self.enroll(expired_course.id)
+
+        response = self.api_response(api_version=API_V4, data={'status': EnrollmentStatuses.COMPLETED.value})
+        enrollments = response.data['enrollments']
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(enrollments['count'], 1)
+        self.assertEqual(enrollments['results'][0]['course']['id'], str(expired_course_with_cert.id))
+        self.assertNotIn('primary', response.data)
+
+    def test_user_enrollment_api_v4_status_all(self):
+        """
+        Testing
+        """
+        self.login()
+        old_course = CourseFactory.create(
+            org="edx",
+            mobile_available=True,
+            start=self.THREE_YEARS_AGO,
+            end=self.LAST_WEEK
+        )
+        actual_course = CourseFactory.create(
+            org="edx",
+            mobile_available=True,
+            start=self.LAST_WEEK,
+            end=self.NEXT_WEEK
+        )
+        infinite_course = CourseFactory.create(
+            org="edx",
+            mobile_available=True,
+            start=self.LAST_WEEK,
+            end=None
+        )
+        GeneratedCertificateFactory.create(
+            user=self.user,
+            course_id=infinite_course.id,
+            status=CertificateStatuses.downloadable,
+            mode='verified',
+        )
+
+        self.enroll(old_course.id)
+        self.enroll(actual_course.id)
+        self.enroll(infinite_course.id)
+
+        response = self.api_response(api_version=API_V4, data={'status': EnrollmentStatuses.ALL.value})
+        enrollments = response.data['enrollments']
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(enrollments['count'], 3)
+        self.assertEqual(enrollments['results'][0]['course']['id'], str(infinite_course.id))
+        self.assertEqual(enrollments['results'][1]['course']['id'], str(actual_course.id))
+        self.assertEqual(enrollments['results'][2]['course']['id'], str(old_course.id))
+        self.assertNotIn('primary', response.data)
+
 
 @override_settings(MKTG_URLS={'ROOT': 'dummy-root'})
 class TestUserEnrollmentCertificates(UrlResetMixin, MobileAPITestCase, MilestonesTestCaseMixin):