Merge pull request #157 from nelc/tehreem/fix_enrollment_api_issue_154

fix: issue-154 for enrollment api

futurex_openedx_extensions/dashboard/details/learners.py

@@ -12,7 +12,10 @@
 from lms.djangoapps.certificates.models import GeneratedCertificate
 from lms.djangoapps.courseware.models import StudentModule
 from lms.djangoapps.grades.models import PersistentCourseGrade
+from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
 
+from futurex_openedx_extensions.helpers.exceptions import FXCodedException, FXExceptionCodes
+from futurex_openedx_extensions.helpers.extractors import get_partial_access_course_ids, verify_course_ids
 from futurex_openedx_extensions.helpers.querysets import (
     check_staff_exist_queryset,
     get_base_queryset_courses,
@@ -263,7 +266,11 @@ def get_learner_info_queryset(
 
 
 def get_learners_enrollments_queryset(
-    user_ids: list = None, course_ids: list = None, include_staff: bool = False
+    fx_permission_info: dict,
+    user_ids: list = None,
+    course_ids: list = None,
+    search_text: str | None = None,
+    include_staff: bool = False,
 ) -> QuerySet:
     """
     Get the enrollment details. If no course_ids or user_ids are provided,
@@ -275,21 +282,33 @@ def get_learners_enrollments_queryset(
     :param include_staff: Flag to include staff users (default: False).
     :return: List of dictionaries containing user and course details.
     """
-
-    course_filter = Q(is_active=True)
-    if course_ids:
-        course_filter &= Q(course_id__in=course_ids)
+    accessible_users = get_permitted_learners_queryset(
+        queryset=get_learners_search_queryset(search_text),
+        fx_permission_info=fx_permission_info,
+        include_staff=include_staff,
+    )
     if user_ids:
-        course_filter &= Q(user_id__in=user_ids)
-
-    queryset = CourseEnrollment.objects.filter(course_filter)
+        invalid_user_ids = [user_id for user_id in user_ids if not isinstance(user_id, int)]
+        if invalid_user_ids:
+            raise FXCodedException(
+                code=FXExceptionCodes.INVALID_INPUT,
+                message=f'Invalid user ids: {invalid_user_ids}',
+            )
+        accessible_users = accessible_users.filter(id__in=user_ids)
 
-    if not include_staff:
-        queryset = queryset.filter(
-            ~check_staff_exist_queryset(ref_user_id='user_id', ref_org='course__org', ref_course_id='course_id'),
-        )
+    accessible_courses = CourseOverview.objects.filter(
+        Q(id__in=get_partial_access_course_ids(fx_permission_info)) |
+        Q(org__in=fx_permission_info['view_allowed_full_access_orgs'])
+    )
+    if course_ids:
+        verify_course_ids(course_ids)
+        accessible_courses = accessible_courses.filter(id__in=course_ids)
 
-    queryset = queryset.annotate(
+    queryset = CourseEnrollment.objects.filter(
+        is_active=True,
+        course__in=Subquery(accessible_courses.values('id')),
+        user__in=Subquery(accessible_users.values('id'))
+    ).annotate(
         certificate_available=Exists(
             GeneratedCertificate.objects.filter(
                 user_id=OuterRef('user_id'),
@@ -320,4 +339,7 @@ def get_learners_enrollments_queryset(
             output_field=BooleanField(),
         )
     ).select_related('user', 'user__profile')
+
+    update_removable_annotations(queryset, removable=['certificate_available', 'course_score', 'active_in_course'])
+
     return queryset

futurex_openedx_extensions/dashboard/serializers.py

@@ -187,9 +187,9 @@ def _get_extra_field(self: Any, obj: get_user_model, field_name: str) -> Any:
         user = self._get_user(obj)
         return getattr(user.extrainfo, field_name) if hasattr(user, 'extrainfo') and user.extrainfo else None
 
-    def get_user_id(self, obj: get_user_model) -> Any:  # pylint: disable=no-self-use
+    def get_user_id(self, obj: get_user_model) -> int:
         """Return user ID."""
-        return obj.id
+        return self._get_user(obj).id  # type: ignore
 
     def get_email(self, obj: get_user_model) -> str:
         """Return user ID."""

futurex_openedx_extensions/dashboard/views.py

@@ -463,9 +463,11 @@ def get_queryset(self, *args: Any, **kwargs: Any) -> QuerySet:
             int(user.strip()) for user in user_ids.split(',') if user.strip().isdigit()
         ] if user_ids else None
         return get_learners_enrollments_queryset(
+            fx_permission_info=self.request.fx_permission_info,
             user_ids=user_ids_list,
             course_ids=course_ids_list,
-            include_staff=self.request.query_params.get('include_staff', '0') == '1'
+            search_text=self.request.query_params.get('search_text'),
+            include_staff=self.request.query_params.get('include_staff', '0') == '1',
         )
 
     def get_serializer_context(self) -> Dict[str, Any]:
@@ -539,7 +541,7 @@ def get_queryset(self) -> QuerySet:
                     excluded_role_types=dummy_serializers.query_params['excluded_role_types'],
                 ).values('user_id').distinct().order_by()
             ).select_related('profile').order_by('id')
-        except ValueError as exc:
+        except (ValueError, FXCodedException) as exc:
             raise ParseError(f'Invalid parameter: {exc}') from exc
 
         return q_set

futurex_openedx_extensions/helpers/extractors.py

@@ -10,6 +10,7 @@
 from xmodule.modulestore.django import modulestore
 
 from futurex_openedx_extensions.helpers import constants as cs
+from futurex_openedx_extensions.helpers.exceptions import FXCodedException, FXExceptionCodes
 
 
 @dataclass
@@ -118,13 +119,22 @@ def verify_course_ids(course_ids: List[str]) -> None:
     :type course_ids: List[str]
     """
     if course_ids is None:
-        raise ValueError('course_ids must be a list of course IDs, but got None')
+        raise FXCodedException(
+            code=FXExceptionCodes.INVALID_INPUT,
+            message='course_ids must be a list of course IDs, but got None',
+        )
 
     for course_id in course_ids:
         if not isinstance(course_id, str):
-            raise ValueError(f'course_id must be a string, but got {type(course_id).__name__}')
+            raise FXCodedException(
+                code=FXExceptionCodes.INVALID_INPUT,
+                message=f'course_id must be a string, but got {type(course_id).__name__}',
+            )
         if not re.search(cs.COURSE_ID_REGX_EXACT, course_id) and not re.search(cs.LIBRARY_ID_REGX_EXACT, course_id):
-            raise ValueError(f'Invalid course ID format: {course_id}')
+            raise FXCodedException(
+                code=FXExceptionCodes.INVALID_INPUT,
+                message=f'Invalid course ID format: {course_id}',
+            )
 
 
 def get_orgs_of_courses(course_ids: List[str]) -> Dict[str, Any]:

tests/test_dashboard/test_details/test_details_learners.py

@@ -197,10 +197,16 @@ def test_get_learners_by_course_queryset_include_staff(base_data):  # pylint: di
 
 
 @pytest.mark.django_db
-def test_get_learners_enrollments_queryset_annotations(base_data):  # pylint: disable=unused-argument
+def test_get_learners_enrollments_queryset_annotations(
+    base_data, fx_permission_info
+):  # pylint: disable=unused-argument
     """Verify that get_learners_by_course_queryset returns the correct QuerySet."""
     PersistentCourseGrade.objects.create(user_id=15, course_id='course-v1:ORG1+5+5', percent_grade=0.67)
-    queryset = get_learners_enrollments_queryset(course_ids=['course-v1:ORG1+5+5'], user_ids=[15])
+    queryset = get_learners_enrollments_queryset(
+        fx_permission_info=fx_permission_info,
+        course_ids=['course-v1:ORG1+5+5'],
+        user_ids=[15]
+    )
     assert queryset.count() == 1, 'unexpected test data'
     assert queryset[0].certificate_available is not None, 'certificate_available should be in the queryset'
     assert queryset[0].course_score == 0.67, \
@@ -211,20 +217,108 @@ def test_get_learners_enrollments_queryset_annotations(base_data):  # pylint: di
     enrollment = queryset[0]
     enrollment.is_active = False
     enrollment.save()
-    assert get_learners_enrollments_queryset(course_ids=['course-v1:ORG1+5+5'], user_ids=[15]).count() == 0, \
-        'only active enrollments should be filtered'
+    assert get_learners_enrollments_queryset(
+        fx_permission_info=fx_permission_info,
+        course_ids=['course-v1:ORG1+5+5'],
+        user_ids=[15],
+    ).count() == 0, 'only active enrollments should be filtered'
 
 
 @pytest.mark.django_db
-@pytest.mark.parametrize('course_ids, user_ids, include_staff, expected_count, usecase', [
-    (['course-v1:ORG1+5+5'], None, False, 5, 'only course_ids'),
-    (['course-v1:ORG1+5+5'], None, True, 7, 'only course_ids and include_staff'),
-    (None, [15], False, 3, 'only user_ids'),
-    (['course-v1:ORG1+5+5'], [15], False, 1, 'user_ids and course_ids'),
-])
-def test_get_learners_enrollments_queryset(
-    course_ids, user_ids, include_staff, expected_count, usecase
-):
-    """Verify that get_learners_by_course_queryset returns the correct QuerySet."""
-    queryset = get_learners_enrollments_queryset(user_ids=user_ids, course_ids=course_ids, include_staff=include_staff)
-    assert queryset.count() == expected_count, f'unexpected test data with {usecase}'
+@pytest.mark.parametrize(
+    'filter_user_ids, filter_course_ids, expected_error, expected_count, usecase', [
+        (None, None, '', 6, 'No filter'),
+        ([21], None, '', 3, 'only user_ids filter'),
+        ([], ['course-v1:ORG1+5+5'], '', 2, 'only course_ids filter'),
+        ([15, 29], ['course-v1:ORG1+5+5'], '', 1, 'both course_ids and user_ids filter'),
+        ([15, 29, 'not-int', 0.2], [], 'Invalid user ids: [\'not-int\', 0.2]', 0, 'invalid user ids.'),
+        ([15, 29], ['course-v1:ORG1+5+5', 'invalid'], 'Invalid course ID format: invalid', 0, ''),
+    ]
+)
+@patch('futurex_openedx_extensions.dashboard.details.learners.get_permitted_learners_queryset')
+@patch('futurex_openedx_extensions.dashboard.details.learners.get_partial_access_course_ids')
+def test_get_learners_enrollments_queryset_for_course_ids_and_user_ids_filter(
+    mocked_partial_course_ids, mocked_permitted_learners,
+    filter_user_ids, filter_course_ids,
+    expected_error, expected_count, usecase, fx_permission_info,
+):  # pylint: disable=too-many-arguments, unused-argument
+    """Test get_learners_by_course_queryset result for course_ids and user_ids filtration."""
+    fake_accessible_user_ids = [15, 21, 29]
+    fake_accessible_course_ids = [
+        'course-v1:ORG1+5+5', 'course-v1:ORG2+4+4', 'course-v1:ORG2+5+5', 'course-v1:ORG2+6+6', 'course-v1:ORG2+7+7'
+    ]
+    mocked_permitted_learners.return_value = get_user_model().objects.filter(id__in=fake_accessible_user_ids)
+
+    if expected_error:
+        with pytest.raises(FXCodedException) as exc_info:
+            queryset = get_learners_enrollments_queryset(
+                fx_permission_info=fx_permission_info,
+                course_ids=filter_course_ids,
+                user_ids=filter_user_ids
+            )
+        assert str(exc_info.value) == expected_error
+    else:
+        queryset = get_learners_enrollments_queryset(
+            fx_permission_info=fx_permission_info,
+            course_ids=filter_course_ids,
+            user_ids=filter_user_ids
+        )
+        expected_user_ids = filter_user_ids or fake_accessible_user_ids
+        expected_course_ids = filter_course_ids or fake_accessible_course_ids
+        assert queryset.count() == expected_count, f'unexpected enrollment queryset count for case: {usecase}'
+        assert not (
+            set(queryset.values_list('user_id', flat=True)) - set(expected_user_ids)
+        ), f'unexpected enrollment user ids for case: {usecase}'
+        assert not (
+            {str(enrollment.course.id) for enrollment in queryset} - set(expected_course_ids)
+        ), f'unexpected enrollment course ids for case: {usecase}'
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    'permitted_user_ids, limited_access_course_ids, full_access_orgs, expected_count, expected_course_ids, usecase',
+    [
+        (
+            [], [], ['org1', 'org2'], 0, [], 'no user is accessible'
+        ),
+        (
+            [15, 21, 29], [], ['org1', 'org2'], 6,
+            [
+                'course-v1:ORG1+5+5', 'course-v1:ORG2+4+4', 'course-v1:ORG2+5+5',
+                'course-v1:ORG2+6+6', 'course-v1:ORG2+7+7'
+            ],
+            'full access orgs',
+        ),
+        (
+            [15, 21, 29], ['course-v1:ORG1+5+5'], [], 2,
+            ['course-v1:ORG1+5+5'],
+            'course limited access',
+        ),
+        (
+            [15, 21, 29], ['course-v1:ORG2+4+4'], ['org1'], 3,
+            ['course-v1:ORG1+5+5', 'course-v1:ORG2+4+4'],
+            'full access orgs with course limited access',
+        ),
+    ],
+)
+@patch('futurex_openedx_extensions.dashboard.details.learners.get_permitted_learners_queryset')
+@patch('futurex_openedx_extensions.dashboard.details.learners.get_partial_access_course_ids')
+def test_get_learners_enrollments_queryset_for_accessible_courses_and_users(
+    mocked_partial_course_ids, mocked_permitted_learners,
+    permitted_user_ids, limited_access_course_ids, full_access_orgs,
+    expected_count, expected_course_ids, usecase, fx_permission_info,
+):  # pylint: disable=too-many-arguments
+    """Test get_learners_by_course_queryset result for accessible users and courses."""
+    mocked_partial_course_ids.return_value = limited_access_course_ids
+    mocked_permitted_learners.return_value = get_user_model().objects.filter(id__in=permitted_user_ids)
+    fx_permission_info['view_allowed_full_access_orgs'] = full_access_orgs
+    queryset = get_learners_enrollments_queryset(
+        fx_permission_info=fx_permission_info,
+    )
+    assert queryset.count() == expected_count, f'unexpected enrollment queryset count for case: {usecase}'
+    assert not (
+        set(queryset.values_list('user_id', flat=True)) - set(permitted_user_ids)
+    ), f'unexpected enrollment user ids for case: {usecase}'
+    assert not (
+        {str(enrollment.course.id) for enrollment in queryset} - set(expected_course_ids)
+    ), f'unexpected enrollment course ids for case: {usecase}'

tests/test_dashboard/test_views.py

@@ -804,8 +804,14 @@ def test_success(self):
         self.login_user(self.staff_user)
         response = self.client.get(self.url)
         self.assertEqual(response.status_code, http_status.HTTP_200_OK)
-        response = self.client.get(self.url, {'course_ids': 'course-v1:ORG1+5+5', 'user_ids': 15})
+
+        user_id = 15
+        course_id = 'course-v1:ORG1+5+5'
+        response = self.client.get(self.url, data={'course_ids': course_id, 'user_ids': user_id})
         self.assertEqual(response.status_code, http_status.HTTP_200_OK)
+        self.assertEqual(response.data['count'], 1)
+        self.assertEqual(response.data['results'][0]['user_id'], user_id)
+        self.assertEqual(response.data['results'][0]['course_id'], course_id)
 
 
 class MockClickhouseQuery:

tests/test_helpers/test_extractors.py

@@ -5,6 +5,7 @@
 from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
 
 from futurex_openedx_extensions.helpers import constants as cs
+from futurex_openedx_extensions.helpers.exceptions import FXCodedException
 from futurex_openedx_extensions.helpers.extractors import (
     DictHashcode,
     DictHashcodeSet,
@@ -68,7 +69,7 @@ def test_verify_course_ids_success(course_ids):
 ])
 def test_verify_course_ids_fail(course_ids, error_message):
     """Verify that verify_course_ids raises an error for invalid course IDs."""
-    with pytest.raises(ValueError) as exc:
+    with pytest.raises(FXCodedException) as exc:
         verify_course_ids(course_ids)
 
     assert str(exc.value) == error_message

tests/test_helpers/test_roles.py

@@ -858,7 +858,7 @@ def test_get_roles_for_users_queryset_bad_exclude(bad_role_type, expected_error_
 
 def test_get_roles_for_users_queryset_bad_course_id():
     """Verify that get_roles_for_users_queryset raises an error if the course_id parameter is invalid."""
-    with pytest.raises(ValueError) as exc_info:
+    with pytest.raises(FXCodedException) as exc_info:
         get_course_access_roles_queryset(
             orgs_filter=['org1', 'org2'],
             remove_redundant=False,